255 lines
8.7 KiB
Python
255 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from geoalchemy2.shape import from_shape
|
|
from shapely.geometry import box
|
|
|
|
from app.core.errors import AppError
|
|
from app.db.session import get_db
|
|
from app.main import app
|
|
from app.models import Detection, DetectionReview, QualityCheck, VectorFeature
|
|
from app.schemas.detection_review import (
|
|
DetectionReviewList,
|
|
DetectionReviewRead,
|
|
DetectionReviewSummary,
|
|
DetectionReviewUpsert,
|
|
)
|
|
from app.services.detection_review_service import DetectionReviewService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows):
|
|
self.rows = list(rows)
|
|
|
|
def filter(self, *criteria):
|
|
for criterion in criteria:
|
|
left = getattr(criterion, "left", None)
|
|
right = getattr(criterion, "right", None)
|
|
operator = getattr(criterion, "operator", None)
|
|
name = getattr(left, "name", None)
|
|
value = getattr(right, "value", right)
|
|
if name and operator and operator.__name__ == "eq":
|
|
self.rows = [row for row in self.rows if getattr(row, name) == value]
|
|
return self
|
|
|
|
def all(self):
|
|
return list(self.rows)
|
|
|
|
def first(self):
|
|
return self.rows[0] if self.rows else None
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, objects=None, query_rows=None) -> None:
|
|
self.objects = objects or {}
|
|
self.query_rows = query_rows or {}
|
|
|
|
def get(self, model, item_id):
|
|
return self.objects.get((model, item_id))
|
|
|
|
def query(self, model):
|
|
return FakeQuery(self.query_rows.setdefault(model, []))
|
|
|
|
def add(self, row):
|
|
rows = self.query_rows.setdefault(type(row), [])
|
|
if row not in rows:
|
|
rows.append(row)
|
|
self.objects[(type(row), row.id)] = row
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def refresh(self, _row):
|
|
return None
|
|
|
|
|
|
def _review_context() -> tuple[FakeSession, UUID, UUID, Detection, VectorFeature]:
|
|
project_id = uuid4()
|
|
quality_check_id = uuid4()
|
|
analysis_run_id = uuid4()
|
|
candidate_dataset_id = uuid4()
|
|
reference_dataset_id = uuid4()
|
|
detection = Detection(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
dataset_id=candidate_dataset_id,
|
|
analysis_run_id=analysis_run_id,
|
|
model_name="yolo-configured",
|
|
class_name="building",
|
|
confidence=0.62,
|
|
geometry=from_shape(box(5.0, 51.0, 5.001, 51.001), srid=4326),
|
|
)
|
|
reference = VectorFeature(
|
|
id=uuid4(),
|
|
dataset_id=reference_dataset_id,
|
|
source_feature_id="grb-missed",
|
|
feature_class="building",
|
|
properties_json={},
|
|
geometry=from_shape(box(5.002, 51.002, 5.003, 51.003), srid=4326),
|
|
)
|
|
quality_check = QualityCheck(
|
|
id=quality_check_id,
|
|
project_id=project_id,
|
|
analysis_run_id=analysis_run_id,
|
|
candidate_dataset_id=candidate_dataset_id,
|
|
reference_dataset_id=reference_dataset_id,
|
|
check_type="detections_vs_reference",
|
|
status="ok",
|
|
findings_json={
|
|
"false_positive_evidence": [{"candidate_feature_id": str(detection.id)}],
|
|
"false_negative_evidence": [{"reference_feature_id": str(reference.id)}],
|
|
},
|
|
)
|
|
db = FakeSession(
|
|
objects={
|
|
(QualityCheck, quality_check_id): quality_check,
|
|
(Detection, detection.id): detection,
|
|
(VectorFeature, reference.id): reference,
|
|
},
|
|
query_rows={DetectionReview: []},
|
|
)
|
|
return db, project_id, quality_check_id, detection, reference
|
|
|
|
|
|
def test_detection_review_model_and_migration_are_aligned() -> None:
|
|
migration = (ROOT / "backend" / "alembic" / "versions" / "202607150001_detection_reviews.py").read_text(encoding="utf-8")
|
|
columns = DetectionReview.__table__.columns
|
|
|
|
for name in (
|
|
"project_id",
|
|
"quality_check_id",
|
|
"analysis_run_id",
|
|
"evidence_role",
|
|
"evidence_feature_id",
|
|
"detection_id",
|
|
"reference_feature_id",
|
|
"decision",
|
|
"notes",
|
|
"reviewed_by",
|
|
"created_at",
|
|
"updated_at",
|
|
):
|
|
assert name in columns
|
|
assert f'"{name}"' in migration
|
|
assert 'op.create_table(\n "detection_reviews"' in migration
|
|
assert 'down_revision = "202607140001"' in migration
|
|
|
|
|
|
def test_detection_review_queue_persists_only_valid_operator_decisions() -> None:
|
|
db, project_id, quality_check_id, detection, _reference = _review_context()
|
|
|
|
initial = DetectionReviewService.list_reviews(
|
|
db,
|
|
project_id=project_id,
|
|
quality_check_id=quality_check_id,
|
|
)
|
|
assert initial.summary.total == 2
|
|
assert initial.summary.reviewed == 0
|
|
assert initial.summary.decision_counts == {"unreviewed": 2}
|
|
|
|
saved = DetectionReviewService.upsert_review(
|
|
db,
|
|
project_id=project_id,
|
|
quality_check_id=quality_check_id,
|
|
payload=DetectionReviewUpsert(
|
|
evidence_role="false_positive",
|
|
evidence_feature_id=str(detection.id),
|
|
decision="qa_alignment_mismatch",
|
|
notes="Box overlaps the official footprint but is not a training negative.",
|
|
),
|
|
)
|
|
assert saved.decision == "qa_alignment_mismatch"
|
|
assert saved.detection_id == detection.id
|
|
|
|
reviewed = DetectionReviewService.list_reviews(
|
|
db,
|
|
project_id=project_id,
|
|
quality_check_id=quality_check_id,
|
|
reviewed=True,
|
|
)
|
|
assert reviewed.total == 1
|
|
assert reviewed.summary.reviewed == 1
|
|
assert reviewed.summary.remaining == 1
|
|
|
|
with pytest.raises(AppError) as exc:
|
|
DetectionReviewService.upsert_review(
|
|
db,
|
|
project_id=project_id,
|
|
quality_check_id=quality_check_id,
|
|
payload=DetectionReviewUpsert(
|
|
evidence_role="false_positive",
|
|
evidence_feature_id=str(detection.id),
|
|
decision="confirmed_model_false_negative",
|
|
),
|
|
)
|
|
assert exc.value.code == "INVALID_DETECTION_REVIEW_DECISION"
|
|
|
|
|
|
def test_detection_review_endpoints_use_canonical_envelopes(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
quality_check_id = uuid4()
|
|
item = DetectionReviewRead(
|
|
project_id=project_id,
|
|
quality_check_id=quality_check_id,
|
|
evidence_role="false_positive",
|
|
evidence_feature_id=str(uuid4()),
|
|
decision="unreviewed",
|
|
)
|
|
result = DetectionReviewList(
|
|
items=[item],
|
|
total=1,
|
|
limit=50,
|
|
offset=0,
|
|
summary=DetectionReviewSummary(
|
|
total=1,
|
|
reviewed=0,
|
|
remaining=1,
|
|
false_positive_total=1,
|
|
false_negative_total=0,
|
|
decision_counts={"unreviewed": 1},
|
|
),
|
|
)
|
|
monkeypatch.setattr(DetectionReviewService, "list_reviews", lambda *_args, **_kwargs: result)
|
|
monkeypatch.setattr(DetectionReviewService, "upsert_review", lambda *_args, **_kwargs: item)
|
|
app.dependency_overrides[get_db] = lambda: FakeSession()
|
|
try:
|
|
listed = TestClient(app).get(f"/api/v1/projects/{project_id}/quality-checks/{quality_check_id}/reviews")
|
|
saved = TestClient(app).post(
|
|
f"/api/v1/projects/{project_id}/quality-checks/{quality_check_id}/reviews",
|
|
json={
|
|
"evidence_role": "false_positive",
|
|
"evidence_feature_id": item.evidence_feature_id,
|
|
"decision": "unreviewed",
|
|
},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.pop(get_db, None)
|
|
|
|
assert listed.status_code == 200
|
|
assert set(listed.json()) == {"data"}
|
|
assert listed.json()["data"]["summary"]["remaining"] == 1
|
|
assert saved.status_code == 200
|
|
assert saved.json() == {"data": item.model_dump(mode="json")}
|
|
|
|
|
|
def test_map_detection_qa_uses_documented_footprint_threshold_and_honest_labels() -> None:
|
|
hook = (ROOT / "frontend" / "src" / "hooks" / "useMapOrthophotoAnalysis.ts").read_text(encoding="utf-8")
|
|
app_source = (ROOT / "frontend" / "src" / "App.tsx").read_text(encoding="utf-8")
|
|
evidence_service = (ROOT / "backend" / "app" / "services" / "quality_evidence_service.py").read_text(encoding="utf-8")
|
|
|
|
assert "MAP_BUILDING_QA_IOU_THRESHOLD = 0.25" in hook
|
|
assert "kandidaten" in hook
|
|
assert "precision" in hook.lower()
|
|
assert "false, iouThreshold" in app_source
|
|
assert "AI-kandidaten" in (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
assert "VectorFeature.id.in_(uuid_identifiers)" in evidence_service
|
|
assert "VectorFeature.source_feature_id.in_(identifiers)" in evidence_service
|
|
assert "for row in db.query(VectorFeature).filter(VectorFeature.dataset_id" not in evidence_service
|