JensandClaude Opus 5
6572e4ad5f
93 test files read a single frontend source and asserted identifiers in it. The
MapWorkspace split showed what that costs: 24 tests went red for a move that
changed no behaviour at all. A contract belongs to the feature — a container,
its hooks, its domain layer — not to whichever file currently holds it.
232 read sites now resolve through read_feature(). The distinction that makes
this safe is direction: a *positive* contract ("this is wired") may widen,
because the identifier must still exist somewhere in the feature; a *negative*
one ("this component performs no transport") is a statement about one file, and
widening it would quietly weaken the check. The 73 single-file reads that
remain are exactly those, and a guard now enforces the rule for new tests.
Verified rather than assumed: of the 732 migrated positive assertions, 644 still
match exactly one module — as specific as before — and the other 86 already
spanned a container and its hook by nature. Two apparent misses are an artefact
of the checking regex reading an escaped newline literally.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
181 lines
7.2 KiB
Python
181 lines
7.2 KiB
Python
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
|