Complete temporal detection safety gate
GeoIntel CI / docs-smoke (push) Canceled after 0s
GeoIntel CI / contract-smoke (push) Canceled after 0s

This commit is contained in:
Codex
2026-07-18 00:53:00 +02:00
parent fc42ea9af5
commit 9146981802
16 changed files with 689 additions and 23 deletions
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
from datetime import UTC, datetime
from uuid import uuid4
import pytest
@@ -11,7 +12,7 @@ from shapely.geometry import Polygon, box
from app.core.errors import AppError
from app.main import app
from app.db.session import get_db
from app.models import AnalysisRun, Dataset, Detection, Job, Metric, Project, QualityCheck, VectorFeature
from app.models import AnalysisRun, Dataset, Detection, Metric, QualityCheck, VectorFeature
from app.services.detection_service import DetectionService
@@ -89,6 +90,17 @@ def _detection(project_id, dataset_id, analysis_run_id, class_name="building", c
)
def _source_dataset(project_id, dataset_id):
return Dataset(
id=dataset_id,
project_id=project_id,
name="source.tif",
dataset_type="raster",
source="manual",
source_name="manual",
)
def test_detection_geojson_feature_collection_shape() -> None:
project_id = uuid4()
dataset_id = uuid4()
@@ -206,6 +218,7 @@ def test_detection_qa_persists_quality_check_and_metrics() -> None:
db = FakeSession(
objects={
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="detection", status="success", parameters_json={}),
(Dataset, dataset_id): _source_dataset(project_id, dataset_id),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [reference_feature]},
@@ -227,6 +240,8 @@ def test_detection_qa_persists_quality_check_and_metrics() -> None:
assert result["quality_check_id"] == str(quality_checks[0].id)
assert quality_checks[0].analysis_run_id == analysis_run_id
assert quality_checks[0].reference_dataset_id == reference_dataset_id
assert quality_checks[0].parameters_json["temporal_compatibility"]["status"] == "compatible"
assert quality_checks[0].findings_json["temporal_compatibility"]["status"] == "compatible"
assert [metric.metric_key for metric in metrics] == [
"precision",
"recall",
@@ -237,6 +252,53 @@ def test_detection_qa_persists_quality_check_and_metrics() -> None:
]
def test_detection_qa_rejects_non_overlapping_historical_reference_editions() -> None:
project_id = uuid4()
dataset_id = uuid4()
reference_dataset_id = uuid4()
analysis_run_id = uuid4()
source_dataset = _source_dataset(project_id, dataset_id)
source_dataset.source_name = "digitaal_vlaanderen_orthophoto"
source_dataset.source_metadata = {"product_key": "2020", "supports_detection": False}
source_dataset.valid_from = datetime(2020, 1, 1, tzinfo=UTC)
source_dataset.valid_to = datetime(2020, 12, 31, 23, 59, 59, tzinfo=UTC)
reference_dataset = Dataset(
id=reference_dataset_id,
project_id=project_id,
name="current-grb.geojson",
dataset_type="vector",
source="grb",
source_name="grb",
dataset_role="reference",
valid_from=datetime(2026, 7, 1, tzinfo=UTC),
valid_to=datetime(2026, 7, 31, 23, 59, 59, tzinfo=UTC),
)
db = FakeSession(
objects={
(AnalysisRun, analysis_run_id): AnalysisRun(
id=analysis_run_id,
project_id=project_id,
dataset_id=dataset_id,
analysis_type="detection",
status="success",
parameters_json={},
),
(Dataset, dataset_id): source_dataset,
(Dataset, reference_dataset_id): reference_dataset,
},
)
with pytest.raises(AppError) as exc_info:
DetectionService.compare_detections_with_reference(
db=db,
analysis_run_id=analysis_run_id,
reference_dataset_id=reference_dataset_id,
)
assert exc_info.value.code == "DETECTION_QA_TEMPORAL_MISMATCH"
assert db.added == []
def test_detection_qa_no_match_case_persists_zero_scores() -> None:
project_id = uuid4()
dataset_id = uuid4()
@@ -260,6 +322,7 @@ def test_detection_qa_no_match_case_persists_zero_scores() -> None:
db = FakeSession(
objects={
(AnalysisRun, analysis_run_id): AnalysisRun(id=analysis_run_id, project_id=project_id, dataset_id=dataset_id, analysis_type="detection", status="success", parameters_json={}),
(Dataset, dataset_id): _source_dataset(project_id, dataset_id),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [reference_feature]},
@@ -311,6 +374,7 @@ def test_configured_yolo_qa_requires_persisted_tile_manifest_provenance() -> Non
model_name="yolo-configured",
parameters_json={"model_id": "yolo-configured"},
),
(Dataset, dataset_id): _source_dataset(project_id, dataset_id),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [reference_feature]},
@@ -391,6 +455,7 @@ def test_detection_qa_excludes_references_outside_persisted_tile_coverage(tmp_pa
"tile_manifest_path": str(manifest_path),
},
),
(Dataset, dataset_id): _source_dataset(project_id, dataset_id),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [inside_reference, outside_reference]},
@@ -452,6 +517,7 @@ def test_detection_qa_reports_box_to_footprint_diagnostic_without_changing_stric
"tile_manifest_path": str(manifest_path),
},
),
(Dataset, dataset_id): _source_dataset(project_id, dataset_id),
(Dataset, reference_dataset_id): reference_dataset,
},
query_rows={Detection: [detection], VectorFeature: [reference_feature]},