feat: extend local GeoAI capability workflows

shuishen

10 hours ago fbb068ec702338d609c1ca6eddbdb9f182d8f211

 tests/test_serve_workbench_console.py

@@ -6,6 +6,7 @@
import tempfile
import unittest
import json
from http import HTTPStatus
from urllib.parse import quote
from pathlib import Path

@@ -96,6 +97,41 @@
            self.assertEqual(staged.read_bytes(), b"reference-bytes")
            self.assertEqual(result["sha256"], MODULE.file_sha256(staged))

    def test_binary_photo_reconstruction_upload_preserves_jpeg_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/3d-pointcloud/photo-uploads/4123456789abcdef0123456789abcdef?role=photo"
            handler.headers = {"Content-Length": "10", "X-Upload-Name": "../flight.JPG"}
            handler.rfile = io.BytesIO(b"jpeg-bytes")
            result = handler.receive_photo_reconstruction_upload("/api/3d-pointcloud/photo-uploads/4123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / result["uploadId"] / "photo.jpg"
            self.assertEqual(result["role"], "photo")
            self.assertEqual(result["name"], "flight.jpg")
            self.assertEqual(staged.read_bytes(), b"jpeg-bytes")

    def test_binary_pointcloud_upload_preserves_las_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            handler.path = "/api/3d-pointcloud/pointcloud-uploads/5123456789abcdef0123456789abcdef?role=pointcloud"
            handler.headers = {"Content-Length": "9", "X-Upload-Name": "../corridor.LAS"}
            handler.rfile = io.BytesIO(b"las-bytes")
            result = handler.receive_pointcloud_upload("/api/3d-pointcloud/pointcloud-uploads/5123456789abcdef0123456789abcdef")
            staged = Path(temp_dir) / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / result["uploadId"] / "pointcloud.las"
            self.assertEqual(result["role"], "pointcloud")
            self.assertEqual(result["name"], "corridor.las")
            self.assertEqual(staged.read_bytes(), b"las-bytes")

    def test_photo_reconstruction_request_requires_three_to_thirty_photos(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "at least three"):
            handler.create_photo_reconstruction_run({"photos": []})
        with self.assertRaisesRegex(MODULE.ApiError, "at most"):
            handler.create_photo_reconstruction_run({"photos": [{}] * (MODULE.MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN + 1)})
        with self.assertRaisesRegex(MODULE.ApiError, "true or false"):
            handler.create_photo_reconstruction_run({"photos": [{}, {}, {}], "usePositionPriors": "yes"})

    def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
@@ -158,6 +194,78 @@
        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_pointcloud_validation_run_is_discovered(self) -> None:
        runs = MODULE.pointcloud_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260821-v2")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/05-3d-pointcloud/"))

    def test_pointcloud_request_requires_allowlisted_files(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "PLY, PCD, XYZ, LAS, or LAZ"):
            handler.create_pointcloud_run({"pointClouds": []})
        with self.assertRaisesRegex(MODULE.ApiError, "Unsupported file type"):
            handler.create_pointcloud_run({"pointClouds": [{"name": "unsafe.exe", "content": "eA=="}]})

    def test_semantic_model_discovery_requires_complete_training_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            complete = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "semantic-good"
            incomplete = complete.parent / "semantic-incomplete"
            complete.mkdir(parents=True)
            incomplete.mkdir(parents=True)
            (complete / "model.pt").write_bytes(b"weights")
            (complete / "metrics.json").write_text(json.dumps({"capability": "05-3d-pointcloud", "classification": "B", "created_at": "2026-08-24", "classes": {"5": {"key": "vegetation", "label": "Vegetation", "color": [1, 2, 3]}, "16": {"key": "power_line", "label": "Power line", "color": [4, 5, 6]}}, "test": {"report": {"vegetation": {"f1-score": 0.9}}}}), encoding="utf-8")
            (incomplete / "model.pt").write_bytes(b"weights")
            models = MODULE.pointcloud_semantic_models(root)
            self.assertEqual([item["id"] for item in models], ["semantic-good"])
            self.assertEqual(models[0]["testF1"], {"vegetation": 0.9})

    def test_model_inference_rejects_unsafe_or_unknown_model_id(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "Invalid trained model id"):
            handler.create_pointcloud_model_inference_run({"modelId": "../model", "pointCloud": {}})
        with self.assertRaisesRegex(MODULE.ApiError, "unavailable or incomplete"):
            handler.create_pointcloud_model_inference_run({"modelId": "does-not-exist", "pointCloud": {}})

    def test_annotation_delete_removes_only_selected_revision(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            annotation_id = "annotation-20260822-091458-b72099"
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id
            revision.mkdir(parents=True)
            (revision / "annotation.json").write_text(json.dumps({
                "schema_version": 1,
                "id": annotation_id,
                "source_id": "source:preview.ply",
                "created_at": "2026-08-22T09:14:58+00:00",
                "labels": [[1, 15]],
            }), encoding="utf-8")
            source_sentinel = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "part_01.las"
            source_sentinel.parent.mkdir(parents=True)
            source_sentinel.write_bytes(b"source-must-remain")

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            handler.path = f"/api/3d-pointcloud/annotations/{annotation_id}"
            responses: list[tuple[HTTPStatus, dict[str, object]]] = []
            handler.send_json = lambda status, body: responses.append((status, body))

            handler.do_DELETE()

            self.assertEqual(responses, [(HTTPStatus.OK, {"deletedId": annotation_id})])
            self.assertFalse(revision.exists())
            self.assertEqual(source_sentinel.read_bytes(), b"source-must-remain")

    def test_risk_rule_validation_run_is_discovered(self) -> None:
        runs = MODULE.risk_rule_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260820")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/07-risk-rule-engine/"))

    def test_risk_rule_request_requires_fixed_three_file_contract(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "observations, zones and rules"):
            handler.create_risk_rule_run({"files": {"observations": {}}})

    def test_change_run_requires_both_allowlisted_images(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "name and Base64"):