430 lines
17 KiB
Python
430 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import pytest
|
|
import rasterio
|
|
from fastapi.testclient import TestClient
|
|
from geoalchemy2.shape import from_shape
|
|
from pyproj import Transformer
|
|
from rasterio.io import MemoryFile
|
|
from rasterio.transform import from_origin
|
|
from shapely.geometry import MultiPolygon, box
|
|
|
|
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, DatasetVersion, Job, Project
|
|
from app.schemas.dhmv import DhmvAcquireRequest, TerrainSelectionRequest
|
|
from app.services.dhmv_acquisition_service import DhmvAcquisitionService
|
|
from app.services.terrain_analysis_service import TerrainAnalysisService
|
|
|
|
|
|
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 first(self):
|
|
return self.result
|
|
|
|
|
|
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 FakeResponse:
|
|
def __init__(self, content: bytes, content_type: str):
|
|
self.content = content
|
|
self.headers = {"Content-Type": content_type, "Content-Length": str(len(content))}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return None
|
|
|
|
def read(self, limit: int):
|
|
return self.content[:limit]
|
|
|
|
|
|
def lambert_bbox_payload(*, side_m: float = 100.0, product_key: str = "dtm_1m", area_id=None) -> DhmvAcquireRequest:
|
|
west, south = 200_000.0, 210_000.0
|
|
transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(west, south)
|
|
max_x, max_y = transformer.transform(west + side_m, south + side_m)
|
|
return DhmvAcquireRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
|
|
area_id=area_id,
|
|
product_key=product_key,
|
|
resolution_m=5.0,
|
|
force_refresh=True,
|
|
)
|
|
|
|
|
|
def elevation_tiff(*, left: float, top: float, width: int, height: int, resolution: float = 5.0) -> bytes:
|
|
rows, columns = np.indices((height, width))
|
|
values = (20.0 + columns * 0.5 + rows * 1.0).astype("float32")
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=width,
|
|
height=height,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(left, top, resolution, resolution),
|
|
nodata=-9999.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def multipart_tiff(content: bytes) -> tuple[bytes, str]:
|
|
boundary = "wcs-test"
|
|
payload = (
|
|
f"--{boundary}\r\nContent-Type: text/xml\r\nContent-ID: GML-Part\r\n\r\n<coverage/>\r\n"
|
|
f"--{boundary}\r\nContent-Type: image/tiff\r\nContent-ID: coverage.tif\r\n\r\n"
|
|
).encode() + content + f"\r\n--{boundary}--\r\n".encode()
|
|
return payload, f'multipart/mixed; boundary="{boundary}"'
|
|
|
|
|
|
def test_dhmv_registry_is_governed_and_semantically_explicit() -> None:
|
|
products = DhmvAcquisitionService.list_products()
|
|
|
|
assert [item["key"] for item in products] == ["dtm_1m", "dsm_1m"]
|
|
assert {item["coverage_id"] for item in products} == {"DHMVII_DTM_1m", "DHMVII_DSM_1m"}
|
|
assert all(item["native_resolution_m"] == 1.0 for item in products)
|
|
assert all(item["source_crs"] == "EPSG:31370" for item in products)
|
|
assert all("TAW" in item["vertical_reference"] for item in products)
|
|
assert all(item["acquisition_period"] == "2013-2015" for item in products)
|
|
assert "waterdiepte" in products[0]["limitation_message"]
|
|
|
|
|
|
def test_dhmv_request_uses_bounded_official_wcs_scaling() -> None:
|
|
prepared = DhmvAcquisitionService._prepared_request(lambert_bbox_payload(), Settings(_env_file=None))
|
|
|
|
assert prepared["coverage_id"] == "DHMVII_DTM_1m"
|
|
assert prepared["params"]["SCALEFACTOR"] == "5"
|
|
assert prepared["params"]["SUBSET"][0].startswith("x(")
|
|
assert prepared["params"]["SUBSET"][1].startswith("y(")
|
|
assert "geo.api.vlaanderen.be%2FDHMV" not in prepared["request_url"]
|
|
assert prepared["request_url"].startswith("https://geo.api.vlaanderen.be/DHMV/wcs?")
|
|
assert prepared["width"] * prepared["height"] <= 12_000_000
|
|
assert len(prepared["request_hash"]) == 64
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
DhmvAcquisitionService._prepared_request(
|
|
lambert_bbox_payload(product_key="arbitrary"),
|
|
Settings(_env_file=None),
|
|
)
|
|
assert exc_info.value.code == "DHMV_PRODUCT_NOT_SUPPORTED"
|
|
|
|
|
|
def test_dhmv_request_rejects_unsafe_size_and_resolution() -> None:
|
|
with pytest.raises(AppError) as exc_info:
|
|
DhmvAcquisitionService._prepared_request(lambert_bbox_payload(side_m=5.0), Settings(_env_file=None))
|
|
assert exc_info.value.code == "DHMV_SELECTION_TOO_SMALL"
|
|
|
|
payload = lambert_bbox_payload()
|
|
payload.resolution_m = 0.5
|
|
with pytest.raises(Exception):
|
|
DhmvAcquireRequest.model_validate(payload.model_dump())
|
|
|
|
|
|
def test_dhmv_large_scope_is_bounded_into_mosaicable_wcs_tiles() -> None:
|
|
prepared = DhmvAcquisitionService._prepared_request(
|
|
lambert_bbox_payload(side_m=15_000.0),
|
|
Settings(_env_file=None),
|
|
)
|
|
|
|
tile_bounds = DhmvAcquisitionService._tile_bounds(prepared)
|
|
|
|
assert len(tile_bounds) == 4
|
|
assert all(bounds[2] - bounds[0] <= 10_000.0 for bounds in tile_bounds)
|
|
assert all(bounds[3] - bounds[1] <= 10_000.0 for bounds in tile_bounds)
|
|
|
|
left = elevation_tiff(left=200_000, top=210_100, width=20, height=20)
|
|
right = elevation_tiff(left=200_100, top=210_100, width=20, height=20)
|
|
mosaic = DhmvAcquisitionService._mosaic_geotiffs([left, right])
|
|
with MemoryFile(mosaic) as memory, memory.open() as dataset:
|
|
assert dataset.crs.to_epsg() == 31370
|
|
assert dataset.res == pytest.approx((5.0, 5.0))
|
|
assert dataset.width == 40
|
|
assert dataset.height == 20
|
|
assert dataset.nodata == -9999.0
|
|
|
|
|
|
def test_dhmv_multipart_geotiff_is_extracted_and_invalid_response_fails_closed() -> None:
|
|
tiff = elevation_tiff(left=200_000, top=210_100, width=20, height=20)
|
|
multipart, content_type = multipart_tiff(tiff)
|
|
|
|
assert DhmvAcquisitionService._extract_geotiff(multipart, content_type) == tiff
|
|
with pytest.raises(AppError) as exc_info:
|
|
DhmvAcquisitionService._extract_geotiff(b"<ServiceException/>", "text/xml")
|
|
assert exc_info.value.code == "DHMV_PROVIDER_INVALID_RESPONSE"
|
|
|
|
|
|
def test_dhmv_acquisition_clips_validates_and_persists_via_dataset_service(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
area_id = uuid4()
|
|
payload = lambert_bbox_payload(area_id=area_id)
|
|
area_geometry = MultiPolygon([box(payload.bbox.min_x, payload.bbox.min_y, payload.bbox.max_x, payload.bbox.max_y)])
|
|
db = FakeSession(
|
|
{
|
|
(Project, project_id): Project(id=project_id, name="Mol"),
|
|
(Area, area_id): Area(
|
|
id=area_id,
|
|
project_id=project_id,
|
|
name="Gemeente Mol",
|
|
geometry=from_shape(area_geometry, srid=4326),
|
|
),
|
|
}
|
|
)
|
|
settings = Settings(_env_file=None, storage_root=str(tmp_path), dhmv_resolution_m=5.0)
|
|
prepared = DhmvAcquisitionService._prepared_request(payload, settings)
|
|
tiff = elevation_tiff(
|
|
left=prepared["bbox_epsg31370"][0],
|
|
top=prepared["bbox_epsg31370"][3],
|
|
width=prepared["width"],
|
|
height=prepared["height"],
|
|
)
|
|
multipart, content_type = multipart_tiff(tiff)
|
|
|
|
result = DhmvAcquisitionService.acquire(
|
|
db,
|
|
project_id,
|
|
payload,
|
|
settings=settings,
|
|
opener=lambda *_args, **_kwargs: FakeResponse(multipart, content_type),
|
|
)
|
|
|
|
dataset = next(item for item in db.added if isinstance(item, Dataset))
|
|
version = next(item for item in db.added if isinstance(item, DatasetVersion))
|
|
assert result["output_dataset_id"] == str(dataset.id)
|
|
assert dataset.source_name == "digitaal_vlaanderen_dhmv"
|
|
assert dataset.area_id == area_id
|
|
assert dataset.dataset_type == "raster"
|
|
assert dataset.crs == "EPSG:31370"
|
|
assert dataset.checksum_sha256 == version.checksum_sha256
|
|
assert dataset.source_metadata["surface_model"] == "terrain"
|
|
assert dataset.source_metadata["native_resolution_m"] == 1.0
|
|
assert dataset.source_metadata["analysis_resolution_m"] == 5.0
|
|
assert dataset.source_metadata["nodata_value"] == -9999.0
|
|
assert dataset.provenance_metadata["water_depth_available"] is False
|
|
assert dataset.provenance_metadata["water_volume_available"] is False
|
|
assert len(dataset.provenance_metadata["response_sha256"]) == 64
|
|
with rasterio.open(dataset.storage_path) as stored:
|
|
assert stored.crs.to_epsg() == 31370
|
|
assert stored.count == 1
|
|
assert stored.nodata == -9999.0
|
|
assert stored.res == pytest.approx((5.0, 5.0))
|
|
|
|
|
|
def test_terrain_analysis_returns_governed_elevation_relief_and_slope(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "terrain.tif"
|
|
path.write_bytes(elevation_tiff(left=200_000, top=210_100, width=20, height=20))
|
|
to_wgs84 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = to_wgs84.transform(200_000, 210_000)
|
|
max_x, max_y = to_wgs84.transform(200_100, 210_100)
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="dhmvii_terrain_5m.tif",
|
|
dataset_type="raster",
|
|
source="official WCS",
|
|
source_name="digitaal_vlaanderen_dhmv",
|
|
source_metadata={
|
|
"product_key": "dtm_1m",
|
|
"surface_model": "terrain",
|
|
"vertical_reference": "TAW (Tweede Algemene Waterpassing)",
|
|
},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
payload = TerrainSelectionRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"}
|
|
)
|
|
|
|
result = TerrainAnalysisService.analyze(db, project_id, dataset_id, payload, settings=Settings(_env_file=None))
|
|
metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]}
|
|
|
|
assert result["sample_count"] > 300
|
|
assert result["coverage_ratio"] > 0.99
|
|
assert result["resolution_m"] == 5.0
|
|
assert result["summary"]["metric_unit"] == "m TAW"
|
|
assert metrics["relief_m"]["metric_value"] > 20
|
|
assert metrics["slope_mean_deg"]["metric_value"] == pytest.approx(12.6044, abs=0.01)
|
|
assert result["unsupported_metrics"] == ["water_depth_m", "water_volume_m3"]
|
|
assert "Waterdiepte" in result["limitation_message"]
|
|
|
|
|
|
def test_terrain_analysis_rejects_non_dhmv_raster(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "other.tif"
|
|
path.write_bytes(elevation_tiff(left=200_000, top=210_100, width=20, height=20))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="other.tif",
|
|
dataset_type="raster",
|
|
source="manual",
|
|
source_name="manual",
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
TerrainAnalysisService.analyze(db, project_id, dataset_id, TerrainSelectionRequest(bbox=lambert_bbox_payload().bbox))
|
|
assert exc_info.value.code == "INVALID_TERRAIN_DATASET"
|
|
|
|
|
|
def test_terrain_renderer_returns_browser_png(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "terrain.tif"
|
|
path.write_bytes(elevation_tiff(left=200_000, top=210_100, width=20, height=20))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="terrain.tif",
|
|
dataset_type="raster",
|
|
source="official",
|
|
source_name="digitaal_vlaanderen_dhmv",
|
|
source_metadata={"product_key": "dtm_1m", "surface_model": "terrain"},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
assert TerrainAnalysisService.render_png(db, project_id, dataset_id).startswith(b"\x89PNG\r\n\x1a\n")
|
|
|
|
|
|
def test_dhmv_endpoints_use_canonical_envelopes(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
output_dataset_id = uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")})
|
|
monkeypatch.setattr(
|
|
DhmvAcquisitionService,
|
|
"acquire",
|
|
lambda *_args, **_kwargs: {
|
|
"output_dataset_id": str(output_dataset_id),
|
|
"provider": "digitaal_vlaanderen_dhmv",
|
|
"reused": False,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
TerrainAnalysisService,
|
|
"analyze",
|
|
lambda *_args, **_kwargs: {
|
|
"dataset_id": str(output_dataset_id),
|
|
"sample_count": 100,
|
|
"summary": {"metric_value": 25.0, "metric_unit": "m TAW", "metrics": []},
|
|
"unsupported_metrics": ["water_depth_m", "water_volume_m3"],
|
|
},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
try:
|
|
products = TestClient(app).get(f"/api/v1/projects/{project_id}/datasets/dhmv/products")
|
|
acquisition = TestClient(app).post(
|
|
f"/api/v1/projects/{project_id}/datasets/dhmv/acquire",
|
|
json=lambert_bbox_payload().model_dump(mode="json"),
|
|
)
|
|
terrain = TestClient(app).post(
|
|
f"/api/v1/projects/{project_id}/datasets/{output_dataset_id}/raster/terrain/select",
|
|
json={"bbox": lambert_bbox_payload().bbox.model_dump()},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert products.status_code == 200
|
|
assert set(products.json()) == {"data"}
|
|
assert products.json()["data"]["total"] == 2
|
|
assert acquisition.status_code == 200
|
|
assert set(acquisition.json()) == {"data"}
|
|
assert acquisition.json()["data"]["job_type"] == "raster.dhmv.acquire"
|
|
assert acquisition.json()["data"]["output_dataset_id"] == str(output_dataset_id)
|
|
assert terrain.status_code == 200
|
|
assert set(terrain.json()) == {"data"}
|
|
assert terrain.json()["data"]["sample_count"] == 100
|
|
assert terrain.json()["data"]["unsupported_metrics"] == ["water_depth_m", "water_volume_m3"]
|
|
assert any(isinstance(item, Job) for item in db.added)
|
|
|
|
|
|
def test_frontend_and_runtime_expose_dhmv_workflow() -> None:
|
|
app_source = (ROOT / "frontend" / "src" / "App.tsx").read_text(encoding="utf-8")
|
|
map_source = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
hook_source = (ROOT / "frontend" / "src" / "hooks" / "useMapSelectionExtract.ts").read_text(encoding="utf-8")
|
|
service_source = (ROOT / "frontend" / "src" / "services" / "api" / "datasets.ts").read_text(encoding="utf-8")
|
|
|
|
assert "digitaal_vlaanderen_dhmv" in app_source
|
|
assert "Hoogte & reliëf" in map_source
|
|
assert "terrainImageUrl" in map_source
|
|
assert "selectTerrain" in hook_source
|
|
assert "/raster/terrain/select" in service_source
|
|
for path in (
|
|
ROOT / ".env.example",
|
|
ROOT / "docker-compose.yml",
|
|
ROOT / "docker-compose.unraid.yml",
|
|
ROOT / "deploy" / "unraid" / "run-dockerman-container.sh",
|
|
ROOT / "deploy" / "unraid" / "geointel-unraid-template.xml",
|
|
):
|
|
content = path.read_text(encoding="utf-8")
|
|
assert "DHMV_ENABLED" in content
|
|
assert "DHMV_RESOLUTION_M" in content
|
|
assert "DHMV_MAX_PIXELS" in content
|
|
|
|
|
|
def test_dhmv_operator_is_packaged_and_release_checked() -> None:
|
|
operator = (ROOT / "scripts" / "provision_mol_dhmv.py").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
|
|
backend_dockerfile = (ROOT / "backend" / "Dockerfile").read_text(encoding="utf-8")
|
|
all_in_one_dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
|
|
assert "/datasets/dhmv/acquire" in operator
|
|
assert "/raster/terrain/select" in operator
|
|
assert "water_depth_m" in operator
|
|
assert "py_compile scripts/provision_mol_dhmv.py" in readiness
|
|
assert "COPY . /app" in backend_dockerfile
|
|
assert "COPY scripts/provision_mol_dhmv.py /app/scripts/provision_mol_dhmv.py" in all_in_one_dockerfile
|