from __future__ import annotations import asyncio import json from pathlib import Path from uuid import uuid4 import pytest from geoalchemy2.shape import to_shape from pyproj import Transformer from app.api.routes.qa import compare_candidate_with_reference from app.models import Dataset, Metric, Project, QualityCheck, VectorFeature from app.schemas.qa import QaProviderComparisonRequest from app.providers.registry import list_provider_capabilities from app.services.dataset_service import DatasetService from app.services.quality_service import QualityService from app.services.vector_feature_service import VectorFeatureService class FakeSession: def __init__(self, objects=None) -> None: self.added = [] self.objects = objects or {} self.commits = 0 self.refreshes = [] self.flushes = 0 self.rollbacks = 0 def get(self, model, item_id): return self.objects.get((model, item_id)) def add(self, item) -> None: self.added.append(item) def commit(self) -> None: self.commits += 1 def flush(self) -> None: self.flushes += 1 def refresh(self, item) -> None: self.refreshes.append(item) def rollback(self) -> None: self.rollbacks += 1 def test_vector_feature_service_persists_geojson_features_with_properties() -> None: db = FakeSession() dataset_id = uuid4() payload = { "type": "FeatureCollection", "features": [ { "type": "Feature", "id": "building-1", "properties": {"class": "building", "height": 7}, "geometry": { "type": "Polygon", "coordinates": [ [ [4.0, 51.0], [4.1, 51.0], [4.1, 51.1], [4.0, 51.1], [4.0, 51.0], ] ], }, } ], } persisted = VectorFeatureService.persist_geojson_features( db=db, dataset_id=dataset_id, payload=payload, feature_class="building", ) assert len(persisted) == 1 assert isinstance(persisted[0], VectorFeature) assert persisted[0].dataset_id == dataset_id assert persisted[0].feature_class == "building" assert persisted[0].source_feature_id == "building-1" assert persisted[0].properties_json == {"class": "building", "height": 7} assert db.added == persisted assert db.flushes == 1 assert db.commits == 1 assert db.refreshes == [] def test_vector_feature_service_normalizes_source_z_coordinates_to_canonical_2d() -> None: db = FakeSession() persisted = VectorFeatureService.persist_geojson_features( db=db, dataset_id=uuid4(), payload={ "type": "FeatureCollection", "features": [ { "type": "Feature", "id": "sector-3d", "properties": {"population_total": 100}, "geometry": { "type": "Polygon", "coordinates": [ [[5.0, 51.0, 0.0], [5.1, 51.0, 0.0], [5.1, 51.1, 0.0], [5.0, 51.0, 0.0]] ], }, } ], }, feature_class="population", ) assert len(persisted) == 1 assert to_shape(persisted[0].geometry).has_z is False def test_vector_feature_service_transforms_declared_source_crs_before_epsg4326_storage() -> None: to_lambert = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True) x, y = to_lambert.transform(4.7, 51.1) db = FakeSession() persisted = VectorFeatureService.persist_geojson_features( db=db, dataset_id=uuid4(), payload={ "type": "FeatureCollection", "features": [ { "type": "Feature", "id": "lambert-point", "properties": {}, "geometry": {"type": "Point", "coordinates": [x, y]}, } ], }, source_crs="EPSG:31370", ) geometry = to_shape(persisted[0].geometry) assert geometry.x == pytest.approx(4.7, abs=0.000001) assert geometry.y == pytest.approx(51.1, abs=0.000001) def test_vector_feature_service_rejects_invalid_declared_source_crs() -> None: with pytest.raises(Exception) as exc_info: VectorFeatureService.persist_geojson_features( db=FakeSession(), dataset_id=uuid4(), payload={ "type": "FeatureCollection", "features": [ { "type": "Feature", "properties": {}, "geometry": {"type": "Point", "coordinates": [4.7, 51.1]}, } ], }, source_crs="EPSG:not-a-crs", ) assert getattr(exc_info.value, "code", None) == "INVALID_DATASET_CRS" def test_dataset_upload_persists_vector_features(monkeypatch, tmp_path) -> None: project_id = uuid4() db = FakeSession(objects={(Project, project_id): Project(id=project_id, name="Geel")}) payload = { "type": "FeatureCollection", "features": [ { "type": "Feature", "properties": {"class": "building"}, "geometry": { "type": "Point", "coordinates": [4.0, 51.0], }, } ], } class Upload: filename = "reference.geojson" content_type = "application/geo+json" def __init__(self) -> None: import json self._content = json.dumps(payload).encode("utf-8") async def read(self, size: int) -> bytes: chunk, self._content = self._content[:size], self._content[size:] return chunk storage_path = tmp_path / "reference.geojson" storage_path.write_text(json.dumps(payload), encoding="utf-8") async def persist_upload_file(**_kwargs): return { "storage_path": str(storage_path), "original_filename": "reference.geojson", "stored_filename": "reference.geojson", "content_type": "application/geo+json", "size_bytes": storage_path.stat().st_size, "checksum_sha256": "0" * 64, } monkeypatch.setattr( "app.services.dataset_service.StorageService.persist_upload_file", persist_upload_file, ) result = asyncio.run( DatasetService.upload_dataset( db=db, project_id=project_id, file=Upload(), dataset_type="vector", source="user_upload", dataset_role="reference", reference_layer_name="buildings", ) ) persisted_features = [item for item in db.added if isinstance(item, VectorFeature)] assert result.dataset_role == "reference" assert result.source_name == "manual" assert len(persisted_features) == 1 assert persisted_features[0].dataset_id == result.id assert db.commits == 1 def test_dataset_upload_rolls_back_dataset_and_file_when_vector_indexing_fails(monkeypatch, tmp_path) -> None: project_id = uuid4() db = FakeSession(objects={(Project, project_id): Project(id=project_id, name="Mol")}) storage_path = tmp_path / "invalid.geojson" storage_path.write_text("{}", encoding="utf-8") class Upload: filename = "invalid.geojson" content_type = "application/geo+json" def __init__(self) -> None: self._content = b'{"type":"FeatureCollection","features":[]}' async def read(self, size: int) -> bytes: chunk, self._content = self._content[:size], self._content[size:] return chunk storage_path.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8") async def persist_upload_file(**_kwargs): return { "storage_path": str(storage_path), "original_filename": "invalid.geojson", "stored_filename": "invalid.geojson", "content_type": "application/geo+json", "size_bytes": storage_path.stat().st_size, "checksum_sha256": "0" * 64, } monkeypatch.setattr( "app.services.dataset_service.StorageService.persist_upload_file", persist_upload_file, ) monkeypatch.setattr( VectorFeatureService, "persist_geojson_features", lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("PostGIS indexing failed")), ) with pytest.raises(RuntimeError, match="PostGIS indexing failed"): asyncio.run( DatasetService.upload_dataset( db=db, project_id=project_id, file=Upload(), dataset_type="vector", source="user_upload", ) ) assert db.commits == 0 assert db.rollbacks == 1 assert storage_path.exists() is False def test_quality_service_persists_quality_check_and_metrics() -> None: db = FakeSession() project_id = uuid4() candidate_dataset_id = uuid4() reference_dataset_id = uuid4() job_id = uuid4() quality_check = QualityService.persist_quality_check( db=db, project_id=project_id, reference_dataset_id=reference_dataset_id, check_type="candidate_vs_reference", status="ok", score=1.0, parameters={"iou_threshold": 0.5}, findings={"matches": 1, "false_positives": 0, "false_negatives": 0}, candidate_dataset_id=candidate_dataset_id, job_id=job_id, metrics={ "precision": 1.0, "recall": 1.0, "f1": 1.0, "false_positive_count": 0, }, ) assert isinstance(quality_check, QualityCheck) assert quality_check.project_id == project_id assert quality_check.job_id == job_id assert quality_check.candidate_dataset_id == candidate_dataset_id assert quality_check.reference_dataset_id == reference_dataset_id assert quality_check.parameters_json == {"iou_threshold": 0.5} assert quality_check.findings_json["matches"] == 1 persisted_metrics = [item for item in db.added if isinstance(item, Metric)] assert [metric.metric_key for metric in persisted_metrics] == [ "precision", "recall", "f1", "false_positive_count", ] assert persisted_metrics[0].quality_check_id == quality_check.id assert db.flushes == 1 assert db.commits == 1 def test_dataset_role_validation_accepts_only_source_derived_reference() -> None: assert DatasetService._normalize_dataset_role("source") == "source" assert DatasetService._normalize_dataset_role("derived") == "derived" assert DatasetService._normalize_dataset_role("reference") == "reference" with pytest.raises(Exception) as exc_info: DatasetService._normalize_dataset_role("osm") assert getattr(exc_info.value, "code", None) == "INVALID_DATASET_ROLE" def test_provider_capabilities_expose_sprint7a_contract() -> None: capabilities = {capability.provider_name: capability.to_dict() for capability in list_provider_capabilities()} assert capabilities["osm"]["supported_layers"] == ["buildings", "roads", "water", "landuse"] assert capabilities["osm"]["supported_geometry_types"] == ["Polygon", "MultiPolygon", "LineString", "MultiLineString"] assert capabilities["osm"]["supported_query_modes"] == ["area"] assert capabilities["osm"]["status"] == "not_configured" assert capabilities["grb"]["supported_layers"] == ["buildings", "roads", "water", "parcels"] assert capabilities["grb"]["supported_geometry_types"] == ["Polygon", "MultiPolygon", "LineString", "MultiLineString"] assert capabilities["grb"]["supported_query_modes"] == ["bbox", "persisted_area"] assert capabilities["grb"]["status"] == "configured" def test_sprint7a_migration_declares_foundation_tables_and_indexes() -> None: migration_path = Path(__file__).parents[1] / "alembic" / "versions" / "202606120700_sprint7a_persistence_foundation.py" migration_text = migration_path.read_text(encoding="utf-8") for required_text in ( "vector_features", "quality_checks", "metrics", "ix_vector_features_geometry", 'postgresql_using="gist"', "ix_quality_checks_project_id", "ix_metrics_quality_check_id", ): assert required_text in migration_text def test_qa_route_persists_quality_check_domain_record(monkeypatch) -> None: project_id = uuid4() candidate_dataset_id = uuid4() reference_dataset_id = uuid4() job_id = uuid4() candidate_dataset = Dataset( id=candidate_dataset_id, project_id=project_id, name="candidate.geojson", dataset_type="vector", source="test", ) db = FakeSession(objects={(Dataset, candidate_dataset_id): candidate_dataset}) def run_sync_job(**kwargs): result = kwargs["operation"]() return { "id": str(job_id), "project_id": str(project_id), "status": "success", "result_json": result, } monkeypatch.setattr("app.api.routes.qa.JobService.run_sync_job", run_sync_job) monkeypatch.setattr( "app.api.routes.qa.QaService.compare_candidate_with_reference", lambda **_kwargs: type( "Result", (), { "model_dump": lambda self, **_kwargs: { "status": "ok", "matches": 1, "false_positives": 0, "false_negatives": 0, "precision": 1.0, "recall": 1.0, "f1_score": 1.0, "mean_iou": 1.0, "iou_threshold": 0.5, "warnings": [], "match_evidence": [ { "candidate_feature_id": "candidate-1", "reference_feature_id": "reference-1", "iou": 1.0, } ], "false_positive_evidence": [{"candidate_feature_id": "candidate-extra"}], "false_negative_evidence": [{"reference_feature_id": "reference-missing"}], } }, )(), ) response = compare_candidate_with_reference( payload=QaProviderComparisonRequest( candidate_dataset_id=candidate_dataset_id, reference_dataset_id=reference_dataset_id, iou_threshold=0.5, ), db=db, ) persisted_quality_checks = [item for item in db.added if isinstance(item, QualityCheck)] persisted_metrics = [item for item in db.added if isinstance(item, Metric)] assert response["data"]["result_json"]["quality_check_id"] == str(persisted_quality_checks[0].id) assert persisted_quality_checks[0].job_id == job_id assert persisted_quality_checks[0].candidate_dataset_id == candidate_dataset_id assert persisted_quality_checks[0].reference_dataset_id == reference_dataset_id assert persisted_quality_checks[0].findings_json["match_evidence"][0]["candidate_feature_id"] == "candidate-1" assert persisted_quality_checks[0].findings_json["false_positive_evidence"][0]["candidate_feature_id"] == "candidate-extra" assert persisted_quality_checks[0].findings_json["false_negative_evidence"][0]["reference_feature_id"] == "reference-missing" assert [metric.metric_key for metric in persisted_metrics] == [ "precision", "recall", "f1", "mean_iou", "false_positive_count", "false_negative_count", ]