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 geoalchemy2.shape import from_shape from shapely.geometry import box, mapping 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.model_validation_scope_service import ModelValidationScopeService 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_tiles(self, model, tile_paths, confidence_threshold: float) -> list[list[dict]]: # The service batches tiles; this double still answers per tile. return [self.predict_tile(model, tile_path, confidence_threshold) for tile_path in tile_paths] 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 # Tiles are handed to the model in batches, so ``source`` is a list. for item in source if isinstance(source, list) else [source]: with Image.open(item) as image: self.seen_sources.append( { "path": str(item), "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 = { # The runtime only consumes artifacts under the storage root, so a # test that writes tiles into tmp_path must say that is the root. "storage_root": str(tmp_path), "yolo_enabled": True, "yolo_model_path": str(model_path), "yolo_max_tiles": 4, } values.update(overrides) return Settings(**values) def _scope_settings(tmp_path: Path, scope_geometry=None, **overrides) -> Settings: model_path = tmp_path / "model.pt" model_path.write_bytes(b"scope-bound-model") payload = { "schema_version": ModelValidationScopeService.SCHEMA_VERSION, "model_id": "yolo-configured", "model_sha256": sha256(model_path.read_bytes()).hexdigest(), "scope_key": "mol-kempen-test", "crs": "EPSG:4326", "geometry": mapping(scope_geometry or box(4.0, 50.8, 5.5, 52.0)), } manifest_path = tmp_path / "model-validation-scope.json" manifest_path.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") values = { "yolo_model_path": str(model_path), "yolo_validation_scope_manifest_path": str(manifest_path), "yolo_validation_scope_manifest_sha256": sha256(manifest_path.read_bytes()).hexdigest(), } values.update(overrides) return _settings(tmp_path, **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], "crs": "EPSG:4326", "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()), "crs": "EPSG:4326", "bounds": [4.0, 51.0, 5.0, 52.0], "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="Mol validation bypass", geometry=from_shape(box(-74.1, 40.6, -73.8, 40.9), srid=4326), ) db = FakeSession(objects={(Area, dataset.area_id): wrong_area}) with pytest.raises(AppError) as exc_info: DetectionService._validate_model_area_scope(db, dataset, _scope_settings(tmp_path)) 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="Een wijzigbare weergavenaam", geometry=from_shape(box(5.0, 51.1, 5.2, 51.3), srid=4326), ) db = FakeSession(objects={(Area, dataset.area_id): area}) DetectionService._validate_model_area_scope(db, dataset, _scope_settings(tmp_path)) def test_yolo_validation_scope_rejects_tampered_manifest(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=from_shape(box(5.0, 51.1, 5.2, 51.3), srid=4326), ) settings = _scope_settings(tmp_path) Path(settings.yolo_validation_scope_manifest_path).write_text("{}", encoding="utf-8") db = FakeSession(objects={(Area, dataset.area_id): area}) with pytest.raises(AppError) as exc_info: DetectionService._validate_model_area_scope(db, dataset, settings) assert exc_info.value.code == "DETECTION_VALIDATION_SCOPE_CHECKSUM_MISMATCH" 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)