geointel/backend/tests/test_sprint218_regional_dhmv.py

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path
from typing import Any

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"


def load_operator():
    if str(SCRIPTS) not in sys.path:
        sys.path.insert(0, str(SCRIPTS))
    spec = importlib.util.spec_from_file_location(
        "test_provision_regional_dhmv",
        SCRIPTS / "provision_regional_dhmv.py",
    )
    module = importlib.util.module_from_spec(spec)
    assert spec and spec.loader
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


class FakeResponse:
    def __init__(self, payload: dict[str, Any]):
        self.payload = payload
        self.url = "http://test.local"

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict[str, Any]:
        return self.payload


class FakeSession:
    def __init__(self, module):
        self.module = module
        self.posts: list[tuple[str, dict[str, Any]]] = []
        self.gets: list[tuple[str, dict[str, Any]]] = []
        self.headers: dict[str, str] = {}

    def get(self, url: str, **kwargs):
        self.gets.append((url, kwargs.get("params") or {}))
        if url.endswith("/api/v1/projects"):
            return FakeResponse({"data": {"items": [{"id": "project-1", "name": "Kempen Regional Workbench"}]}})
        if url.endswith("/areas"):
            return FakeResponse(
                {
                    "data": {
                        "items": [
                            {
                                "id": "area-mol",
                                "name": "Gemeente Mol - officiele grens",
                                "geometry": {
                                    "type": "Polygon",
                                    "coordinates": [[
                                        [5.0, 51.0],
                                        [5.1, 51.0],
                                        [5.1, 51.1],
                                        [5.0, 51.1],
                                        [5.0, 51.0],
                                    ]],
                                },
                            }
                        ],
                        "total": 1,
                    }
                }
            )
        if url.endswith("/datasets/dhmv/products"):
            return FakeResponse({"data": {"items": [{"key": key} for key in self.module.PRODUCTS]}})
        raise AssertionError(url)

    def post(self, url: str, json: dict[str, Any], **_kwargs):
        self.posts.append((url, json))
        if url.endswith("/datasets/dhmv/acquire"):
            return FakeResponse(
                {
                    "data": {
                        "id": "job-1",
                        "status": "success",
                        "output_dataset_id": "dataset-1",
                        "result_json": {"reused": True},
                    }
                }
            )
        if url.endswith("/raster/terrain/select"):
            return FakeResponse(
                {
                    "data": {
                        "resolution_m": 5.0,
                        "sample_count": 100,
                        "coverage_ratio": 1.0,
                        "unsupported_metrics": ["water_depth_m", "water_volume_m3"],
                        "summary": {"metrics": [{"metric_key": "elevation_mean_m", "metric_value": 24.5}]},
                    }
                }
            )
        raise AssertionError(url)


def test_regional_dhmv_operator_resolves_products_and_members() -> None:
    module = load_operator()

    products = module.requested_products("dtm_1m,dsm_1m", set(module.PRODUCTS))
    members = module.requested_members("Mol,13008", module.KEMPEN_TRANSPORT_REGION_SCOPE.members)

    assert products == ["dtm_1m", "dsm_1m"]
    assert [member.nis_code for member in members] == ["13025", "13008"]


def test_regional_dhmv_operator_rejects_unknown_scope_inputs() -> None:
    module = load_operator()

    with pytest.raises(RuntimeError, match="Unsupported DHMV"):
        module.requested_products("custom", set(module.PRODUCTS))
    with pytest.raises(RuntimeError, match="Unknown scope members"):
        module.requested_members("Atlantis", module.KEMPEN_TRANSPORT_REGION_SCOPE.members)


def test_regional_dhmv_operator_dry_run_uses_canonical_registry(monkeypatch, capsys) -> None:
    module = load_operator()
    fake_session = FakeSession(module)
    monkeypatch.setattr(module.requests, "Session", lambda: fake_session)

    result = module.main(["--members", "Mol", "--products", "dtm_1m", "--dry-run"])
    output = capsys.readouterr().out

    assert result == 0
    assert '"status": "dry_run"' in output
    assert '"planned_acquisitions": 1' in output
    assert fake_session.posts == []
    assert any(params.get("limit") == 200 for url, params in fake_session.gets if url.endswith("/areas"))


def test_regional_dhmv_operator_calls_acquisition_and_selection(monkeypatch, capsys) -> None:
    module = load_operator()
    fake_session = FakeSession(module)
    monkeypatch.setattr(module.requests, "Session", lambda: fake_session)

    result = module.main(["--members", "Mol", "--products", "dtm_1m"])
    output = capsys.readouterr().out

    assert result == 0
    assert '"completed_count": 1' in output
    assert '"status": "completed_item"' in output
    assert len(fake_session.posts) == 2
    acquisition_payload = fake_session.posts[0][1]
    assert fake_session.posts[0][0].endswith("/datasets/dhmv/acquire")
    assert acquisition_payload["area_id"] == "area-mol"
    assert acquisition_payload["product_key"] == "dtm_1m"
    assert acquisition_payload["bbox"]["crs"] == "EPSG:4326"


def test_regional_dhmv_operator_rejects_missing_water_limitations(monkeypatch) -> None:
    module = load_operator()
    fake_session = FakeSession(module)
    original_post = fake_session.post

    def post_without_limitations(url: str, json: dict[str, Any], **kwargs):
        response = original_post(url, json, **kwargs)
        if url.endswith("/raster/terrain/select"):
            response.payload["data"]["unsupported_metrics"] = []
        return response

    fake_session.post = post_without_limitations
    monkeypatch.setattr(module.requests, "Session", lambda: fake_session)

    assert module.main(["--members", "Mol", "--products", "dtm_1m"]) == 2


def test_regional_dhmv_operator_is_packaged() -> None:
    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")
    docs = (ROOT / "scripts" / "README.md").read_text(encoding="utf-8")

    assert "py_compile scripts/provision_regional_dhmv.py" in readiness
    assert "COPY scripts/provision_regional_dhmv.py" in dockerfile
    assert "provision_regional_dhmv.py" in docs