from __future__ import annotations from hashlib import sha256 import json from pathlib import Path import sys from types import SimpleNamespace from uuid import uuid4 import pytest from app.core.config import Settings from app.core.errors import AppError from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project, SourceRegistry, SourceSnapshot from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon from app.services.detection_service import DetectionService from app.services.model_registry_service import ModelRegistryService from app.services.runtime_model_provenance_service import RuntimeModelProvenanceService from app.services.yolo_adapter import YoloDetectionAdapter ROOT = Path(__file__).resolve().parents[2] class FakeSession: def __init__(self, objects=None) -> None: self.objects = objects or {} self.added = [] self.commits = 0 self.refreshes = [] def get(self, model, item_id): return self.objects.get((model, item_id)) def add(self, item) -> None: self.added.append(item) if getattr(item, "id", None) is not None: self.objects[(item.__class__, item.id)] = item def commit(self) -> None: self.commits += 1 def refresh(self, item) -> None: self.refreshes.append(item) class AvailableAdapter: @staticmethod def dependencies_available() -> bool: return True class MissingDependencyAdapter: @staticmethod def dependencies_available() -> bool: return False class MockYoloAdapter: def __init__(self, settings: Settings) -> None: self.settings = settings self.loaded_model_path: Path | None = None @staticmethod def dependencies_available() -> bool: return True def load_model(self, model_path: Path): self.loaded_model_path = model_path return object() def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: assert tile_path.name == "tile_0000.tif" assert confidence_threshold == 0.5 return [ { "class_name": "building", "confidence": 0.91, "bbox": [10.0, 20.0, 30.0, 40.0], "properties": {"adapter": "mock"}, } ] class NeverLoadUnboundModelAdapter(MockYoloAdapter): load_calls = 0 def load_model(self, model_path: Path): type(self).load_calls += 1 raise AssertionError("unbound model provenance must be rejected before adapter.load_model") class MixedCaseYoloAdapter(MockYoloAdapter): def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: return [ { "class_name": "Building", "confidence": 0.91, "bbox": [10.0, 20.0, 30.0, 40.0], "properties": {"adapter": "mock"}, } ] class OverlappingTileYoloAdapter(MockYoloAdapter): def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict]: tile_index = int(tile_path.stem.split("_")[-1]) if tile_index == 0: bbox = [10.0, 20.0, 30.0, 40.0] confidence = 0.82 else: bbox = [11.0, 21.0, 31.0, 41.0] confidence = 0.91 return [ { "class_name": "building", "confidence": confidence, "bbox": bbox, "properties": {"adapter": "overlap"}, } ] class RecordingPredictModel: def __init__(self) -> None: self.seen_sources: list[dict] = [] def predict(self, *, source, conf, imgsz, device, verbose, max_det): from PIL import Image with Image.open(source) as image: self.seen_sources.append( { "path": str(source), "mode": image.mode, "bands": len(image.getbands()), "conf": conf, "imgsz": imgsz, "device": device, "verbose": verbose, "max_det": max_det, } ) return [] class ExplodingPredictModel: def predict(self, *, source, conf, imgsz, device, verbose): raise RuntimeError("expected input[1, 1, 480, 640] to have 3 channels, but got 1 channels instead") def _project_and_dataset(dataset_type: str = "raster"): project_id = uuid4() dataset_id = uuid4() source_registry_id = uuid4() source_snapshot_id = uuid4() checksum = "a" * 64 project = Project(id=project_id, name="Geel") source_registry = SourceRegistry( id=source_registry_id, source_key="test-derived-raster", display_name="Governed test-derived raster", classification="derived", authority_name="GeoIntel test fixture", usage_policy_json={"ground_truth_allowed": False}, ) source_snapshot = SourceSnapshot( id=source_snapshot_id, source_registry_id=source_registry_id, snapshot_key="test-derived-raster-v1", checksum_sha256=checksum, freshness_status="current", ingest_status="ingested", ) dataset = Dataset( id=dataset_id, project_id=project_id, name="source.tif", dataset_type=dataset_type, source="test-derived-raster", source_name="test-derived-raster", storage_path="storage/uploads/source.tif", checksum_sha256=checksum, source_registry_id=source_registry_id, source_snapshot_id=source_snapshot_id, data_contract_key="geointel.raster.geotiff", data_contract_version="1.0.0", validation_status="passed", provenance_status="complete", lineage_status="not_applicable", quarantine_status="not_quarantined", status="ready", ) dataset.source_registry = source_registry dataset.source_snapshot = source_snapshot db = FakeSession(objects={(Project, project_id): project, (Dataset, dataset_id): dataset}) return db, project_id, dataset_id def _settings(tmp_path: Path, **overrides) -> Settings: model_path = tmp_path / "model.pt" values = { "yolo_enabled": True, "yolo_model_path": str(model_path), "yolo_max_tiles": 4, } values.update(overrides) return Settings(**values) def _write_model_sidecar( model_path: Path, settings: Settings, *, db: FakeSession | None = None, ) -> None: """Create explicit test-only evidence; production never self-generates it.""" model_sha256 = sha256(model_path.read_bytes()).hexdigest() source_registry_id = uuid4() source_snapshot_id = uuid4() source_version = settings.yolo_model_version or "test-v1" if db is not None: source_registry = SourceRegistry( id=source_registry_id, source_key="model", display_name="Governed test model artifact", classification="experimental", authority_name="GeoIntel test fixture", freshness_status="current", ingest_status="configured", ) source_snapshot = SourceSnapshot( id=source_snapshot_id, source_registry_id=source_registry_id, snapshot_key=f"model-{source_version}", source_version=source_version, checksum_sha256=model_sha256, freshness_status="current", ingest_status="ingested", ) db.objects[(SourceRegistry, source_registry_id)] = source_registry db.objects[(SourceSnapshot, source_snapshot_id)] = source_snapshot payload = { "schema_version": RuntimeModelProvenanceService.MANIFEST_SCHEMA_VERSION, "data_contract": {"key": "geointel.model.pytorch", "version": "1.0.0"}, "model": { "model_id": settings.yolo_model_id, "task_type": "object_detection", "sha256": model_sha256, "model_format": "pytorch", "framework": "ultralytics/pytorch", "class_mapping": {"0": "building"}, "source_version": source_version, }, "source": { "source_registry_id": str(source_registry_id), "source_snapshot_id": str(source_snapshot_id), "source_registry_key": "model", "source_snapshot_checksum_sha256": model_sha256, }, "lineage": { "upstream_asset_ids": ["test-training-corpus"], "upstream_checksums_sha256": ["a" * 64], "transformations": [ {"name": "test-training", "version": "1.0.0", "checksum_sha256": "b" * 64} ], }, "metadata": {"training_manifest_sha256": "c" * 64}, "imported_at": "2026-08-01T10:00:00+00:00", } payload["metadata"]["runtime_manifest_sha256"] = RuntimeModelProvenanceService.manifest_self_checksum(payload) RuntimeModelProvenanceService.manifest_path_for_model(model_path).write_text( json.dumps(payload, sort_keys=True), encoding="utf-8", ) def _manifest(tmp_path: Path, tile_count: int = 1) -> Path: tiles = [] for index in range(tile_count): tile_path = tmp_path / f"tile_{index:04d}.tif" tile_path.write_bytes(b"fixture") tiles.append( { "path": str(tile_path), "pixel_window": [0, 0, 100, 100], "bounds": [4.0, 51.0, 5.0, 52.0], "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], "index": index, } ) manifest_path = tmp_path / "manifest.json" manifest_path.write_text( json.dumps( { "tile_set_id": "tiles-fixture", "source_dataset_id": str(uuid4()), "source_raster_id": str(uuid4()), "tile_size": 100, "overlap": 0, "count": tile_count, "tiles": tiles, } ), encoding="utf-8", ) return manifest_path def test_yolo_configured_model_reports_not_configured_when_disabled(tmp_path: Path) -> None: settings = _settings(tmp_path, yolo_enabled=False) models = {model.model_id: model for model in ModelRegistryService.list_model_capabilities(settings=settings)} assert "yolo-configured" in models assert models["yolo-configured"].configured is False assert models["yolo-configured"].status == "not_configured" def test_yolo_configured_model_reports_dependency_unavailable(tmp_path: Path) -> None: model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) model = ModelRegistryService.get_model_capability( "yolo-configured", settings=settings, yolo_adapter_class=MissingDependencyAdapter, ) assert model is not None assert model.configured is False assert model.status == "dependency_unavailable" def test_yolo_configured_model_requires_a_runtime_provenance_sidecar(tmp_path: Path) -> None: model_path = tmp_path / "model.pt" model_path.write_bytes(b"unmanifested local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) model = ModelRegistryService.get_model_capability( "yolo-configured", settings=settings, yolo_adapter_class=AvailableAdapter, ) assert model is not None assert model.configured is False assert model.status == "contract_incomplete" assert "sidecar" in model.limitation_message def test_yolo_configured_model_reports_configured_with_local_model_and_dependencies(tmp_path: Path) -> None: model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) _write_model_sidecar(model_path, settings) model = ModelRegistryService.get_model_capability("yolo-configured", settings=settings, yolo_adapter_class=AvailableAdapter) assert model is not None assert model.configured is True assert model.status == "configured" assert model.version == settings.yolo_model_version assert model.nationally_validated is False assert model.operator_review_required is True assert model.validated_regions == ["flanders_mol_kempen"] assert model.supported_classes == ["building"] assert "Mol and the Kempen" in (model.validation_scope or "") def test_yolo_dependency_check_uses_real_imports_not_find_spec() -> None: source = (ROOT / "backend" / "app" / "services" / "yolo_adapter.py").read_text(encoding="utf-8") assert 'find_spec("ultralytics")' not in source assert "import ultralytics" in source assert "import torch" in source def test_yolo_runtime_fails_closed_when_cuda_is_required_but_unavailable(tmp_path: Path, monkeypatch) -> None: settings = _settings(tmp_path, yolo_device="cuda:0", yolo_require_cuda=True) monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False))) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).validate_runtime() assert exc_info.value.code == "DETECTION_ACCELERATOR_UNAVAILABLE" def test_yolo_runtime_rejects_cpu_device_when_cuda_is_required(tmp_path: Path, monkeypatch) -> None: settings = _settings(tmp_path, yolo_device="cpu", yolo_require_cuda=True) monkeypatch.setitem(sys.modules, "torch", SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: True))) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).validate_runtime() assert exc_info.value.code == "DETECTION_ACCELERATOR_MISCONFIGURED" def test_yolo_validation_scope_requires_persisted_validated_area(tmp_path: Path) -> None: dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) wrong_area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Brussels", geometry="MULTIPOLYGON EMPTY") db = FakeSession(objects={(Area, dataset.area_id): wrong_area}) with pytest.raises(AppError) as exc_info: DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) assert exc_info.value.code == "DETECTION_VALIDATION_SCOPE_UNAVAILABLE" def test_yolo_validation_scope_accepts_bound_mol_area(tmp_path: Path) -> None: dataset = Dataset(id=uuid4(), project_id=uuid4(), name="image.tif", dataset_type="raster", source="test", area_id=uuid4()) area = Area(id=dataset.area_id, project_id=dataset.project_id, name="Gemeente Mol", geometry="MULTIPOLYGON EMPTY") db = FakeSession(objects={(Area, dataset.area_id): area}) DetectionService._validate_model_area_scope(db, dataset, _settings(tmp_path, yolo_validated_area_names="Mol,Kempen")) def test_yolo_run_requires_tile_manifest_path(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() settings = _settings(tmp_path) with pytest.raises(Exception) as exc_info: DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, settings=settings, yolo_adapter_class=AvailableAdapter, ) assert getattr(exc_info.value, "code", None) == "DETECTION_TILE_MANIFEST_REQUIRED" def test_yolo_run_fails_closed_before_adapter_load_without_sidecar(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"unmanifested local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) # AvailableAdapter intentionally has no load_model method. If runtime # provenance were checked after adapter loading, this would raise instead # of returning the explicit unavailable capability state. result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(_manifest(tmp_path)), settings=settings, yolo_adapter_class=AvailableAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_MODEL_UNAVAILABLE" assert "sidecar" in result.message def test_yolo_run_rejects_manifest_over_tile_limit(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_max_tiles=1) _write_model_sidecar(model_path, settings, db=db) manifest_path = _manifest(tmp_path, tile_count=2) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_LIMIT_EXCEEDED" def test_yolo_run_rejects_missing_tile_manifest_file(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) _write_model_sidecar(model_path, settings, db=db) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(tmp_path / "missing-manifest.json"), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_MANIFEST_NOT_FOUND" def test_yolo_run_rejects_invalid_tile_manifest_json(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) _write_model_sidecar(model_path, settings, db=db) manifest_path = tmp_path / "manifest.json" manifest_path.write_text("{not-json", encoding="utf-8") result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) assert result.status == "failed" assert result.error_code == "DETECTION_TILE_MANIFEST_INVALID" def test_yolo_run_rejects_unbound_model_snapshot_before_adapter_load(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"structurally valid but unbound model") settings = _settings(tmp_path, yolo_model_path=str(model_path)) # A catalog/preflight sidecar alone is deliberately insufficient for a # production call. Do not register the declared source IDs in ``db``. _write_model_sidecar(model_path, settings) NeverLoadUnboundModelAdapter.load_calls = 0 result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, tile_manifest_path=str(_manifest(tmp_path)), settings=settings, yolo_adapter_class=NeverLoadUnboundModelAdapter, ) assert result.status == "failed" assert result.error_code == "MODEL_PROVENANCE_SOURCE_REGISTRY_NOT_FOUND" assert NeverLoadUnboundModelAdapter.load_calls == 0 def test_pixel_bbox_to_epsg4326_polygon_from_gdal_transform() -> None: polygon = pixel_bbox_to_epsg4326_polygon( bbox=[10, 20, 30, 40], tile={ "transform": [4.0, 0.01, 0.0, 52.0, 0.0, -0.01], "bounds": [4.0, 51.0, 5.0, 52.0], }, crs="EPSG:4326", ) assert polygon.bounds == pytest.approx((4.1, 51.6, 4.3, 51.8)) def test_yolo_run_persists_mocked_georeferenced_detections(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_model_version="local-test") _write_model_sidecar(model_path, settings, db=db) manifest_path = _manifest(tmp_path, tile_count=1) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MockYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] runs = [item for item in db.added if isinstance(item, AnalysisRun)] jobs = [item for item in db.added if isinstance(item, Job)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].model_name == "yolo-configured" assert detections[0].model_version == "local-test" assert detections[0].class_name == "building" assert detections[0].confidence == 0.91 assert detections[0].source_tile_path.endswith("tile_0000.tif") assert detections[0].bbox_json == {"x_min": 10.0, "y_min": 20.0, "x_max": 30.0, "y_max": 40.0} assert detections[0].properties_json["adapter"] == "mock" assert detections[0].properties_json["tile_index"] == 0 assert detections[0].properties_json["runtime_model_provenance"]["model_sha256"] == sha256(model_path.read_bytes()).hexdigest() assert runs[0].parameters_json["runtime_model_provenance"]["data_contract_key"] == "geointel.model.pytorch" assert runs[0].status == "success" assert jobs[0].status == "success" def test_yolo_class_filter_is_case_insensitive_and_persists_canonical_class(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path)) _write_model_sidecar(model_path, settings, db=db) manifest_path = _manifest(tmp_path, tile_count=1) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=MixedCaseYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].class_name == "building" assert detections[0].properties_json["model_class_name"] == "Building" def test_yolo_run_suppresses_cross_tile_duplicate_detections(tmp_path: Path) -> None: db, project_id, dataset_id = _project_and_dataset() model_path = tmp_path / "model.pt" model_path.write_bytes(b"local weights") settings = _settings(tmp_path, yolo_model_path=str(model_path), yolo_duplicate_iou_threshold=0.5) _write_model_sidecar(model_path, settings, db=db) manifest_path = _manifest(tmp_path, tile_count=2) result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-configured", confidence_threshold=0.5, class_filter=["building"], tile_manifest_path=str(manifest_path), settings=settings, yolo_adapter_class=OverlappingTileYoloAdapter, ) detections = [item for item in db.added if isinstance(item, Detection)] runs = [item for item in db.added if isinstance(item, AnalysisRun)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].confidence == 0.91 assert detections[0].source_tile_path.endswith("tile_0001.tif") assert runs[0].result_json["raw_detection_count"] == 2 assert runs[0].result_json["suppressed_detection_count"] == 1 assert runs[0].result_json["duplicate_iou_threshold"] == 0.5 def test_yolo_adapter_converts_single_band_tiles_to_rgb_before_prediction(tmp_path: Path) -> None: Image = pytest.importorskip("PIL.Image") tile_path = tmp_path / "single_band_tile.tif" Image.new("L", (16, 16), 128).save(tile_path) model = RecordingPredictModel() settings = _settings(tmp_path, yolo_image_size=64, yolo_device="cpu") detections = YoloDetectionAdapter(settings).predict_tile(model, tile_path, confidence_threshold=0.25) assert detections == [] assert model.seen_sources[0]["mode"] == "RGB" assert model.seen_sources[0]["bands"] == 3 assert model.seen_sources[0]["path"] != str(tile_path) assert model.seen_sources[0]["conf"] == 0.25 assert model.seen_sources[0]["imgsz"] == 64 assert model.seen_sources[0]["device"] == "cpu" assert model.seen_sources[0]["verbose"] is False assert model.seen_sources[0]["max_det"] == 1000 def test_yolo_adapter_uses_configured_max_detections(tmp_path: Path) -> None: Image = pytest.importorskip("PIL.Image") tile_path = tmp_path / "rgb_tile.png" Image.new("RGB", (16, 16), (10, 20, 30)).save(tile_path) model = RecordingPredictModel() settings = _settings(tmp_path, yolo_max_detections=1500) detections = YoloDetectionAdapter(settings).predict_tile(model, tile_path, confidence_threshold=0.25) assert detections == [] assert model.seen_sources[0]["max_det"] == 1500 def test_yolo_adapter_wraps_prediction_runtime_errors(tmp_path: Path) -> None: tile_path = tmp_path / "tile.tif" tile_path.write_bytes(b"not an image but present") settings = _settings(tmp_path) with pytest.raises(AppError) as exc_info: YoloDetectionAdapter(settings).predict_tile(ExplodingPredictModel(), tile_path, confidence_threshold=0.25) assert exc_info.value.code == "DETECTION_INFERENCE_FAILED" assert "Configured YOLO inference failed for a raster tile" in exc_info.value.message assert exc_info.value.details["tile_path"] == str(tile_path)