from __future__ import annotations from pathlib import Path from types import SimpleNamespace from uuid import uuid4 import numpy as np import pytest import rasterio from fastapi.testclient import TestClient from pyproj import Transformer from rasterio.io import MemoryFile from rasterio.transform import from_origin 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.thematic_raster import ThematicRasterAcquireRequest, ThematicRasterSelectionRequest from app.schemas.assistant import AssistantQueryRequest from app.services.geo_assistant_service import GeoAssistantService from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService from app.services.thematic_raster_analysis_service import ThematicRasterAnalysisService from app.services.dataset_service import DatasetService 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) class FakeResponse: def __init__(self, content: bytes): self.content = content self.headers = {"Content-Type": "image/tiff", "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 payload(product_key: str = "space_occupation_2025", *, side_m: float = 1000.0) -> ThematicRasterAcquireRequest: 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 ThematicRasterAcquireRequest( bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"}, product_key=product_key, force_refresh=True, ) def raster_bytes(values: np.ndarray, resolution: float, *, nodata: float = -9999.0) -> bytes: with MemoryFile() as memory: with memory.open( driver="GTiff", width=values.shape[1], height=values.shape[0], count=1, dtype=str(values.dtype), crs="EPSG:31370", transform=from_origin(200_000, 210_000 + values.shape[0] * resolution, resolution, resolution), nodata=nodata, ) as output: output.write(values, 1) return memory.read() def test_registry_contains_five_governed_non_water_policy_products() -> None: products = ThematicRasterAcquisitionService.list_products() assert [item["key"] for item in products] == [ "space_occupation_2025", "open_space_2022", "population_density_2019", "node_value_2022", "service_level_2022", ] assert {item["theme"] for item in products} == {"space_occupation", "open_space", "population", "accessibility", "services"} assert {item["native_resolution_m"] for item in products} == {10.0, 100.0} assert all(item["coverage_id"].startswith(("lu:", "ni:")) for item in products) assert all(item["source_crs"] == "EPSG:31370" for item in products) assert all(item["attribution"] and item["license_note"] and item["limitation_message"] for item in products) def test_request_is_bounded_allowlisted_and_uses_native_wcs_resolution() -> None: settings = Settings(_env_file=None) prepared = ThematicRasterAcquisitionService._prepared_request(payload("population_density_2019"), settings) url = ThematicRasterAcquisitionService._wcs_request_url(settings, prepared["product"], tuple(prepared["bbox_epsg31370"])) assert "VERSION=1.0.0" in url assert "COVERAGE=ni%3Ani_inw_ha_vlaa_2019" in url assert "RESX=100" in url and "RESY=100" in url assert prepared["width"] * prepared["height"] <= settings.thematic_raster_max_pixels with pytest.raises(AppError) as exc_info: ThematicRasterAcquisitionService._prepared_request(payload("arbitrary_remote_layer"), settings) assert exc_info.value.code == "THEMATIC_RASTER_PRODUCT_NOT_SUPPORTED" def test_binary_and_normalized_products_fail_closed_on_invalid_values() -> None: binary = ThematicRasterAcquisitionService._product("space_occupation_2025") score = ThematicRasterAcquisitionService._product("service_level_2022") with pytest.raises(AppError, match="Binary"): ThematicRasterAcquisitionService._validate_values(np.asarray([0.0, 2.0]), binary) with pytest.raises(AppError, match="0-1"): ThematicRasterAcquisitionService._validate_values(np.asarray([0.2, 1.2]), score) def test_acquisition_clips_validates_and_delegates_persistence(monkeypatch) -> None: project_id, output_dataset_id = uuid4(), uuid4() db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}) content = raster_bytes(np.ones((100, 100), dtype="float32"), 10.0) captured: dict = {} def fake_import(_db, **kwargs): captured.update(kwargs) return SimpleNamespace(id=output_dataset_id) monkeypatch.setattr(DatasetService, "import_raster_bytes", fake_import) result = ThematicRasterAcquisitionService.acquire( db, project_id, payload(side_m=1000.0), settings=Settings(_env_file=None), opener=lambda *_args, **_kwargs: FakeResponse(content), ) assert result["output_dataset_id"] == str(output_dataset_id) assert captured["source_name"] == ThematicRasterAcquisitionService.PROVIDER assert captured["source_metadata"]["product_key"] == "space_occupation_2025" assert captured["source_metadata"]["metric_kind"] == "binary_area" assert captured["source_metadata"]["valid_pixel_count"] > 9_800 assert captured["provenance_metadata"]["acquisition"] == "explicit_bounded_tiled_wcs_coverage" assert len(captured["provenance_metadata"]["normalized_sha256"]) == 64 def test_binary_area_analysis_returns_hectares_and_share(tmp_path) -> None: project_id, dataset_id = uuid4(), uuid4() values = np.zeros((10, 10), dtype="float32") values[:, :5] = 1.0 path = tmp_path / "space.tif" path.write_bytes(raster_bytes(values, 10.0)) dataset = Dataset( id=dataset_id, project_id=project_id, name="space.tif", dataset_type="raster", source="official", source_name=ThematicRasterAcquisitionService.PROVIDER, source_metadata={"product_key": "space_occupation_2025", "coverage_id": "lu:lu_ruibes_vlaa_2025_v3"}, status="ready", storage_path=str(path), ) db = FakeSession({(Dataset, dataset_id): dataset}) result = ThematicRasterAnalysisService.analyze( db, project_id, dataset_id, ThematicRasterSelectionRequest(bbox=payload(side_m=100.0).bbox), ) metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]} assert result["valid_cell_count"] == 100 assert metrics["space_occupation_area_ha"]["metric_value"] == pytest.approx(0.5) assert metrics["space_occupation_share_pct"]["metric_value"] == pytest.approx(50.0) assert "object_count" in result["unsupported_metrics"] def test_population_analysis_sums_one_hectare_density_cells_without_claiming_current_counts(tmp_path) -> None: project_id, dataset_id = uuid4(), uuid4() values = np.asarray([[10.0, 20.0], [30.0, 40.0]], dtype="float32") path = tmp_path / "population.tif" path.write_bytes(raster_bytes(values, 100.0)) dataset = Dataset( id=dataset_id, project_id=project_id, name="population.tif", dataset_type="raster", source="official", source_name=ThematicRasterAcquisitionService.PROVIDER, source_metadata={"product_key": "population_density_2019", "coverage_id": "ni:ni_inw_ha_vlaa_2019"}, status="ready", storage_path=str(path), ) db = FakeSession({(Dataset, dataset_id): dataset}) result = ThematicRasterAnalysisService.analyze( db, project_id, dataset_id, ThematicRasterSelectionRequest(bbox=payload("population_density_2019", side_m=200.0).bbox), ) metrics = {item["metric_key"]: item for item in result["summary"]["metrics"]} assert metrics["estimated_inhabitants"]["metric_value"] == pytest.approx(100.0) assert metrics["population_density_mean_per_ha"]["metric_value"] == pytest.approx(25.0) assert metrics["estimated_inhabitants"]["is_estimate"] is True assert "current_population" in result["unsupported_metrics"] def test_assistant_context_receives_persisted_thematic_metrics(tmp_path) -> None: project_id, dataset_id = uuid4(), uuid4() values = np.asarray([[10.0, 20.0], [30.0, 40.0]], dtype="float32") path = tmp_path / "assistant-population.tif" path.write_bytes(raster_bytes(values, 100.0)) project = Project(id=project_id, name="Mol", region="Mol") dataset = Dataset( id=dataset_id, project_id=project_id, name="population.tif", dataset_type="raster", source="official", source_name=ThematicRasterAcquisitionService.PROVIDER, source_metadata={"product_key": "population_density_2019", "coverage_id": "ni:ni_inw_ha_vlaa_2019"}, status="ready", storage_path=str(path), ) db = FakeSession({(Project, project_id): project, (Dataset, dataset_id): dataset}, query_result=[dataset]) context, metrics, _series, dataset_ids, _warnings, _scope = GeoAssistantService(Settings(_env_file=None))._build_context( db, project_id=project_id, payload=AssistantQueryRequest(question="Hoeveel inwoners?", bbox=payload("population_density_2019", side_m=200.0).bbox), ) assert any(metric.theme == "population" and metric.label.startswith("Geraamd aantal") for metric in metrics) assert dataset_id in dataset_ids assert context["rules"]["thematic_policy_rasters_available"] is True def test_index_renderer_returns_browser_png(tmp_path) -> None: project_id, dataset_id = uuid4(), uuid4() values = np.linspace(0.1, 4.0, 100, dtype="float32").reshape((10, 10)) path = tmp_path / "node.tif" path.write_bytes(raster_bytes(values, 100.0)) dataset = Dataset( id=dataset_id, project_id=project_id, name="node.tif", dataset_type="raster", source="official", source_name=ThematicRasterAcquisitionService.PROVIDER, source_metadata={ "product_key": "node_value_2022", "coverage_id": "lu:lu_knptw_ha_2022_v3", "render_min_value": 0.1, "render_max_value": 4.0, }, status="ready", storage_path=str(path), ) db = FakeSession({(Dataset, dataset_id): dataset}) assert ThematicRasterAnalysisService.render_png(db, project_id, dataset_id).startswith(b"\x89PNG\r\n\x1a\n") def test_api_uses_canonical_envelopes(monkeypatch) -> None: project_id, dataset_id = uuid4(), uuid4() db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}) monkeypatch.setattr( ThematicRasterAcquisitionService, "acquire", lambda *_args, **_kwargs: {"output_dataset_id": str(dataset_id), "provider": ThematicRasterAcquisitionService.PROVIDER}, ) monkeypatch.setattr( ThematicRasterAnalysisService, "analyze", lambda *_args, **_kwargs: {"dataset_id": str(dataset_id), "theme": "population", "summary": {"metric_value": 10.0}}, ) app.dependency_overrides[get_db] = lambda: db try: client = TestClient(app) products = client.get(f"/api/v1/projects/{project_id}/datasets/thematic-raster/products") acquisition = client.post( f"/api/v1/projects/{project_id}/datasets/thematic-raster/acquire", json=payload().model_dump(mode="json"), ) selection = client.post( f"/api/v1/projects/{project_id}/datasets/{dataset_id}/raster/thematic/select", json={"bbox": payload().bbox.model_dump()}, ) finally: app.dependency_overrides.clear() assert products.status_code == 200 and set(products.json()) == {"data"} assert products.json()["data"]["total"] == 5 assert acquisition.status_code == 200 and set(acquisition.json()) == {"data"} assert acquisition.json()["data"]["job_type"] == "raster.thematic.acquire" assert selection.status_code == 200 and selection.json()["data"]["theme"] == "population" assert any(isinstance(item, Job) for item in db.added)