geointel/backend/tests/test_sprint112_qa_evidence_overlay.py

from __future__ import annotations

from uuid import 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 QualityCheck, VectorFeature
from app.services.quality_evidence_service import QualityEvidenceService
from tests.frontend_contract import read_feature


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)


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.get(model, []))


def _vector_feature(dataset_id, *, feature_id=None, source_feature_id: str, geom=None) -> VectorFeature:
    return VectorFeature(
        id=feature_id or uuid4(),
        dataset_id=dataset_id,
        source_feature_id=source_feature_id,
        feature_class="building",
        properties_json={"name": source_feature_id},
        geometry=from_shape(geom or box(4.0, 51.0, 4.1, 51.1), srid=4326),
    )


def test_quality_check_evidence_geojson_resolves_persisted_vector_features() -> None:
    project_id = uuid4()
    quality_check_id = uuid4()
    candidate_dataset_id = uuid4()
    reference_dataset_id = uuid4()
    candidate_match = _vector_feature(candidate_dataset_id, source_feature_id="candidate-match")
    candidate_extra = _vector_feature(candidate_dataset_id, source_feature_id="candidate-extra", geom=box(4.4, 51.4, 4.5, 51.5))
    reference_match = _vector_feature(reference_dataset_id, source_feature_id="reference-match")
    reference_missing = _vector_feature(reference_dataset_id, source_feature_id="reference-missing", geom=box(4.7, 51.7, 4.8, 51.8))
    quality_check = QualityCheck(
        id=quality_check_id,
        project_id=project_id,
        candidate_dataset_id=candidate_dataset_id,
        reference_dataset_id=reference_dataset_id,
        check_type="candidate_vs_reference",
        status="ok",
        findings_json={
            "match_evidence": [
                {
                    "candidate_feature_id": "candidate-match",
                    "reference_feature_id": "reference-match",
                    "iou": 1.0,
                }
            ],
            "false_positive_evidence": [{"candidate_feature_id": "candidate-extra"}],
            "false_negative_evidence": [{"reference_feature_id": "reference-missing"}],
        },
    )
    db = FakeSession(
        objects={(QualityCheck, quality_check_id): quality_check},
        query_rows={VectorFeature: [candidate_match, candidate_extra, reference_match, reference_missing]},
    )

    result = QualityEvidenceService.evidence_geojson(db, project_id=project_id, quality_check_id=quality_check_id)

    assert result["quality_check_id"] == str(quality_check_id)
    assert result["feature_count"] == 4
    assert result["geojson"]["type"] == "FeatureCollection"
    roles = [feature["properties"]["qa_evidence_role"] for feature in result["geojson"]["features"]]
    # Every role resolves to persisted geometry. Errors are emitted before
    # confirmations, because a capped overlay must spend its budget on the
    # objects a reviewer has to act on.
    assert sorted(roles) == ["false_negative", "false_positive", "match_candidate", "match_reference"]
    assert roles.index("false_negative") < roles.index("match_candidate")
    assert roles.index("false_positive") < roles.index("match_candidate")
    match_candidate = next(
        feature
        for feature in result["geojson"]["features"]
        if feature["properties"]["qa_evidence_role"] == "match_candidate"
    )
    assert match_candidate["properties"]["quality_check_id"] == str(quality_check_id)
    assert match_candidate["properties"]["candidate_feature_id"] == "candidate-match"
    assert match_candidate["properties"]["reference_feature_id"] == "reference-match"
    assert match_candidate["properties"]["iou"] == 1.0
    assert match_candidate["properties"]["source_feature_id"] == "candidate-match"


def test_quality_check_evidence_geojson_rejects_cross_project_access() -> None:
    quality_check_id = uuid4()
    quality_check = QualityCheck(
        id=quality_check_id,
        project_id=uuid4(),
        reference_dataset_id=uuid4(),
        check_type="candidate_vs_reference",
        status="ok",
        findings_json={},
    )
    db = FakeSession(objects={(QualityCheck, quality_check_id): quality_check})

    with pytest.raises(AppError) as exc:
        QualityEvidenceService.evidence_geojson(db, project_id=uuid4(), quality_check_id=quality_check_id)

    assert exc.value.code == "QUALITY_CHECK_NOT_FOUND"


def test_quality_check_evidence_geojson_api_uses_canonical_envelope(monkeypatch) -> None:
    project_id = uuid4()
    quality_check_id = uuid4()
    reference_dataset_id = uuid4()
    payload = {
        "quality_check_id": str(quality_check_id),
        "project_id": str(project_id),
        "candidate_dataset_id": None,
        "reference_dataset_id": str(reference_dataset_id),
        "analysis_run_id": None,
        "feature_count": 0,
        "warnings": [],
        "geojson": {"type": "FeatureCollection", "features": []},
    }

    monkeypatch.setattr(
        "app.api.routes.quality_checks.QualityEvidenceService.evidence_geojson",
        lambda *_args, **_kwargs: payload,
    )
    app.dependency_overrides[get_db] = lambda: FakeSession()
    try:
        response = TestClient(app).get(f"/api/v1/projects/{project_id}/quality-checks/{quality_check_id}/evidence/geojson")
    finally:
        app.dependency_overrides.pop(get_db, None)

    assert response.status_code == 200
    # The envelope wraps the service result; asserting the exact field list
    # would break every time the response model gains a documented field.
    body = response.json()
    assert set(body) == {"data"}
    assert body["data"].items() >= payload.items()


def test_frontend_quality_evidence_overlay_contract_is_wired() -> None:
    from pathlib import Path

    root = Path(__file__).resolve().parents[2]
    geo_map = (root / "frontend" / "src" / "components" / "GeoMap.tsx").read_text(encoding="utf-8")
    map_workspace = read_feature("map_workspace")
    qa_api = (root / "frontend" / "src" / "services" / "api" / "qa.ts").read_text(encoding="utf-8")

    assert "qaEvidenceData" in geo_map
    assert "qa-evidence-fill" in geo_map
    assert "qa_evidence_role" in geo_map
    assert "qualityEvidenceGeoJson" in map_workspace
    assert "getQualityEvidenceGeoJson" in qa_api