geointel/backend/tests/test_sprint197_accuracy_review_loop.py

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
from tests.frontend_contract import read_feature


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 = read_feature("shell")
    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 read_feature("map_workspace")
    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