from __future__ import annotations import json from pathlib import Path from uuid import uuid4 from fastapi.testclient import TestClient from app.main import app from app.models import Dataset from app.schemas.dataset import DatasetCreateResponse from app.services.storage_service import StorageService from app.services.vector_feature_service import VectorFeatureService from app.services.vector_operations_service import VectorOperationsService from tests.frontend_contract import read_feature ROOT = Path(__file__).resolve().parents[2] class FakeSession: def __init__(self, rows): self.rows = rows self.added = [] def get(self, model, row_id): row = self.rows.get((model, row_id)) if row is not None: return row for item in self.added: if isinstance(item, model) and item.id == row_id: return item return None def add(self, row): self.added.append(row) def commit(self): return None def refresh(self, row): return row def test_vector_selection_derive_persists_queryable_derived_dataset(tmp_path, monkeypatch) -> None: project_id = uuid4() dataset_id = uuid4() output_path = tmp_path / "selection-derived.geojson" source_dataset = Dataset( id=dataset_id, project_id=project_id, area_id=None, name="candidate.geojson", dataset_type="vector", source="fixture", dataset_role="source", source_name="fixture", storage_path=str(tmp_path / "candidate.geojson"), status="ready", ) db = FakeSession({(Dataset, dataset_id): source_dataset}) selection_bbox = {"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"} selection_payload = { "selection_bbox": selection_bbox, "feature_count": 1, "limit": 250, "truncated": False, "geojson": { "type": "FeatureCollection", "features": [ { "type": "Feature", "id": "source-row-1", "geometry": {"type": "Point", "coordinates": [5.0, 51.0]}, "properties": { "vector_feature_id": "source-row-1", "dataset_id": str(dataset_id), "source_feature_id": "pred-1", "feature_class": "building", "confidence": 0.8, }, } ], }, } persisted_features = [] def _persist_dataset_file(project_id: str, dataset_id: str, dataset_type: str, original_filename: str, content: bytes, content_type: str | None): output_path.write_bytes(content) return { "original_filename": original_filename, "stored_filename": output_path.name, "content_type": content_type or "application/geo+json", "size_bytes": len(content), "checksum_sha256": "selection-checksum", "storage_path": str(output_path), } def _persist_geojson_features(db, dataset_id, payload, feature_class=None, *, commit=True): persisted_features.append({"dataset_id": dataset_id, "payload": payload, "feature_class": feature_class, "commit": commit}) return [] monkeypatch.setattr(StorageService, "persist_dataset_file", _persist_dataset_file) monkeypatch.setattr(VectorFeatureService, "select_features_by_bbox", lambda *_args, **_kwargs: selection_payload) monkeypatch.setattr(VectorFeatureService, "persist_geojson_features", _persist_geojson_features) response = VectorOperationsService.derive_selection_dataset( db=db, dataset_id=dataset_id, bbox=selection_bbox, limit=250, output_name="selected-buildings", ) derived = [item for item in db.added if isinstance(item, Dataset)][0] assert response.id == derived.id assert response.project_id == project_id assert response.dataset_role == "derived" assert response.source == "operation:selection" assert response.source_name == "map_selection" assert response.derived_from_dataset_id == dataset_id assert response.feature_count == 1 assert response.metadata_json["selection_bbox"] == selection_bbox assert response.metadata_json["source_feature_count"] == 1 assert response.provenance_metadata["source_dataset_id"] == str(dataset_id) assert response.provenance_metadata["source_table"] == "vector_features" assert persisted_features[0]["dataset_id"] == derived.id assert persisted_features[0]["commit"] is False derived_payload = json.loads(output_path.read_text(encoding="utf-8")) props = derived_payload["features"][0]["properties"] assert props["source_vector_feature_id"] == "source-row-1" assert props["source_dataset_id"] == str(dataset_id) assert "vector_feature_id" not in props def test_vector_selection_derive_rejects_empty_selection(monkeypatch, tmp_path) -> None: dataset_id = uuid4() source_dataset = Dataset( id=dataset_id, project_id=uuid4(), name="candidate.geojson", dataset_type="vector", source="fixture", storage_path=str(tmp_path / "candidate.geojson"), status="ready", ) db = FakeSession({(Dataset, dataset_id): source_dataset}) monkeypatch.setattr( VectorFeatureService, "select_features_by_bbox", lambda *_args, **_kwargs: { "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": 250, "truncated": False, "geojson": {"type": "FeatureCollection", "features": []}, }, ) try: VectorOperationsService.derive_selection_dataset( db=db, 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=250, output_name="empty-selection", ) except Exception as exc: assert getattr(exc, "code") == "VECTOR_OPERATION_EMPTY_RESULT" else: raise AssertionError("Empty selection should not create a derived dataset") def test_vector_selection_derive_endpoint_returns_canonical_dataset_envelope(monkeypatch) -> None: project_id = uuid4() dataset_id = uuid4() derived_id = uuid4() bbox = {"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"} monkeypatch.setattr( "app.api.routes.datasets.DatasetService.get_dataset", lambda _db, requested_id: Dataset(id=requested_id, project_id=project_id, name="source.geojson", dataset_type="vector", source="fixture"), ) monkeypatch.setattr( "app.api.routes.datasets.VectorOperationsService.derive_selection_dataset", lambda *_args, **_kwargs: DatasetCreateResponse( id=derived_id, name="selected-buildings.geojson", dataset_type="vector", source="operation:selection", dataset_role="derived", source_name="map_selection", project_id=project_id, status="ready", derived_from_dataset_id=dataset_id, feature_count=1, metadata_json={"selection_bbox": bbox, "source_feature_count": 1}, ), ) response = TestClient(app).post( f"/api/v1/projects/{project_id}/datasets/{dataset_id}/vector/select/derive", json={"bbox": bbox, "limit": 250, "output_name": "selected-buildings"}, ) assert response.status_code == 201 payload = response.json() assert set(payload) == {"data"} assert payload["data"]["id"] == str(derived_id) assert payload["data"]["dataset_role"] == "derived" assert payload["data"]["source_name"] == "map_selection" assert payload["data"]["derived_from_dataset_id"] == str(dataset_id) def test_frontend_exposes_map_selection_derive_action() -> None: types = (ROOT / "frontend" / "src" / "types.ts").read_text(encoding="utf-8") datasets_api = read_feature("datasets") app = read_feature("shell") map_workspace = read_feature("map_workspace") assert "VectorSelectionDeriveRequest" in types assert "deriveVectorSelection" in datasets_api assert "deriveMapSelectionDataset" in app assert "Als resultaatlaag bewaren" in map_workspace assert "selectionDatasetError" in map_workspace