from __future__ import annotations from pathlib import Path from uuid import uuid4 import pytest from fastapi.testclient import TestClient from app.main import app from app.models import AnalysisRun, Dataset, Detection, Job, Project from app.services.detection_service import DetectionService from app.services.model_registry_service import ModelRegistryService 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) def _project_and_dataset(dataset_type: str = "raster"): project_id = uuid4() dataset_id = uuid4() project = Project(id=project_id, name="Geel") dataset = Dataset( id=dataset_id, project_id=project_id, name="source.tif", dataset_type=dataset_type, source="test", storage_path="storage/uploads/source.tif", ) db = FakeSession(objects={(Project, project_id): project, (Dataset, dataset_id): dataset}) return db, project_id, dataset_id def test_model_registry_returns_detection_placeholders() -> None: models = {model.model_id: model for model in ModelRegistryService.list_model_capabilities()} assert set(models) == {"yolo-placeholder", "yolo-configured", "manual-fixture-detector"} assert models["yolo-placeholder"].task_type == "object_detection" assert models["yolo-placeholder"].configured is False assert models["yolo-placeholder"].status == "not_configured" assert models["yolo-configured"].configured is False assert models["yolo-configured"].status == "not_configured" assert models["manual-fixture-detector"].configured is True assert "fixture" in models["manual-fixture-detector"].limitation_message.lower() def test_unavailable_detector_creates_failed_run_and_job_without_detections() -> None: db, project_id, dataset_id = _project_and_dataset() result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-placeholder", confidence_threshold=0.5, class_filter=["building"], parameters_json={}, ) runs = [item for item in db.added if isinstance(item, AnalysisRun)] jobs = [item for item in db.added if isinstance(item, Job)] detections = [item for item in db.added if isinstance(item, Detection)] assert result.status == "failed" assert result.error_code == "DETECTION_MODEL_UNAVAILABLE" assert result.detection_count == 0 assert runs[0].analysis_type == "detection" assert runs[0].status == "failed" assert jobs[0].job_type == "detection.run" assert jobs[0].status == "failed" assert detections == [] def test_non_raster_dataset_request_is_rejected() -> None: db, project_id, dataset_id = _project_and_dataset(dataset_type="vector") with pytest.raises(Exception) as exc_info: DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-placeholder", confidence_threshold=0.5, ) assert getattr(exc_info.value, "code", None) == "INVALID_DATASET_TYPE" def test_historical_raster_marked_unsupported_is_rejected_before_run_creation() -> None: db, project_id, dataset_id = _project_and_dataset() source = db.get(Dataset, dataset_id) source.source_name = "digitaal_vlaanderen_orthophoto" source.source_metadata = {"product_key": "2020", "supports_detection": False} with pytest.raises(Exception) as exc_info: DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="yolo-placeholder", confidence_threshold=0.5, ) assert getattr(exc_info.value, "code", None) == "DETECTION_SOURCE_TEMPORALLY_UNSUPPORTED" assert db.added == [] def test_fixture_detector_persists_detections_only_with_explicit_fixture_mode() -> None: db, project_id, dataset_id = _project_and_dataset() with pytest.raises(Exception) as exc_info: DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="manual-fixture-detector", confidence_threshold=0.5, parameters_json={"fixture_detections": []}, ) assert getattr(exc_info.value, "code", None) == "FIXTURE_MODE_REQUIRED" result = DetectionService.run_detection( db=db, project_id=project_id, dataset_id=dataset_id, model_id="manual-fixture-detector", confidence_threshold=0.5, class_filter=["building"], parameters_json={ "fixture_mode": True, "fixture_detections": [ { "class_name": "building", "confidence": 0.92, "bbox_json": {"x_min": 1, "y_min": 2, "x_max": 10, "y_max": 12}, "geometry": { "type": "Polygon", "coordinates": [ [ [4.0, 51.0], [4.1, 51.0], [4.1, 51.1], [4.0, 51.1], [4.0, 51.0], ] ], }, "properties_json": {"source": "unit-test-fixture"}, } ], }, ) detections = [item for item in db.added if isinstance(item, Detection)] assert result.status == "success" assert result.detection_count == 1 assert detections[0].project_id == project_id assert detections[0].dataset_id == dataset_id assert detections[0].analysis_run_id == result.analysis_run_id assert detections[0].model_name == "manual-fixture-detector" assert detections[0].class_name == "building" assert detections[0].confidence == 0.92 assert detections[0].bbox_json == {"x_min": 1, "y_min": 2, "x_max": 10, "y_max": 12} def test_detection_models_api_uses_envelope() -> None: response = TestClient(app).get("/api/v1/detection/models") assert response.status_code == 200 assert "data" in response.json() assert {model["model_id"] for model in response.json()["data"]["models"]} == { "yolo-placeholder", "yolo-configured", "manual-fixture-detector", } def test_sprint8_migration_declares_detection_foundation() -> None: migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "202606120800_sprint8_detection_foundation.py" migration_text = migration_path.read_text(encoding="utf-8") for required_text in ( "detections", "analysis_runs", "dataset_id", "job_id", "model_name", "model_version", "result_json", "ix_detections_project_id", "ix_detections_dataset_id", "ix_detections_analysis_run_id", "ix_detections_class_name", "ix_detections_geometry", 'postgresql_using="gist"', ): assert required_text in migration_text