From fbb068ec702338d609c1ca6eddbdb9f182d8f211 Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Mon, 24 Aug 2026 11:37:16 +0800
Subject: [PATCH] feat: extend local GeoAI capability workflows
---
tests/test_serve_workbench_console.py | 108 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 108 insertions(+), 0 deletions(-)
diff --git a/tests/test_serve_workbench_console.py b/tests/test_serve_workbench_console.py
index bb14f49..674c884 100644
--- a/tests/test_serve_workbench_console.py
+++ b/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"):
--
Gitblit v1.9.3