298 lines
10 KiB
Python
298 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
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
|
|
|
|
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 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.commits == 1
|
|
|
|
|
|
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"
|
|
|
|
async def read(self) -> bytes:
|
|
import json
|
|
|
|
return json.dumps(payload).encode("utf-8")
|
|
|
|
storage_path = tmp_path / "reference.geojson"
|
|
storage_path.write_text("{}", encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
"app.services.dataset_service.StorageService.persist_dataset_file",
|
|
lambda **_kwargs: {
|
|
"storage_path": str(storage_path),
|
|
"original_filename": "reference.geojson",
|
|
"stored_filename": "reference.geojson",
|
|
"content_type": "application/geo+json",
|
|
"size_bytes": 2,
|
|
"checksum_sha256": "0" * 64,
|
|
},
|
|
)
|
|
|
|
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
|
|
|
|
|
|
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", "parcels"]
|
|
assert capabilities["grb"]["supported_geometry_types"] == ["Polygon", "MultiPolygon", "LineString", "MultiLineString"]
|
|
assert capabilities["grb"]["supported_query_modes"] == ["area"]
|
|
assert capabilities["grb"]["status"] == "not_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": [],
|
|
}
|
|
},
|
|
)(),
|
|
)
|
|
|
|
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 [metric.metric_key for metric in persisted_metrics] == [
|
|
"precision",
|
|
"recall",
|
|
"f1",
|
|
"mean_iou",
|
|
"false_positive_count",
|
|
"false_negative_count",
|
|
]
|