550 lines
21 KiB
Python
550 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pyproj import Transformer
|
|
from rasterio.io import MemoryFile
|
|
from rasterio.transform import from_origin
|
|
from shapely.geometry import 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 Dataset, Job, Project
|
|
from app.schemas.flood_hazard import (
|
|
FloodHazardAcquireRequest,
|
|
FloodHazardPartitionSelectionRequest,
|
|
FloodHazardSelectionRequest,
|
|
)
|
|
from app.schemas.assistant import AssistantQueryRequest
|
|
from app.services.geo_assistant_service import GeoAssistantService
|
|
from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService
|
|
from app.services.flood_hazard_analysis_service import FloodHazardAnalysisService
|
|
|
|
|
|
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
|
|
|
|
def all(self):
|
|
return self.result if isinstance(self.result, list) else []
|
|
|
|
|
|
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)
|
|
|
|
|
|
def flood_payload(*, product_key: str = "pluviaal_current_t100", side_m: float = 100.0) -> FloodHazardAcquireRequest:
|
|
transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(200_000, 210_000)
|
|
max_x, max_y = transformer.transform(200_000 + side_m, 210_000 + side_m)
|
|
return FloodHazardAcquireRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
|
|
product_key=product_key,
|
|
resolution_m=5.0,
|
|
force_refresh=True,
|
|
)
|
|
|
|
|
|
def depth_tiff(*, normalized_metres: bool = False) -> bytes:
|
|
values = np.zeros((20, 20), dtype="float32")
|
|
values[:, :10] = 1.0 if normalized_metres else 100.0
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(200_000, 210_100, 5.0, 5.0),
|
|
nodata=-9999.0 if normalized_metres else 0.0,
|
|
) as output:
|
|
if normalized_metres:
|
|
values[:, 10:] = -9999.0
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def edge_depth_tiff(*, left: float, top: float, x_resolution: float, y_resolution: float = 5.0) -> bytes:
|
|
values = np.full((20, 20), 100.0, dtype="float32")
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(left, top, x_resolution, y_resolution),
|
|
nodata=0.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def normalized_depth_tiff(*, left: float, top: float, value: float) -> bytes:
|
|
values = np.full((20, 20), value, dtype="float32")
|
|
with MemoryFile() as memory:
|
|
with memory.open(
|
|
driver="GTiff",
|
|
width=20,
|
|
height=20,
|
|
count=1,
|
|
dtype="float32",
|
|
crs="EPSG:31370",
|
|
transform=from_origin(left, top, 5.0, 5.0),
|
|
nodata=-9999.0,
|
|
) as output:
|
|
output.write(values, 1)
|
|
return memory.read()
|
|
|
|
|
|
def test_flood_hazard_registry_is_complete_and_semantically_honest() -> None:
|
|
products = FloodHazardAcquisitionService.list_products()
|
|
|
|
assert len(products) == 12
|
|
assert {item["mechanism"] for item in products} == {"pluviaal", "fluviaal"}
|
|
assert {item["climate_context"] for item in products} == {"huidig klimaat", "klimaatprojectie 2050"}
|
|
assert {item["return_period_years"] for item in products} == {10, 100, 1000}
|
|
assert all(item["coverage_id"].startswith("Overstromingsgevaarkaarten-") for item in products)
|
|
assert all(item["source_value_unit"] == "cm" and item["normalized_value_unit"] == "m" for item in products)
|
|
assert all("geen bathymetrie" in item["limitation_message"] for item in products)
|
|
|
|
|
|
def test_flood_hazard_request_is_bounded_and_rejects_arbitrary_products() -> None:
|
|
settings = Settings(_env_file=None)
|
|
prepared = FloodHazardAcquisitionService._prepared_request(flood_payload(), settings)
|
|
product = prepared["product"]
|
|
url = FloodHazardAcquisitionService._wcs_request_url(
|
|
settings,
|
|
product,
|
|
tuple(prepared["bbox_epsg31370"]),
|
|
prepared["resolution_m"],
|
|
)
|
|
|
|
assert "VERSION=1.1.0" in url
|
|
assert "IDENTIFIER=Overstromingsgevaarkaarten-PLUVIAAL%3Awaterdiepte_PLU_noCC_T100" in url
|
|
assert "GRIDOFFSETS=5%2C-5" in url
|
|
assert prepared["width"] * prepared["height"] <= settings.flood_hazard_max_pixels
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._prepared_request(flood_payload(product_key="custom"), settings)
|
|
assert exc_info.value.code == "FLOOD_HAZARD_PRODUCT_NOT_SUPPORTED"
|
|
|
|
|
|
def test_flood_hazard_tiles_stay_below_the_observed_vmm_coverage_limit() -> None:
|
|
prepared = FloodHazardAcquisitionService._prepared_request(flood_payload(side_m=15_000), Settings(_env_file=None))
|
|
tiles = FloodHazardAcquisitionService._tile_bounds(prepared)
|
|
|
|
assert 9 <= len(tiles) <= 16
|
|
assert all((max_x - min_x) <= 5_000 for min_x, _min_y, max_x, _max_y in tiles)
|
|
assert all((max_y - min_y) <= 5_000 for _min_x, min_y, _max_x, max_y in tiles)
|
|
assert all(
|
|
((max_x - min_x) / prepared["resolution_m"]) * ((max_y - min_y) / prepared["resolution_m"])
|
|
<= 1_000_000
|
|
for min_x, min_y, max_x, max_y in tiles
|
|
)
|
|
|
|
|
|
def test_flood_hazard_mosaic_harmonizes_only_bounded_wcs_edge_grid_rounding() -> None:
|
|
regular = edge_depth_tiff(left=200_000, top=210_100, x_resolution=5.0)
|
|
rounded_edge = edge_depth_tiff(
|
|
left=200_100,
|
|
top=210_100,
|
|
x_resolution=4.76555,
|
|
y_resolution=5.0008,
|
|
)
|
|
diagnostics: dict[str, object] = {}
|
|
|
|
mosaic = FloodHazardAcquisitionService._mosaic_geotiffs(
|
|
[regular, rounded_edge],
|
|
expected_resolution_m=5.0,
|
|
diagnostics=diagnostics,
|
|
)
|
|
|
|
with MemoryFile(mosaic) as memory, memory.open() as dataset:
|
|
assert dataset.res == pytest.approx((5.0, 5.0))
|
|
assert diagnostics["harmonized_tile_indexes"] == [1]
|
|
assert diagnostics["harmonization_method"] == "rasterio_merge_target_resolution"
|
|
|
|
unsafe_edge = edge_depth_tiff(left=200_100, top=210_100, x_resolution=4.5)
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._mosaic_geotiffs(
|
|
[regular, unsafe_edge],
|
|
expected_resolution_m=5.0,
|
|
)
|
|
assert exc_info.value.code == "FLOOD_HAZARD_TILE_MISMATCH"
|
|
assert exc_info.value.details["invalid_resolution_tiles"] == [
|
|
{"tile_index": 1, "resolution": [4.5, 5.0]}
|
|
]
|
|
|
|
|
|
def test_flood_hazard_xml_provider_error_is_exposed_without_losing_the_canonical_error() -> None:
|
|
response = b"""<?xml version="1.0"?>
|
|
<ExceptionReport xmlns="http://www.opengis.net/ows/1.1">
|
|
<Exception exceptionCode="NoApplicableCode">
|
|
<ExceptionText>This request is trying to generate too much data</ExceptionText>
|
|
</Exception>
|
|
</ExceptionReport>"""
|
|
|
|
with pytest.raises(AppError) as exc_info:
|
|
FloodHazardAcquisitionService._extract_geotiff(response, "application/xml")
|
|
|
|
assert exc_info.value.code == "FLOOD_HAZARD_PROVIDER_INVALID_RESPONSE"
|
|
assert exc_info.value.details["provider_exception"] == "This request is trying to generate too much data"
|
|
|
|
|
|
def test_flood_hazard_normalization_converts_centimetres_and_clips_zero_values() -> None:
|
|
payload = flood_payload()
|
|
prepared = FloodHazardAcquisitionService._prepared_request(payload, Settings(_env_file=None))
|
|
scope = box(
|
|
payload.bbox.min_x,
|
|
payload.bbox.min_y,
|
|
payload.bbox.max_x,
|
|
payload.bbox.max_y,
|
|
)
|
|
|
|
normalized, validation = FloodHazardAcquisitionService._normalize_raster(depth_tiff(), scope, prepared)
|
|
|
|
assert validation["inundated_pixel_count"] == 200
|
|
assert validation["minimum_depth_m"] == pytest.approx(1.0)
|
|
assert validation["maximum_depth_m"] == pytest.approx(1.0)
|
|
with MemoryFile(normalized) as memory, memory.open() as dataset:
|
|
values = dataset.read(1, masked=True)
|
|
assert dataset.crs.to_epsg() == 31370
|
|
assert dataset.nodata == -9999.0
|
|
assert values.count() == 200
|
|
assert float(values.mean()) == pytest.approx(1.0)
|
|
|
|
|
|
def test_flood_hazard_analysis_reports_scenario_metrics_without_claiming_waterbody_volume(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "flood.tif"
|
|
path.write_bytes(depth_tiff(normalized_metres=True))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="flood.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={"product_key": "pluviaal_current_t100", "normalized_value_unit": "m"},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
result = FloodHazardAnalysisService.analyze(
|
|
db,
|
|
project_id,
|
|
dataset_id,
|
|
FloodHazardSelectionRequest(bbox=flood_payload().bbox),
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]}
|
|
|
|
assert result["inundated_cell_count"] == 200
|
|
assert result["inundated_fraction"] == pytest.approx(0.5)
|
|
assert metrics["modelled_inundated_area_ha"]["metric_value"] == pytest.approx(0.5)
|
|
assert metrics["modelled_depth_mean_m"]["metric_value"] == pytest.approx(1.0)
|
|
assert metrics["modelled_max_depth_area_integral_m3"]["metric_value"] == pytest.approx(5000.0)
|
|
assert "concurrent_flood_volume_m3" in result["unsupported_metrics"]
|
|
assert "geen gelijktijdig" in result["limitation_message"]
|
|
|
|
|
|
def test_partitioned_flood_analysis_is_exact_across_municipality_boundaries(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
transformer = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
|
|
min_x, min_y = transformer.transform(200_000, 210_000)
|
|
middle_x, _ = transformer.transform(200_100, 210_000)
|
|
max_x, max_y = transformer.transform(200_200, 210_100)
|
|
paths = [tmp_path / "left-flood.tif", tmp_path / "right-flood.tif"]
|
|
paths[0].write_bytes(normalized_depth_tiff(left=200_000, top=210_100, value=1.0))
|
|
paths[1].write_bytes(normalized_depth_tiff(left=200_100, top=210_100, value=2.0))
|
|
datasets = [
|
|
Dataset(
|
|
id=uuid4(),
|
|
project_id=project_id,
|
|
area_id=uuid4(),
|
|
name=path.name,
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={
|
|
"product_key": "pluviaal_current_t100",
|
|
"normalized_value_unit": "m",
|
|
"bbox_epsg4326": [left, min_y, right, max_y],
|
|
},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
for path, left, right in (
|
|
(paths[0], min_x, middle_x),
|
|
(paths[1], middle_x, max_x),
|
|
)
|
|
]
|
|
db = FakeSession(query_result=datasets)
|
|
payload = FloodHazardPartitionSelectionRequest(
|
|
bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
|
|
product_key="pluviaal_current_t100",
|
|
)
|
|
|
|
result = FloodHazardAnalysisService.analyze_partitions(
|
|
db,
|
|
project_id,
|
|
payload,
|
|
settings=Settings(_env_file=None),
|
|
)
|
|
metrics = {item["metric_key"]: item["metric_value"] for item in result["summary"]["metrics"]}
|
|
|
|
assert result["partition_count"] == 2
|
|
assert set(result["dataset_ids"]) == {str(dataset.id) for dataset in datasets}
|
|
assert result["inundated_cell_count"] >= 790
|
|
assert result["inundated_fraction"] == pytest.approx(1.0)
|
|
assert metrics["modelled_depth_mean_m"] == pytest.approx(1.5, abs=0.01)
|
|
assert metrics["modelled_depth_p90_m"] == 2.0
|
|
assert metrics["modelled_inundated_area_ha"] == pytest.approx(2.0, abs=0.03)
|
|
assert "2 persistente gemeentelijke rasterpartities" in result["limitation_message"]
|
|
|
|
|
|
def test_flood_hazard_renderer_returns_transparent_png(tmp_path) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
path = tmp_path / "flood.tif"
|
|
path.write_bytes(depth_tiff(normalized_metres=True))
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="flood.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={"product_key": "pluviaal_current_t100", "normalized_value_unit": "m"},
|
|
status="ready",
|
|
storage_path=str(path),
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
|
|
assert FloodHazardAnalysisService.render_png(db, project_id, dataset_id).startswith(b"\x89PNG\r\n\x1a\n")
|
|
|
|
|
|
def test_flood_hazard_api_uses_canonical_envelopes(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
output_dataset_id = uuid4()
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")})
|
|
monkeypatch.setattr(
|
|
FloodHazardAcquisitionService,
|
|
"acquire",
|
|
lambda *_args, **_kwargs: {"output_dataset_id": str(output_dataset_id), "provider": "vmm_flood_hazard", "reused": False},
|
|
)
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze",
|
|
lambda *_args, **_kwargs: {
|
|
"dataset_id": str(output_dataset_id),
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"selection_bbox": flood_payload().bbox.model_dump(),
|
|
"selected_cell_count": 10,
|
|
"inundated_cell_count": 4,
|
|
"inundated_fraction": 0.4,
|
|
"resolution_m": 5.0,
|
|
"summary": {
|
|
"metric_label": "Overstroomde oppervlakte",
|
|
"metric_value": 0.01,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "positive_depth_area",
|
|
"primary_metric_key": "inundated_area_ha",
|
|
"metrics": [],
|
|
},
|
|
"unsupported_metrics": ["permanent_water_volume_m3"],
|
|
"limitation_message": "Scenario depth is not bathymetry.",
|
|
"generated_at": "2026-07-18T00:00:00Z",
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze_partitions",
|
|
lambda *_args, **_kwargs: {
|
|
"dataset_id": str(output_dataset_id),
|
|
"dataset_ids": [str(output_dataset_id)],
|
|
"partition_count": 1,
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"selection_bbox": flood_payload().bbox.model_dump(),
|
|
"selected_cell_count": 10,
|
|
"inundated_cell_count": 4,
|
|
"inundated_fraction": 0.4,
|
|
"resolution_m": 5.0,
|
|
"summary": {
|
|
"metric_label": "Overstroomde oppervlakte",
|
|
"metric_value": 0.01,
|
|
"metric_unit": "ha",
|
|
"aggregation_method": "positive_depth_area",
|
|
"primary_metric_key": "inundated_area_ha",
|
|
"metrics": [],
|
|
},
|
|
"unsupported_metrics": ["permanent_water_volume_m3"],
|
|
"limitation_message": "Scenario depth is not bathymetry.",
|
|
"generated_at": "2026-07-18T00:00:00Z",
|
|
},
|
|
)
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
try:
|
|
client = TestClient(app)
|
|
products = client.get(f"/api/v1/projects/{project_id}/datasets/flood-hazard/products")
|
|
acquisition = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/flood-hazard/acquire",
|
|
json=flood_payload().model_dump(mode="json"),
|
|
)
|
|
selection = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/{output_dataset_id}/raster/flood-hazard/select",
|
|
json={"bbox": flood_payload().bbox.model_dump()},
|
|
)
|
|
regional_selection = client.post(
|
|
f"/api/v1/projects/{project_id}/datasets/raster/flood-hazard/select",
|
|
json={"bbox": flood_payload().bbox.model_dump(), "product_key": "pluviaal_current_t100"},
|
|
)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
assert products.status_code == 200 and set(products.json()) == {"data"}
|
|
assert products.json()["data"]["total"] == 12
|
|
assert acquisition.status_code == 200 and set(acquisition.json()) == {"data"}
|
|
assert acquisition.json()["data"]["job_type"] == "raster.flood_hazard.acquire"
|
|
assert selection.status_code == 200 and set(selection.json()) == {"data"}
|
|
assert regional_selection.status_code == 200 and set(regional_selection.json()) == {"data"}
|
|
assert regional_selection.json()["data"]["partition_count"] == 1
|
|
assert any(isinstance(item, Job) for item in db.added)
|
|
|
|
|
|
def test_geo_assistant_receives_scenario_bound_flood_metrics(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
dataset = Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name="pluvial.tif",
|
|
dataset_type="raster",
|
|
source="VMM",
|
|
source_name=FloodHazardAcquisitionService.PROVIDER,
|
|
source_metadata={
|
|
"product_key": "pluviaal_current_t100",
|
|
"product_display_name": "Pluviaal - huidig klimaat - middelgrote kans (T100)",
|
|
},
|
|
status="ready",
|
|
)
|
|
db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}, query_result=[dataset])
|
|
monkeypatch.setattr(
|
|
FloodHazardAnalysisService,
|
|
"analyze",
|
|
lambda *_args, **_kwargs: {
|
|
"product_key": "pluviaal_current_t100",
|
|
"mechanism": "pluviaal",
|
|
"climate_context": "huidig klimaat",
|
|
"probability_class": "middelgrote kans",
|
|
"return_period_years": 100,
|
|
"summary": {
|
|
"metrics": [
|
|
{
|
|
"metric_label": "Gemodelleerd overstroomd oppervlak",
|
|
"metric_value": 12.5,
|
|
"metric_unit": "ha",
|
|
}
|
|
]
|
|
},
|
|
"limitation_message": "Geen werkelijk of gelijktijdig volume.",
|
|
},
|
|
)
|
|
payload = AssistantQueryRequest(question="Wat is het overstromingsgevaar?", bbox=flood_payload().bbox)
|
|
|
|
context, metrics, _series, dataset_ids, warnings, _scope = GeoAssistantService(Settings(_env_file=None))._build_context(
|
|
db,
|
|
project_id=project_id,
|
|
payload=payload,
|
|
)
|
|
|
|
assert warnings == []
|
|
assert dataset_ids == [dataset_id]
|
|
assert metrics[0].theme == "flood_hazard"
|
|
assert "T100" in metrics[0].label
|
|
assert context["rules"]["water_volume_available"] is False
|
|
assert context["rules"]["flood_hazard_scenarios_available"] is True
|
|
assert context["rules"]["flood_depth_area_integral_is_concurrent_volume"] is False
|
|
|
|
|
|
def test_flood_hazard_runtime_contract_is_packaged() -> None:
|
|
for path in (
|
|
ROOT / ".env.example",
|
|
ROOT / "docker-compose.yml",
|
|
ROOT / "docker-compose.unraid.yml",
|
|
ROOT / "deploy" / "unraid" / "geointel.env.example",
|
|
):
|
|
content = path.read_text(encoding="utf-8")
|
|
assert "FLOOD_HAZARD_ENABLED" in content
|
|
assert "FLOOD_HAZARD_WCS_URL" in content
|
|
assert "FLOOD_HAZARD_MAX_PIXELS" in content
|
|
|
|
operator = (ROOT / "scripts" / "provision_mol_flood_hazards.py").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
|
|
dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
frontend = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
assert "/datasets/flood-hazard/acquire" in operator
|
|
assert "/raster/flood-hazard/select" in operator
|
|
assert "concurrent_flood_volume_m3" in operator
|
|
assert "py_compile scripts/provision_mol_flood_hazards.py" in readiness
|
|
assert "COPY scripts/provision_mol_flood_hazards.py" in dockerfile
|
|
assert "Overstromingsscenario" in frontend
|
|
assert "floodHazardImageUrl" in frontend
|
|
assert "dataset.source_name === 'vmm_flood_hazard'" in frontend
|
|
assert "return theme.id === 'flood_hazard'" in frontend
|