fix(platform): govern geospatial analysis and raster handoffs

backend/tests/test_area_crs_semantics.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from geoalchemy2.shape import from_shape, to_shape
+from pyproj import Transformer
+import pytest
+from shapely.geometry import Polygon, mapping
+from shapely.ops import transform
+
+from app.core.errors import AppError
+from app.models import Area, Project
+from app.schemas.area import AreaCreate, AreaUpdate
+from app.services.area_service import AreaService
+from app.utils.geometry import area_m2, normalize_area_to_epsg4326
+
+
+class FakeSession:
+    def __init__(self, objects=None) -> None:
+        self.objects = objects or {}
+        self.added = []
+        self.commits = 0
+        self.refreshes = []
+
+    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 refresh(self, item) -> None:
+        self.refreshes.append(item)
+
+
+def _wgs84_polygon(offset: float = 0.0) -> Polygon:
+    return Polygon(
+        [
+            (5.00 + offset, 51.00),
+            (5.01 + offset, 51.00),
+            (5.01 + offset, 51.01),
+            (5.00 + offset, 51.01),
+            (5.00 + offset, 51.00),
+        ]
+    )
+
+
+def _to_lambert(geometry: Polygon) -> Polygon:
+    transformer = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
+    return transform(transformer.transform, geometry)
+
+
+def test_create_area_transforms_declared_lambert_geometry_before_storage() -> None:
+    project_id = uuid4()
+    db = FakeSession({(Project, project_id): Project(id=project_id, name="Belgium")})
+    source = _wgs84_polygon()
+
+    area = AreaService.create_area(
+        db,
+        project_id,
+        AreaCreate(name="Lambert AOI", geometry=mapping(_to_lambert(source)), crs="EPSG:31370"),
+    )
+
+    stored = to_shape(area.geometry)
+    assert stored.bounds == pytest.approx(source.bounds, abs=1e-7)
+    assert area.original_crs == "EPSG:31370"
+    assert area.area_m2 == pytest.approx(area_m2(normalize_area_to_epsg4326(mapping(source), "EPSG:4326")[0]))
+    assert area.area_m2 and area.area_m2 > 0
+    assert to_shape(area.bbox).bounds == pytest.approx(source.bounds, abs=1e-7)
+
+
+def test_patch_area_replaces_geometry_and_recomputes_all_spatial_fields() -> None:
+    area_id = uuid4()
+    project_id = uuid4()
+    original = _wgs84_polygon()
+    normalized, _ = normalize_area_to_epsg4326(mapping(original), "EPSG:4326")
+    area = Area(
+        id=area_id,
+        project_id=project_id,
+        name="Original",
+        geometry=from_shape(normalized, srid=4326),
+        bbox=from_shape(normalized.envelope, srid=4326),
+        original_crs="EPSG:4326",
+        area_m2=area_m2(normalized),
+    )
+    db = FakeSession({(Area, area_id): area})
+    replacement = _wgs84_polygon(offset=0.05)
+
+    updated = AreaService.update_area(
+        db,
+        area_id,
+        AreaUpdate(
+            name="Replacement",
+            geometry=mapping(_to_lambert(replacement)),
+            crs="EPSG:31370",
+        ),
+    )
+
+    assert updated.name == "Replacement"
+    assert updated.original_crs == "EPSG:31370"
+    assert to_shape(updated.geometry).bounds == pytest.approx(replacement.bounds, abs=1e-7)
+    assert to_shape(updated.bbox).bounds == pytest.approx(replacement.bounds, abs=1e-7)
+    assert updated.area_m2 and updated.area_m2 > 0
+    assert db.commits == 1
+
+
+@pytest.mark.parametrize(
+    ("geometry", "crs", "message_fragment"),
+    [
+        (mapping(_wgs84_polygon()), "EPSG:not-real", "unknown or invalid"),
+        (mapping(_wgs84_polygon()), "EPSG:4979", "exactly two spatial axes"),
+        (
+            {
+                "type": "Polygon",
+                "coordinates": [[[5.0, 51.0], [float("nan"), 51.0], [5.1, 51.1], [5.0, 51.0]]],
+            },
+            "EPSG:4326",
+            "finite",
+        ),
+        (mapping(Polygon([(10.0, 51.0), (10.1, 51.0), (10.1, 51.1), (10.0, 51.0)])), "EPSG:4326", "workbench domain"),
+        (
+            {
+                "type": "Polygon",
+                "coordinates": [[[5.0, 51.0], [5.1, 51.1], [5.1, 51.0], [5.0, 51.1], [5.0, 51.0]]],
+            },
+            "EPSG:4326",
+            "invalid",
+        ),
+        ({"type": "Point", "coordinates": [5.0, 51.0]}, "EPSG:4326", "Polygon or MultiPolygon"),
+    ],
+)
+def test_create_area_rejects_invalid_crs_nonfinite_and_out_of_domain_geometry(
+    geometry: dict,
+    crs: str,
+    message_fragment: str,
+) -> None:
+    project_id = uuid4()
+    db = FakeSession({(Project, project_id): Project(id=project_id, name="Belgium")})
+
+    with pytest.raises(AppError) as exc_info:
+        AreaService.create_area(db, project_id, AreaCreate(name="Invalid", geometry=geometry, crs=crs))
+
+    assert exc_info.value.code == "INVALID_GEOMETRY"
+    assert message_fragment in exc_info.value.message
+    assert db.commits == 0
+
+
+def test_patch_area_rejects_crs_without_replacement_geometry() -> None:
+    area_id = uuid4()
+    area = Area(id=area_id, project_id=uuid4(), name="AOI", original_crs="EPSG:4326")
+    db = FakeSession({(Area, area_id): area})
+
+    with pytest.raises(AppError) as exc_info:
+        AreaService.update_area(db, area_id, AreaUpdate(crs="EPSG:31370"))
+
+    assert exc_info.value.code == "INVALID_AREA_CRS_UPDATE"
+    assert db.commits == 0