from __future__ import annotations import importlib.util import io import sys import tempfile import unittest 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", ) 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") 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_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()