from __future__ import annotations
|
|
import importlib.util
|
import io
|
import sys
|
import tempfile
|
import unittest
|
import json
|
from urllib.parse import quote
|
from pathlib import Path
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
SCRIPT = ROOT / "scripts" / "serve_workbench_console.py"
|
SPEC = importlib.util.spec_from_file_location("serve_workbench_console", SCRIPT)
|
assert SPEC and SPEC.loader
|
MODULE = importlib.util.module_from_spec(SPEC)
|
SPEC.loader.exec_module(MODULE)
|
|
|
class WorkbenchConsoleHandlerTests(unittest.TestCase):
|
def make_handler(self) -> MODULE.WorkbenchConsoleHandler:
|
handler = object.__new__(MODULE.WorkbenchConsoleHandler)
|
handler.directory = str(ROOT)
|
return handler
|
|
def test_allows_only_console_artifacts_and_detection_originals(self) -> None:
|
handler = self.make_handler()
|
self.assertEqual(
|
Path(handler.translate_path("/apps/workbench-console/index.html")),
|
ROOT / "apps" / "workbench-console" / "dist" / "index.html",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/outputs/15-trajectory-analysis/run_metadata.json")),
|
ROOT / "shared" / "outputs" / "15-trajectory-analysis" / "run_metadata.json",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/data/raw/01-object-detection/sample.jpeg")),
|
ROOT / "shared" / "data" / "raw" / "01-object-detection" / "sample.jpeg",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/data/raw/02-semantic-mapping/sample.tif")),
|
ROOT / "shared" / "data" / "raw" / "02-semantic-mapping" / "sample.tif",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/data/raw/00-change-detection/sample.jpg")),
|
ROOT / "shared" / "data" / "raw" / "00-change-detection" / "sample.jpg",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/data/raw/09-anomaly-detection/sample.tif")),
|
ROOT / "shared" / "data" / "raw" / "09-anomaly-detection" / "sample.tif",
|
)
|
self.assertEqual(
|
Path(handler.translate_path("/shared/data/processed/09-anomaly-detection/sample.tif")),
|
ROOT / ".console-forbidden",
|
)
|
|
def test_upload_name_is_sanitized_and_extension_is_allowlisted(self) -> None:
|
self.assertEqual(MODULE.safe_file_name("../../unsafe name.JPG", {".jpg"}), "unsafe_name.jpg")
|
self.assertEqual(MODULE.safe_file_name("../正常参考图.JPG", {".jpg"}), "正常参考图.jpg")
|
with self.assertRaises(MODULE.ApiError):
|
MODULE.safe_file_name("image.exe", {".jpg", ".png"})
|
|
def test_oversized_request_is_rejected_before_reading_body(self) -> None:
|
handler = self.make_handler()
|
handler.headers = {"Content-Length": str(MODULE.MAX_REQUEST_BYTES + 1), "Content-Type": "application/json"}
|
handler.rfile = io.BytesIO(b"")
|
with self.assertRaises(MODULE.ApiError):
|
handler.read_json_body()
|
|
def test_large_tiff_upload_limits_allow_one_gibibyte_files(self) -> None:
|
self.assertEqual(MODULE.MAX_FILE_BYTES, 1024 * 1024 * 1024)
|
self.assertEqual(MODULE.MAX_REQUEST_BYTES, 3072 * 1024 * 1024)
|
|
def test_binary_change_upload_preserves_original_bytes(self) -> None:
|
with tempfile.TemporaryDirectory() as temp_dir:
|
handler = object.__new__(MODULE.WorkbenchConsoleHandler)
|
handler.directory = temp_dir
|
handler.path = "/api/change-detection/uploads/0123456789abcdef0123456789abcdef?role=before"
|
handler.headers = {"Content-Length": "13", "X-Upload-Name": "../1.tif"}
|
handler.rfile = io.BytesIO(b"raw-tif-bytes")
|
result = handler.receive_change_upload("/api/change-detection/uploads/0123456789abcdef0123456789abcdef")
|
staged = Path(temp_dir) / "shared" / "data" / "raw" / "00-change-detection" / "uploads" / result["uploadId"] / "before.tif"
|
self.assertEqual(staged.read_bytes(), b"raw-tif-bytes")
|
self.assertEqual(result["name"], "1.tif")
|
|
def test_binary_anomaly_upload_preserves_bytes_and_checksum(self) -> None:
|
with tempfile.TemporaryDirectory() as temp_dir:
|
handler = object.__new__(MODULE.WorkbenchConsoleHandler)
|
handler.directory = temp_dir
|
handler.path = "/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef?role=reference"
|
handler.headers = {"Content-Length": "15", "X-Upload-Name": "../normal.JPG"}
|
handler.rfile = io.BytesIO(b"reference-bytes")
|
result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/1123456789abcdef0123456789abcdef")
|
staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "reference.jpg"
|
self.assertEqual(staged.read_bytes(), b"reference-bytes")
|
self.assertEqual(result["sha256"], MODULE.file_sha256(staged))
|
|
def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
|
with tempfile.TemporaryDirectory() as temp_dir:
|
handler = object.__new__(MODULE.WorkbenchConsoleHandler)
|
handler.directory = temp_dir
|
handler.path = "/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef?role=input"
|
handler.headers = {"Content-Length": "11", "X-Upload-Name": quote("异常测试图.JPG", safe="")}
|
handler.rfile = io.BytesIO(b"image-bytes")
|
result = handler.receive_anomaly_upload("/api/anomaly-detection/uploads/3123456789abcdef0123456789abcdef")
|
staged = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / result["uploadId"] / "input.jpg"
|
self.assertEqual(result["name"], "异常测试图.jpg")
|
self.assertEqual(staged.read_bytes(), b"image-bytes")
|
|
def test_interrupted_anomaly_upload_removes_partial_file(self) -> None:
|
with tempfile.TemporaryDirectory() as temp_dir:
|
handler = object.__new__(MODULE.WorkbenchConsoleHandler)
|
handler.directory = temp_dir
|
handler.path = "/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef?role=input"
|
handler.headers = {"Content-Length": "20", "X-Upload-Name": "target.tif"}
|
handler.rfile = io.BytesIO(b"short")
|
with self.assertRaisesRegex(MODULE.ApiError, "ended before"):
|
handler.receive_anomaly_upload("/api/anomaly-detection/uploads/2123456789abcdef0123456789abcdef")
|
staging = Path(temp_dir) / "shared" / "data" / "raw" / "09-anomaly-detection" / "uploads" / "2123456789abcdef0123456789abcdef"
|
self.assertFalse(any(staging.glob("*.part")))
|
|
def test_anomaly_parameters_are_bounded(self) -> None:
|
self.assertEqual(MODULE.validate_anomaly_parameters({}), (256, 128, 0.995, 42))
|
with self.assertRaisesRegex(MODULE.ApiError, "Stride"):
|
MODULE.validate_anomaly_parameters({"tileSize": 128, "stride": 256})
|
with self.assertRaisesRegex(MODULE.ApiError, "quantile"):
|
MODULE.validate_anomaly_parameters({"thresholdQuantile": 1.0})
|
|
def test_anomaly_run_discovery_requires_exposed_raw_sources(self) -> None:
|
with tempfile.TemporaryDirectory() as temp_dir:
|
root = Path(temp_dir)
|
artifact = root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / "anomaly-test"
|
raw_input = root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / "anomaly-test" / "input"
|
raw_reference = raw_input.parent / "reference"
|
artifact.mkdir(parents=True)
|
raw_input.mkdir(parents=True)
|
raw_reference.mkdir(parents=True)
|
(artifact / "target.comparison.overlay.png").write_bytes(b"png")
|
metadata = {"capability": "09-anomaly-detection", "created_at": "2026-08-18", "display_name": "五参考图边界案例", "case_note": "保留真实漏检结果。", "raw_input_dir": raw_input.relative_to(root).as_posix(), "raw_reference_dir": raw_reference.relative_to(root).as_posix(), "images": [{"overlay_file": "target.comparison.overlay.png"}]}
|
(artifact / "run_metadata.json").write_text(json.dumps(metadata), encoding="utf-8")
|
runs = MODULE.anomaly_runs(root)
|
self.assertEqual(runs[0]["id"], "anomaly-test")
|
self.assertEqual(runs[0]["label"], "五参考图边界案例")
|
self.assertEqual(runs[0]["note"], "保留真实漏检结果。")
|
|
def test_script_failure_becomes_a_useful_api_error(self) -> None:
|
handler = self.make_handler()
|
with self.assertRaisesRegex(MODULE.ApiError, "Processing failed"):
|
handler.run_command([sys.executable, "-c", "raise SystemExit(2)"], timeout=10)
|
|
def test_semantic_validation_run_is_discovered(self) -> None:
|
runs = MODULE.semantic_runs(ROOT)
|
self.assertTrue(any(item["id"] == "validation-20260817" for item in runs))
|
|
def test_spatial_measurement_run_is_discovered(self) -> None:
|
runs = MODULE.measurement_runs(ROOT)
|
validation = next(item for item in runs if item["id"] == "validation-normal-20260817-v4")
|
self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/04-spatial-measurement/"))
|
|
def test_change_run_requires_both_allowlisted_images(self) -> None:
|
handler = self.make_handler()
|
with self.assertRaisesRegex(MODULE.ApiError, "name and Base64"):
|
handler.create_change_run({"files": {"before": {"name": "before.jpg", "content": "eA=="}}})
|
|
def test_change_threshold_is_validated(self) -> None:
|
handler = self.make_handler()
|
payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
|
with self.assertRaisesRegex(MODULE.ApiError, "between 0.01 and 0.99"):
|
handler.create_change_run({**payload, "threshold": 1.0})
|
with self.assertRaisesRegex(MODULE.ApiError, "must be a number"):
|
handler.create_change_run({**payload, "threshold": "0.5"})
|
|
def test_change_resolution_is_validated(self) -> None:
|
handler = self.make_handler()
|
payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
|
with self.assertRaisesRegex(MODULE.ApiError, "between 512 and 4096"):
|
handler.create_change_run({**payload, "maxDimension": 256})
|
with self.assertRaisesRegex(MODULE.ApiError, "must be an integer"):
|
handler.create_change_run({**payload, "maxDimension": "2048"})
|
|
def test_change_processing_mode_is_validated(self) -> None:
|
handler = self.make_handler()
|
payload = {"files": {"before": {"name": "before.jpg", "content": "eA=="}, "after": {"name": "after.jpg", "content": "eA=="}}}
|
with self.assertRaisesRegex(MODULE.ApiError, "auto, image, or geotiff"):
|
handler.create_change_run({**payload, "processingMode": "wrong"})
|
|
def test_change_validation_run_is_discovered_without_fake_crs(self) -> None:
|
runs = MODULE.change_runs(ROOT)
|
self.assertTrue(any(item["id"].startswith("validation-real-") for item in runs))
|
|
def test_semantic_task_catalog_only_enables_verified_baseline(self) -> None:
|
tasks = MODULE.semantic_tasks(ROOT)
|
self.assertEqual(len(tasks), 5)
|
selectable = [item["id"] for item in tasks if item.get("selectable") is True]
|
self.assertEqual(selectable, ["color_baseline"])
|
handler = self.make_handler()
|
with self.assertRaisesRegex(MODULE.ApiError, "not runnable yet"):
|
handler.create_semantic_run({"taskId": "drainage_blockage", "images": []})
|
|
def test_blocks_repository_files_and_encoded_traversal(self) -> None:
|
handler = self.make_handler()
|
forbidden = ROOT / ".console-forbidden"
|
self.assertEqual(Path(handler.translate_path("/.git/HEAD")), forbidden)
|
self.assertEqual(Path(handler.translate_path("/%2e%2e/.env")), forbidden)
|
self.assertEqual(Path(handler.translate_path("/PROJECT_CONTEXT.md")), forbidden)
|
|
|
if __name__ == "__main__":
|
unittest.main()
|
