geointel/backend/tests/test_sprint8_detection_foundation.py

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