195 lines
7.1 KiB
Python
195 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.models import Area, Dataset, Export
|
|
from app.schemas.export import ExportCreateResponse
|
|
from app.services.export_service import ExportService
|
|
from app.services.storage_service import StorageService
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
|
|
|
|
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_geojson_export_persists_handoff_artifact(tmp_path, monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
export_path = tmp_path / "exports" / "selection.geojson"
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="candidate.geojson",
|
|
dataset_type="vector",
|
|
source="fixture",
|
|
status="ready",
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): 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",
|
|
"geometry": {"type": "Point", "coordinates": [5.0, 51.0]},
|
|
"properties": {"vector_feature_id": "vf-1"},
|
|
}
|
|
],
|
|
},
|
|
}
|
|
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(export_path))
|
|
monkeypatch.setattr(VectorFeatureService, "select_features_by_bbox", lambda *_args, **_kwargs: selection_payload)
|
|
|
|
response = ExportService.export_vector_selection_geojson(db, dataset_id, selection_bbox, limit=250, name="selected-buildings")
|
|
|
|
exports = [item for item in db.added if isinstance(item, Export)]
|
|
assert len(exports) == 1
|
|
assert response.export_id == exports[0].id
|
|
assert response.export_type == "vector_selection_geojson"
|
|
assert response.metadata_json["source"] == "vector_selection"
|
|
assert response.metadata_json["dataset_id"] == str(dataset_id)
|
|
assert response.metadata_json["selection_bbox"] == selection_bbox
|
|
assert response.metadata_json["feature_count"] == 1
|
|
assert response.metadata_json["source_table"] == "vector_features"
|
|
assert json.loads(export_path.read_text(encoding="utf-8"))["features"][0]["properties"]["vector_feature_id"] == "vf-1"
|
|
|
|
|
|
def test_vector_selection_geojson_export_endpoint_returns_canonical_envelope(monkeypatch) -> None:
|
|
export_id = uuid4()
|
|
dataset_id = uuid4()
|
|
area_id = uuid4()
|
|
expected_bbox = {"min_x": 4.9, "min_y": 50.9, "max_x": 5.2, "max_y": 51.2, "crs": "EPSG:4326"}
|
|
captured: dict = {}
|
|
|
|
def fake_export(*_args, **kwargs):
|
|
captured.update(kwargs)
|
|
return ExportCreateResponse(
|
|
export_id=export_id,
|
|
path="storage/exports/demo-selection.geojson",
|
|
status="ready",
|
|
export_type="vector_selection_geojson",
|
|
metadata_json={"source": "vector_selection", "selection_bbox": expected_bbox},
|
|
)
|
|
|
|
monkeypatch.setattr(ExportService, "export_vector_selection_geojson", fake_export)
|
|
|
|
response = TestClient(app).post(
|
|
"/api/v1/exports/geojson",
|
|
json={
|
|
"dataset_id": str(dataset_id),
|
|
"area_id": str(area_id),
|
|
"export_kind": "vector_selection",
|
|
"bbox": expected_bbox,
|
|
"limit": 250,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert set(payload) == {"data"}
|
|
assert payload["data"]["export_id"] == str(export_id)
|
|
assert payload["data"]["export_type"] == "vector_selection_geojson"
|
|
assert payload["data"]["metadata_json"]["selection_bbox"] == expected_bbox
|
|
assert captured["area_id"] == area_id
|
|
|
|
|
|
def test_vector_selection_export_uses_exact_area_scope_when_requested(tmp_path, monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
area_id = uuid4()
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="regional-buildings.geojson",
|
|
dataset_type="vector",
|
|
source="fixture",
|
|
status="ready",
|
|
)
|
|
area_geometry = object()
|
|
area = SimpleNamespace(id=area_id, project_id=project_id, name="Gemeente Mol", geometry=area_geometry)
|
|
db = FakeSession({(Dataset, dataset_id): dataset, (Area, area_id): area})
|
|
selection_bbox = {"min_x": 5.0, "min_y": 51.1, "max_x": 5.2, "max_y": 51.3, "crs": "EPSG:4326"}
|
|
captured: dict = {}
|
|
selection_payload = {
|
|
"selection_bbox": selection_bbox,
|
|
"selection_area_id": str(area_id),
|
|
"feature_count": 0,
|
|
"limit": 250,
|
|
"truncated": False,
|
|
"geojson": {"type": "FeatureCollection", "features": []},
|
|
}
|
|
|
|
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(tmp_path / "selection.geojson"))
|
|
monkeypatch.setattr(VectorFeatureService, "can_use_full_area_fast_path", lambda *_args: True)
|
|
|
|
def fake_select(*_args, **kwargs):
|
|
captured.update(kwargs)
|
|
return selection_payload
|
|
|
|
monkeypatch.setattr(VectorFeatureService, "select_features_by_bbox", fake_select)
|
|
|
|
response = ExportService.export_vector_selection_geojson(
|
|
db,
|
|
dataset_id,
|
|
selection_bbox,
|
|
area_id=area_id,
|
|
)
|
|
|
|
assert captured["selection_geometry"] is area_geometry
|
|
assert captured["selection_area_id"] == area_id
|
|
assert captured["full_dataset_area"] is True
|
|
assert response.metadata_json["selection_area_id"] == str(area_id)
|
|
|
|
|
|
def test_frontend_exposes_map_selection_export_action() -> None:
|
|
types = (ROOT / "frontend" / "src" / "types.ts").read_text(encoding="utf-8")
|
|
exports_api = (ROOT / "frontend" / "src" / "services" / "api" / "exports.ts").read_text(encoding="utf-8")
|
|
export_hook = (ROOT / "frontend" / "src" / "hooks" / "useExportWorkflow.ts").read_text(encoding="utf-8")
|
|
map_workspace = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
app = (ROOT / "frontend" / "src" / "App.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "'vector_selection'" in types
|
|
assert "bbox?: VectorSelectionBBox" in exports_api
|
|
assert "area_id?: string" in exports_api
|
|
assert "exportMapSelectionGeoJson" in export_hook
|
|
assert "vector_selection" in export_hook
|
|
assert "Save area export" in map_workspace
|
|
assert "onExportMapSelection" in map_workspace
|
|
assert "selectionExportError" in map_workspace
|
|
assert "onExportMapSelection={exportMapSelectionGeoJson}" in app
|