218 lines
7.5 KiB
Python
218 lines
7.5 KiB
Python
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="user_upload",
|
|
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
|