from __future__ import annotations import re import uuid from pathlib import Path from types import SimpleNamespace from geoalchemy2.shape import from_shape, to_shape from shapely.geometry import Polygon, box from app.core.errors import AppError from app.models import Dataset, VectorFeature from app.services.vector_feature_service import VectorFeatureService from tests.frontend_contract import assert_calls, assert_mentions, assert_wired, read_map_workspace, read_feature ROOT = Path(__file__).resolve().parents[2] class _FakeQuery: def __init__(self, rows: list[VectorFeature]) -> None: self.rows = rows self.limit_value: int | None = None def filter(self, *args, **kwargs): # noqa: ANN002, ANN003 return self def order_by(self, *args, **kwargs): # noqa: ANN002, ANN003 return self def limit(self, value: int): self.limit_value = value return self def all(self) -> list[VectorFeature]: if self.limit_value is None: return self.rows return self.rows[: self.limit_value] class _FakeSession: def __init__(self, rows: list[VectorFeature]) -> None: self.rows = rows def query(self, model): # noqa: ANN001 assert model is VectorFeature return _FakeQuery(self.rows) def _feature_row(dataset_id: uuid.UUID, *, source_feature_id: str, name: str) -> VectorFeature: return VectorFeature( id=uuid.uuid4(), dataset_id=dataset_id, feature_class="parcel", source_feature_id=source_feature_id, properties_json={"name": name}, geometry=from_shape( Polygon( [ (5.0, 51.0), (5.001, 51.0), (5.001, 51.001), (5.0, 51.001), (5.0, 51.0), ] ), srid=4326, ), ) def test_vector_feature_service_extracts_bbox_geojson_from_persisted_rows() -> None: dataset_id = uuid.uuid4() rows = [_feature_row(dataset_id, source_feature_id="src-1", name="Test parcel")] result = VectorFeatureService.select_features_by_bbox( _FakeSession(rows), dataset_id=dataset_id, bbox={"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"}, limit=25, ) assert result["feature_count"] == 1 assert result["total_feature_count"] == 1 assert result["truncated"] is False assert result["geojson"]["type"] == "FeatureCollection" feature = result["geojson"]["features"][0] assert feature["type"] == "Feature" assert feature["properties"]["vector_feature_id"] == str(rows[0].id) assert feature["properties"]["source_feature_id"] == "src-1" assert feature["properties"]["feature_class"] == "parcel" assert feature["properties"]["name"] == "Test parcel" assert feature["geometry"]["type"] == "Polygon" def test_vector_feature_service_rejects_invalid_bbox() -> None: try: VectorFeatureService.select_features_by_bbox( _FakeSession([]), dataset_id=uuid.uuid4(), bbox={"min_x": 5.2, "min_y": 50.9, "max_x": 5.0, "max_y": 51.2, "crs": "EPSG:4326"}, ) except AppError as exc: assert exc.code == "INVALID_SELECTION_BBOX" else: # pragma: no cover raise AssertionError("Expected INVALID_SELECTION_BBOX") def test_vector_select_route_is_project_scoped_and_enveloped(monkeypatch) -> None: from app.api.routes import datasets as dataset_routes project_id = uuid.uuid4() dataset_id = uuid.uuid4() dataset = Dataset(id=dataset_id, project_id=project_id, dataset_type="vector", source="fixture", name="Vector") expected_payload = { "selection_bbox": {"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"}, "feature_count": 0, "limit": 100, "truncated": False, "geojson": {"type": "FeatureCollection", "features": []}, } monkeypatch.setattr(dataset_routes.DatasetService, "get_dataset", lambda db, selected_id: dataset) monkeypatch.setattr( dataset_routes.VectorFeatureService, "select_features_by_bbox", lambda db, dataset_id, bbox, limit=100: expected_payload, ) response = dataset_routes.select_vector_features( project_id=project_id, dataset_id=dataset_id, payload=dataset_routes.VectorSelectionRequest( bbox=dataset_routes.VectorSelectionBBox(min_x=4.9, min_y=50.9, max_x=5.2, max_y=51.2), limit=100, ), db=SimpleNamespace(), ) assert response == {"data": expected_payload} def test_vector_select_route_rejects_non_vector_dataset(monkeypatch) -> None: from app.api.routes import datasets as dataset_routes project_id = uuid.uuid4() dataset_id = uuid.uuid4() dataset = Dataset(id=dataset_id, project_id=project_id, dataset_type="raster", source="fixture", name="Raster") monkeypatch.setattr(dataset_routes.DatasetService, "get_dataset", lambda db, selected_id: dataset) try: dataset_routes.select_vector_features( project_id=project_id, dataset_id=dataset_id, payload=dataset_routes.VectorSelectionRequest( bbox=dataset_routes.VectorSelectionBBox(min_x=4.9, min_y=50.9, max_x=5.2, max_y=51.2), limit=100, ), db=SimpleNamespace(), ) except AppError as exc: assert exc.code == "DATASET_NOT_VECTOR" else: # pragma: no cover raise AssertionError("Expected DATASET_NOT_VECTOR") def test_vector_select_route_uses_persisted_area_geometry_when_requested(monkeypatch) -> None: from app.api.routes import datasets as dataset_routes project_id = uuid.uuid4() dataset_id = uuid.uuid4() area_id = uuid.uuid4() dataset = Dataset( id=dataset_id, project_id=project_id, dataset_type="vector", source="fixture", name="Regional vector", source_metadata={"selection_aggregation": {"method": "feature_count"}}, ) area_geometry = from_shape(box(5.0, 51.0, 5.3, 51.3), srid=4326) area = SimpleNamespace(id=area_id, project_id=project_id, geometry=area_geometry) captured: dict[str, object] = {} class _AreaSession: @staticmethod def get(model, selected_id): # noqa: ANN001 assert model is dataset_routes.Area assert selected_id == area_id return area def select_features(db, **kwargs): # noqa: ANN001 captured["select"] = kwargs return { "selection_bbox": {"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"}, "selection_area_id": str(area_id), "feature_count": 1, "total_feature_count": 1, "limit": 100, "truncated": False, "geojson": {"type": "FeatureCollection", "features": []}, } def summarize_features(db, **kwargs): # noqa: ANN001 captured["summary"] = kwargs return { "metric_label": "Gebouwen", "metric_value": 1, "metric_unit": "objecten", "aggregation_method": "feature_count", "feature_count": 1, "is_estimate": False, } monkeypatch.setattr(dataset_routes.DatasetService, "get_dataset", lambda db, selected_id: dataset) monkeypatch.setattr(dataset_routes.VectorFeatureService, "select_features_by_bbox", select_features) monkeypatch.setattr(dataset_routes.VectorFeatureService, "summarize_features_by_bbox", summarize_features) response = dataset_routes.select_vector_features( project_id=project_id, dataset_id=dataset_id, payload=dataset_routes.VectorSelectionRequest( bbox=dataset_routes.VectorSelectionBBox(min_x=4.9, min_y=50.9, max_x=5.2, max_y=51.2), area_id=area_id, limit=100, ), db=_AreaSession(), ) assert str(response["data"]["selection_area_id"]) == str(area_id) assert to_shape(captured["select"]["selection_geometry"]).equals(box(5.0, 51.0, 5.2, 51.2)) assert captured["select"]["selection_area_id"] == area_id assert to_shape(captured["summary"]["selection_geometry"]).equals(box(5.0, 51.0, 5.2, 51.2)) assert captured["select"]["full_dataset_area"] is False def test_vector_select_route_uses_bbox_for_dataset_preclipped_to_selected_area(monkeypatch) -> None: from app.api.routes import datasets as dataset_routes project_id = uuid.uuid4() dataset_id = uuid.uuid4() area_id = uuid.uuid4() dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, dataset_type="vector", source="fixture", name="Preclipped population", source_metadata={ "geometry_clipped_to_area": True, "selection_aggregation": {"method": "feature_count"}, }, ) area = SimpleNamespace( id=area_id, project_id=project_id, geometry=from_shape(box(5.0, 51.0, 5.3, 51.3), srid=4326), ) captured: dict[str, dict[str, object]] = {} class _AreaSession: @staticmethod def get(model, selected_id): # noqa: ANN001 assert model is dataset_routes.Area assert selected_id == area_id return area def select_features(_db, **kwargs): # noqa: ANN001 captured["select"] = kwargs return { "selection_bbox": {"min_x": 4.9, "min_y": 51.1, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"}, "selection_area_id": str(area_id), "feature_count": 1, "total_feature_count": 1, "limit": 25, "truncated": False, "geojson": {"type": "FeatureCollection", "features": []}, } def summarize_features(_db, **kwargs): # noqa: ANN001 captured["summary"] = kwargs return { "metric_label": "Inwoners", "metric_value": 1, "metric_unit": "inwoners", "aggregation_method": "feature_count", "feature_count": 1, "is_estimate": False, } monkeypatch.setattr(dataset_routes.DatasetService, "get_dataset", lambda _db, _id: dataset) monkeypatch.setattr(dataset_routes.VectorFeatureService, "select_features_by_bbox", select_features) monkeypatch.setattr(dataset_routes.VectorFeatureService, "summarize_features_by_bbox", summarize_features) dataset_routes.select_vector_features( project_id=project_id, dataset_id=dataset_id, payload=dataset_routes.VectorSelectionRequest( bbox=dataset_routes.VectorSelectionBBox(min_x=4.9, min_y=51.1, max_x=5.2, max_y=51.2), area_id=area_id, limit=25, ), db=_AreaSession(), ) assert captured["select"]["selection_geometry"] is None assert captured["summary"]["selection_geometry"] is None assert captured["select"]["selection_area_id"] == area_id assert captured["select"]["full_dataset_area"] is False def test_vector_select_route_rejects_area_from_another_project(monkeypatch) -> None: from app.api.routes import datasets as dataset_routes project_id = uuid.uuid4() dataset_id = uuid.uuid4() area_id = uuid.uuid4() dataset = Dataset(id=dataset_id, project_id=project_id, dataset_type="vector", source="fixture", name="Vector") other_area = SimpleNamespace(id=area_id, project_id=uuid.uuid4(), geometry=object()) monkeypatch.setattr(dataset_routes.DatasetService, "get_dataset", lambda db, selected_id: dataset) try: dataset_routes.select_vector_features( project_id=project_id, dataset_id=dataset_id, payload=dataset_routes.VectorSelectionRequest( bbox=dataset_routes.VectorSelectionBBox(min_x=4.9, min_y=50.9, max_x=5.2, max_y=51.2), area_id=area_id, ), db=SimpleNamespace(get=lambda model, selected_id: other_area), ) except AppError as exc: assert exc.code == "AREA_NOT_FOUND" else: # pragma: no cover raise AssertionError("Expected AREA_NOT_FOUND") def test_frontend_exposes_map_bbox_selection_contracts() -> None: api_client = read_feature("datasets") map_workspace = read_map_workspace() geomap = (ROOT / "frontend" / "src" / "components" / "GeoMap.tsx").read_text(encoding="utf-8") app = read_feature("shell") extract_hook = read_feature("map_workspace") theme_hook = read_feature("map_workspace") assert "selectVectorFeatures" in api_client assert "DEFAULT_AREA_SELECTION_FILENAME" in map_workspace assert "Teken rechthoek" in map_workspace assert "Objecten in gebied ophalen" in map_workspace assert "Gebiedsdownload bewaren" in map_workspace assert "bboxSelectionMode" in geomap assert "selection-bbox" in geomap assert "selection-result" in geomap assert "useMapSelectionExtract" in app assert "area_id: areaId" in extract_hook assert "area_id: areaId" in theme_hook # The saved area selection triggers analysis with its own area id. # The saved area bbox feeds the analysis path, now through a resolved # bbox rather than being passed positionally. assert_wired(map_workspace, "selectedAreaBbox") assert_calls(map_workspace, "analyzeSelection", first_argument="bbox")