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, SourceRegistry, SourceSnapshot
from app.schemas.dhmv import DhmvAcquireRequest, TerrainPartitionSelectionRequest, TerrainSelectionRequest
from app.services.dhmv_acquisition_service import DhmvAcquisitionService
from app.services.terrain_analysis_service import TerrainAnalysisService
from tests.frontend_contract import read_map_workspace, read_feature


ROOT = Path(__file__).resolve().parents[2]


class FakeQuery:
    def __init__(self, results=None):
        self.results = list(results or [])

    def filter(self, *_args):
        return self

    def order_by(self, *_args):
        return self

    def first(self):
        return self.results[0] if self.results else None

    def one_or_none(self):
        return self.first()

    def all(self):
        return list(self.results)


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 flush(self):
        # Exercise the governed source/snapshot import path with database-like
        # primary-key assignment instead of silently falling back to legacy
        # fixture behavior.
        for row in self.added:
            if getattr(row, "id", None) is None:
                row.id = uuid4()

    def commit(self):
        return None

    def rollback(self):
        return None

    def refresh(self, row):
        return row

    def query(self, model):
        rows = [
            row
            for (row_model, _row_id), row in self.rows.items()
            if row_model is model and isinstance(row, model)
        ]
        rows.extend(row for row in self.added if isinstance(row, model))
        if isinstance(self.query_result, model):
            rows.append(self.query_result)
        elif isinstance(self.query_result, list):
            rows.extend(row for row in self.query_result if isinstance(row, model))
        return FakeQuery(rows)


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 constant_elevation_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 edge_elevation_tiff(*, left: float, top: float, x_resolution: float, y_resolution: float = 5.0) -> bytes:
    rows, columns = np.indices((20, 20))
    values = (20.0 + columns * 0.5 + rows).astype("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=-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_mosaic_harmonizes_only_bounded_wcs_edge_grid_rounding() -> None:
    regular = elevation_tiff(left=200_000, top=210_100, width=20, height=20)
    rounded_edge = edge_elevation_tiff(left=200_100, top=210_100, x_resolution=4.76555, y_resolution=5.0008)
    diagnostics: dict[str, object] = {}

    mosaic = DhmvAcquisitionService._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_elevation_tiff(left=200_100, top=210_100, x_resolution=4.5)
    with pytest.raises(AppError) as exc_info:
        DhmvAcquisitionService._mosaic_geotiffs([regular, unsafe_edge], expected_resolution_m=5.0)
    assert exc_info.value.code == "DHMV_TILE_MISMATCH"


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_fetch_sends_explicit_accept_header_required_by_official_wcs() -> None:
    observed_headers: dict[str, str | None] = {}

    def opener(request, **_kwargs):
        observed_headers["accept"] = request.get_header("Accept")
        observed_headers["user_agent"] = request.get_header("User-agent")
        return FakeResponse(b"II*\x00test", "image/tiff")

    content, content_type = DhmvAcquisitionService._fetch(
        "https://geo.api.vlaanderen.be/DHMV/wcs?bounded=true",
        Settings(_env_file=None),
        opener,
    )

    assert content == b"II*\x00test"
    assert content_type == "image/tiff"
    assert observed_headers == {
        "accept": "*/*",
        "user_agent": "GeoIntel/0.1 bounded-dhmv-acquisition",
    }


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))
    source = next(item for item in db.added if isinstance(item, SourceRegistry))
    snapshot = next(item for item in db.added if isinstance(item, SourceSnapshot))
    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
    assert dataset.source_registry_id == source.id
    assert dataset.source_snapshot_id == snapshot.id
    assert dataset.validation_status == "passed"
    assert dataset.provenance_status == "complete"
    assert dataset.lineage_status == "not_applicable"
    assert dataset.quarantine_status == "not_quarantined"
    assert snapshot.source_registry_id == source.id
    assert snapshot.checksum_sha256 == dataset.checksum_sha256
    assert snapshot.ingest_status == "ingested"
    assert snapshot.freshness_status == "current"
    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_partitioned_terrain_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-terrain.tif", tmp_path / "right-terrain.tif"]
    paths[0].write_bytes(constant_elevation_tiff(left=200_000, top=210_100, value=10.0))
    paths[1].write_bytes(constant_elevation_tiff(left=200_100, top=210_100, value=20.0))
    datasets = [
        Dataset(
            id=uuid4(),
            project_id=project_id,
            area_id=uuid4(),
            name=path.name,
            dataset_type="raster",
            source="official WCS",
            source_name="digitaal_vlaanderen_dhmv",
            source_metadata={
                "product_key": "dtm_1m",
                "surface_model": "terrain",
                "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 = TerrainPartitionSelectionRequest(
        bbox={"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"},
        product_key="dtm_1m",
    )

    result = TerrainAnalysisService.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["sample_count"] >= 790
    assert metrics["terrain_elevation_mean_m"] == pytest.approx(15.0, abs=0.1)
    assert metrics["terrain_elevation_min_m"] == 10.0
    assert metrics["terrain_elevation_max_m"] == 20.0
    assert metrics["terrain_elevation_p90_m"] == 20.0
    assert "2 persistente gemeentelijke rasterpartities" 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),
            "product_key": "dtm_1m",
            "surface_model": "terrain",
            "selection_bbox": lambert_bbox_payload().bbox.model_dump(),
            "sample_count": 100,
            "slope_sample_count": 81,
            "coverage_ratio": 1.0,
            "resolution_m": 5.0,
            "vertical_reference": "TAW",
            "summary": {
                "metric_label": "Gemiddelde terreinhoogte",
                "metric_value": 25.0,
                "metric_unit": "m TAW",
                "aggregation_method": "mean",
                "primary_metric_key": "terrain_elevation_mean_m",
                "metrics": [],
            },
            "unsupported_metrics": ["water_depth_m", "water_volume_m3"],
            "limitation_message": "Terrain height is not water depth.",
            "generated_at": "2026-07-18T00:00:00Z",
        },
    )
    monkeypatch.setattr(
        TerrainAnalysisService,
        "analyze_partitions",
        lambda *_args, **_kwargs: {
            "dataset_id": str(output_dataset_id),
            "dataset_ids": [str(output_dataset_id)],
            "partition_count": 1,
            "product_key": "dtm_1m",
            "surface_model": "terrain",
            "selection_bbox": lambert_bbox_payload().bbox.model_dump(),
            "sample_count": 100,
            "slope_sample_count": 81,
            "coverage_ratio": 1.0,
            "resolution_m": 5.0,
            "vertical_reference": "TAW",
            "summary": {
                "metric_label": "Gemiddelde terreinhoogte",
                "metric_value": 25.0,
                "metric_unit": "m TAW",
                "aggregation_method": "mean",
                "primary_metric_key": "terrain_elevation_mean_m",
                "metrics": [],
            },
            "unsupported_metrics": ["water_depth_m", "water_volume_m3"],
            "limitation_message": "Terrain height is not water depth.",
            "generated_at": "2026-07-18T00:00:00Z",
        },
    )
    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()},
        )
        regional_terrain = TestClient(app).post(
            f"/api/v1/projects/{project_id}/datasets/raster/terrain/select",
            json={"bbox": lambert_bbox_payload().bbox.model_dump(), "product_key": "dtm_1m"},
        )
    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 regional_terrain.status_code == 200
    assert set(regional_terrain.json()) == {"data"}
    assert regional_terrain.json()["data"]["partition_count"] == 1
    assert any(isinstance(item, Job) for item in db.added)


def test_frontend_and_runtime_expose_dhmv_workflow() -> None:
    capabilities_source = (
        ROOT / "frontend" / "src" / "lib" / "datasetCapabilities.ts"
    ).read_text(encoding="utf-8")
    map_source = read_map_workspace()
    hook_source = read_feature("map_workspace")
    service_source = read_feature("datasets")

    assert "digitaal_vlaanderen_dhmv" in capabilities_source
    assert "isMapRasterDataset" in capabilities_source
    assert "Hoogte & reliëf" in map_source
    assert "terrainImageUrl" in map_source
    assert "analysisMode === 'current' && activeTheme.id === 'buildings' && mapSelectionBbox" 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