397 lines
14 KiB
Python
397 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from urllib.parse import parse_qs, urlparse
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from geoalchemy2.shape import from_shape
|
|
from shapely.geometry import MultiPolygon, Polygon
|
|
|
|
from app.core.config import Settings
|
|
from app.core.errors import AppError
|
|
from app.db.session import get_db
|
|
from app.main import app
|
|
from app.models import Area, Dataset, Job, Project
|
|
from app.schemas.grb import GrbAcquireRequest
|
|
from app.services.dataset_service import DatasetService
|
|
from app.services.grb_acquisition_service import GrbAcquisitionService
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, result=None):
|
|
self.result = result
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def order_by(self, *_args):
|
|
return self
|
|
|
|
def all(self):
|
|
return self.result if isinstance(self.result, list) else []
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows=None, query_result=None):
|
|
self.rows = rows or {}
|
|
self.query_result = query_result
|
|
self.added = []
|
|
|
|
def get(self, model, row_id):
|
|
row = self.rows.get((model, row_id))
|
|
if row is not None:
|
|
return row
|
|
return next((item for item in self.added if isinstance(item, model) and item.id == row_id), None)
|
|
|
|
def add(self, row):
|
|
self.added.append(row)
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def rollback(self):
|
|
return None
|
|
|
|
def refresh(self, row):
|
|
return row
|
|
|
|
def query(self, _model):
|
|
return FakeQuery(self.query_result)
|
|
|
|
|
|
class JsonResponse:
|
|
def __init__(self, payload):
|
|
self.content = json.dumps(payload).encode("utf-8")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, size=-1):
|
|
return self.content if size < 0 else self.content[:size]
|
|
|
|
|
|
def request(*, product_key="buildings", area_id=None, force_refresh=True) -> GrbAcquireRequest:
|
|
return GrbAcquireRequest(
|
|
bbox={
|
|
"min_x": 5.15,
|
|
"min_y": 51.18,
|
|
"max_x": 5.17,
|
|
"max_y": 51.20,
|
|
"crs": "EPSG:4326",
|
|
},
|
|
area_id=area_id,
|
|
product_key=product_key,
|
|
force_refresh=force_refresh,
|
|
)
|
|
|
|
|
|
def polygon_feature(feature_id: str, coordinates) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": {"type": "Polygon", "coordinates": [coordinates]},
|
|
"properties": {"source_field": feature_id},
|
|
}
|
|
|
|
|
|
def test_grb_registry_exposes_four_governed_products() -> None:
|
|
products = {item["key"]: item for item in GrbAcquisitionService.list_products()}
|
|
|
|
assert set(products) == {"buildings", "roads", "water", "parcels"}
|
|
assert products["buildings"]["collections"] == ["GBG"]
|
|
assert products["roads"]["collections"] == ["Wegsegment"]
|
|
assert products["water"]["collections"] == ["WTZ", "WLAS", "WGR"]
|
|
assert products["parcels"]["collections"] == ["ADP"]
|
|
assert all(item["authority_level"] == "authoritative" for item in products.values())
|
|
|
|
|
|
def test_grb_fetch_follows_pagination_clips_geometry_and_preserves_official_identity() -> None:
|
|
product = GrbAcquisitionService._product("buildings")
|
|
settings = Settings(_env_file=None)
|
|
pages = []
|
|
|
|
first = polygon_feature(
|
|
"GBG.1",
|
|
[(5.155, 51.185), (5.175, 51.185), (5.175, 51.195), (5.155, 51.195), (5.155, 51.185)],
|
|
)
|
|
second = polygon_feature(
|
|
"GBG.2",
|
|
[(5.151, 51.181), (5.152, 51.181), (5.152, 51.182), (5.151, 51.182), (5.151, 51.181)],
|
|
)
|
|
outside = polygon_feature(
|
|
"GBG.3",
|
|
[(5.3, 51.3), (5.31, 51.3), (5.31, 51.31), (5.3, 51.31), (5.3, 51.3)],
|
|
)
|
|
|
|
def opener(raw_request, timeout):
|
|
assert timeout == settings.grb_timeout_seconds
|
|
parsed = urlparse(raw_request.full_url)
|
|
query = parse_qs(parsed.query)
|
|
pages.append(raw_request.full_url)
|
|
assert query["bbox-crs"] == [GrbAcquisitionService.OGC_CRS84_URI]
|
|
assert query["crs"] == [GrbAcquisitionService.OGC_CRS84_URI]
|
|
if query.get("cursor") == ["next"]:
|
|
return JsonResponse({"type": "FeatureCollection", "features": [second, outside], "links": []})
|
|
return JsonResponse(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [first],
|
|
"links": [
|
|
{
|
|
"rel": "next",
|
|
"href": (
|
|
"https://geo.api.vlaanderen.be/GRB/ogc/features/v1/"
|
|
"collections/GBG/items?cursor=next"
|
|
),
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
scope = Polygon(
|
|
[(5.15, 51.18), (5.17, 51.18), (5.17, 51.20), (5.15, 51.20), (5.15, 51.18)]
|
|
)
|
|
features, transfer = GrbAcquisitionService._fetch_features(
|
|
product,
|
|
scope,
|
|
scope.bounds,
|
|
"bounded_selection",
|
|
settings,
|
|
opener,
|
|
)
|
|
|
|
assert len(pages) == 2
|
|
assert transfer["page_count"] == 2
|
|
assert transfer["candidate_feature_count"] == 3
|
|
assert transfer["feature_count"] == 2
|
|
assert transfer["reference_truncated"] is False
|
|
assert {feature["id"] for feature in features} == {"GBG:GBG.1", "GBG:GBG.2"}
|
|
clipped = next(feature for feature in features if feature["id"] == "GBG:GBG.1")
|
|
assert clipped["properties"]["source_feature_id"] == "GBG:GBG.1"
|
|
assert clipped["properties"]["geometry_clipped_to_selection"] is True
|
|
assert clipped["properties"]["coverage_scope"] == "bounded_selection"
|
|
|
|
|
|
def test_grb_fetch_rejects_untrusted_pagination_and_unbounded_feature_volume() -> None:
|
|
product = GrbAcquisitionService._product("buildings")
|
|
settings = Settings(_env_file=None)
|
|
feature = polygon_feature(
|
|
"GBG.1",
|
|
[(5.151, 51.181), (5.152, 51.181), (5.152, 51.182), (5.151, 51.182), (5.151, 51.181)],
|
|
)
|
|
scope = Polygon(
|
|
[(5.15, 51.18), (5.17, 51.18), (5.17, 51.20), (5.15, 51.20), (5.15, 51.18)]
|
|
)
|
|
|
|
def hostile_opener(_request, timeout):
|
|
del timeout
|
|
return JsonResponse(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [feature],
|
|
"links": [{"rel": "next", "href": "https://example.test/private"}],
|
|
}
|
|
)
|
|
|
|
with pytest.raises(AppError) as invalid_next:
|
|
GrbAcquisitionService._fetch_features(
|
|
product,
|
|
scope,
|
|
scope.bounds,
|
|
"bounded_selection",
|
|
settings,
|
|
hostile_opener,
|
|
)
|
|
assert invalid_next.value.code == "GRB_PROVIDER_INVALID_PAGINATION"
|
|
|
|
def oversized_opener(_request, timeout):
|
|
del timeout
|
|
return JsonResponse(
|
|
{
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
feature,
|
|
{**feature, "id": "GBG.2"},
|
|
],
|
|
"links": [],
|
|
}
|
|
)
|
|
|
|
with pytest.raises(AppError) as oversized:
|
|
GrbAcquisitionService._fetch_features(
|
|
product,
|
|
scope,
|
|
scope.bounds,
|
|
"bounded_selection",
|
|
Settings(_env_file=None, GRB_MAX_FEATURES=1),
|
|
oversized_opener,
|
|
)
|
|
assert oversized.value.code == "GRB_SELECTION_TOO_LARGE"
|
|
|
|
|
|
def test_grb_acquisition_rejects_large_scope_before_network_access() -> None:
|
|
project_id = uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Vlaanderen")})
|
|
payload = GrbAcquireRequest(
|
|
bbox={"min_x": 4.0, "min_y": 50.7, "max_x": 5.0, "max_y": 51.7, "crs": "EPSG:4326"},
|
|
product_key="buildings",
|
|
)
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
GrbAcquisitionService.acquire(db, project_id, payload, settings=Settings(_env_file=None))
|
|
|
|
assert exc_info.value.code == "GRB_SELECTION_TOO_LARGE"
|
|
|
|
|
|
def test_grb_acquisition_persists_via_dataset_service_with_selection_metrics(monkeypatch) -> None:
|
|
project_id, area_id, dataset_id = uuid4(), uuid4(), uuid4()
|
|
municipality = MultiPolygon(
|
|
[
|
|
Polygon(
|
|
[(5.15, 51.18), (5.17, 51.18), (5.17, 51.20), (5.15, 51.20), (5.15, 51.18)]
|
|
)
|
|
]
|
|
)
|
|
project = Project(id=project_id, name="Vlaanderen")
|
|
area = Area(
|
|
id=area_id,
|
|
project_id=project_id,
|
|
name="Gemeente Mol - officieel",
|
|
geometry=from_shape(municipality, srid=4326),
|
|
)
|
|
db = FakeSession({(Project, project_id): project, (Area, area_id): area})
|
|
captured = {}
|
|
|
|
def opener(_request, timeout):
|
|
del timeout
|
|
parsed = urlparse(_request.full_url)
|
|
collection = parsed.path.split("/")[-2]
|
|
if collection == "WTZ":
|
|
features = [
|
|
polygon_feature(
|
|
"WTZ.1",
|
|
[(5.151, 51.181), (5.16, 51.181), (5.16, 51.19), (5.151, 51.19), (5.151, 51.181)],
|
|
)
|
|
]
|
|
else:
|
|
features = []
|
|
return JsonResponse({"type": "FeatureCollection", "features": features, "links": []})
|
|
|
|
def persist(_db, **kwargs):
|
|
captured.update(kwargs)
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
area_id=area_id,
|
|
name=kwargs["filename"],
|
|
dataset_type="vector",
|
|
source=kwargs["source"],
|
|
dataset_role=kwargs["dataset_role"],
|
|
source_name=kwargs["source_name"],
|
|
reference_layer_name=kwargs["reference_layer_name"],
|
|
temporal_series_key=kwargs["temporal_series_key"],
|
|
observed_at=kwargs["observed_at"],
|
|
source_version=kwargs["source_version"],
|
|
source_metadata=kwargs["source_metadata"],
|
|
provenance_metadata=kwargs["provenance_metadata"],
|
|
metadata_json={"feature_count": 1},
|
|
status="ready",
|
|
)
|
|
db.rows[(Dataset, dataset_id)] = dataset
|
|
return SimpleNamespace(id=dataset_id)
|
|
|
|
monkeypatch.setattr(DatasetService, "import_vector_bytes", persist)
|
|
result = GrbAcquisitionService.acquire(
|
|
db,
|
|
project_id,
|
|
request(product_key="water", area_id=area_id),
|
|
settings=Settings(_env_file=None),
|
|
opener=opener,
|
|
)
|
|
|
|
assert result["output_dataset_id"] == str(dataset_id)
|
|
assert result["feature_count"] == 1
|
|
assert captured["dataset_role"] == "reference"
|
|
assert captured["source_name"] == "grb"
|
|
assert captured["reference_layer_name"] == "water"
|
|
assert captured["source_metadata"]["coverage_scope"] == "municipality"
|
|
assert captured["source_metadata"]["selection_aggregation"]["metric_key"] == "water_area"
|
|
assert captured["source_metadata"]["selection_metrics"][0]["metric_key"] == "water_length"
|
|
assert captured["provenance_metadata"]["reference_truncated"] is False
|
|
collection = json.loads(captured["content"])
|
|
assert collection["features"][0]["properties"]["coverage_scope"] == "municipality"
|
|
|
|
|
|
def test_grb_routes_use_canonical_envelopes_and_existing_job_contract(monkeypatch) -> None:
|
|
project_id, dataset_id = uuid4(), uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")})
|
|
monkeypatch.setattr(
|
|
GrbAcquisitionService,
|
|
"acquire",
|
|
lambda *_args, **_kwargs: {
|
|
"output_dataset_id": str(dataset_id),
|
|
"provider": "grb",
|
|
"product_key": "buildings",
|
|
"feature_count": 2,
|
|
},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
try:
|
|
client = TestClient(app)
|
|
products_response = client.get(f"/api/v1/projects/{project_id}/datasets/grb/products")
|
|
acquire_response = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/grb/acquire",
|
|
json=request().model_dump(mode="json"),
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert products_response.status_code == 200
|
|
assert set(products_response.json()) == {"data"}
|
|
assert products_response.json()["data"]["total"] == 4
|
|
assert acquire_response.status_code == 200
|
|
assert set(acquire_response.json()) == {"data"}
|
|
assert acquire_response.json()["data"]["job_type"] == "vector.grb.acquire"
|
|
assert acquire_response.json()["data"]["output_dataset_id"] == str(dataset_id)
|
|
assert any(isinstance(item, Job) for item in db.added)
|
|
|
|
|
|
def test_system_capabilities_reports_bounded_grb_integration() -> None:
|
|
response = TestClient(app).get("/api/v1/system/capabilities")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["grb"] == "bounded"
|
|
grb = next(
|
|
item for item in response.json()["data"]["providers"]
|
|
if item["provider_name"] == "grb"
|
|
)
|
|
assert grb["status"] == "configured"
|
|
assert grb["fetch_signature"].endswith("/datasets/grb/acquire")
|
|
|
|
|
|
def test_grb_frontend_and_contracts_use_only_the_governed_backend_path() -> None:
|
|
selection_hook = (ROOT / "frontend/src/hooks/useMapThemeSelectionInsights.ts").read_text(encoding="utf-8")
|
|
catalog_hook = (ROOT / "frontend/src/hooks/useOfficialMapProducts.ts").read_text(encoding="utf-8")
|
|
workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
contracts = (ROOT / "docs/API_CONTRACTS.md").read_text(encoding="utf-8")
|
|
|
|
assert "datasetsApi.acquireGrb" in selection_hook
|
|
assert "datasetsApi.listGrbProducts" in catalog_hook
|
|
assert "officialMapProducts.grb" in workspace
|
|
assert "result[product.key] = null" in workspace
|
|
assert ": onDemandThemeActive\n ? null\n : mapFeatureCollection" in workspace
|
|
assert "onSetContextLayerLabel" in workspace
|
|
assert "'Op aanvraag'" in workspace
|
|
assert "/datasets/grb/acquire" in contracts
|
|
assert "geo.api.vlaanderen.be" not in workspace
|