geointel/backend/tests/test_sprint229_alz_release_management.py

from __future__ import annotations

import argparse
import importlib.util
import json
from pathlib import Path
import sys
import zipfile

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))


def load_script(name: str):
    path = SCRIPTS / name
    module_name = f"test_{path.stem}_sprint229"
    spec = importlib.util.spec_from_file_location(module_name, path)
    assert spec is not None
    assert spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module


OPERATOR = load_script("provision_agricultural_parcel_history.py")
MANAGER = load_script("manage_alz_agriculture_release.py")


def arguments(tmp_path: Path, **overrides) -> argparse.Namespace:
    values = {
        "action": "plan",
        "project_id": "00000000-0000-0000-0000-000000000001",
        "scope": "kempen-transport-region",
        "api_url": "http://127.0.0.1:8000/api/v1",
        "confirm_edition": None,
        "confirm_plan_sha256": None,
        "confirm_review_sha256": None,
        "approve": False,
        "reviewer": None,
        "review_note": "",
        "plan_path": None,
        "review_path": None,
        "output_root": tmp_path / "operator-evidence" / "agricultural-use-parcels",
        "evidence_root": tmp_path / "operator-evidence" / "alz-agriculture-refresh",
        "refresh_catalog": False,
        "request_timeout": 900,
        "api_timeout": 180,
        "import_timeout": 3600,
        "max_features": 250_000,
        "max_archive_mb": 250,
    }
    values.update(overrides)
    return argparse.Namespace(**values)


def catalog_item(
    *,
    remote: str = "2026-v3",
    local: str | None = "2025-definitive",
    catalog_hash: str = "a" * 64,
    checked_at: str = "2027-03-16T08:00:00Z",
) -> dict:
    year = int(remote[:4])
    published = "2027-03-15T00:00:00Z" if year == 2026 else "2026-05-13T00:00:00Z"
    return {
        "source_name": MANAGER.SOURCE_NAME,
        "status": "available",
        "reachable": True,
        "error_code": None,
        "remote_version": remote,
        "remote_published_at": published,
        "local_source_version": local,
        "message": f"Definitieve editie {remote}; actuele publicatie {year + 1}-v1 is voorlopig.",
        "matched_layers": ["definitive_archive", "current_snapshot"],
        "capabilities_sha256": catalog_hash,
        "metadata_identifier": "alz-agricultural-use-parcels",
        "metadata_url": "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen",
        "checked_at": checked_at,
    }


def decision(args: argparse.Namespace, *, remote: str = "2026-v3", local: str | None = "2025-definitive") -> dict:
    return MANAGER.fetch_release_decision_from_item(args, catalog_item(remote=remote, local=local))


def release_2026():
    return MANAGER.release_from_catalog_item(catalog_item())


def write_staged_artifacts(args: argparse.Namespace, release) -> dict:
    paths = OPERATOR.artifact_paths(
        args.output_root,
        args.scope,
        release.year,
        archive_url=release.archive_url,
    )
    paths["directory"].mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(paths["archive"], "w") as archive:
        archive.writestr(f"agpa_{release.year}.gpkg", b"official geopackage")
    paths["artifact"].write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
    codelist = {
        "year": release.year,
        "crop_entries": [{"code": "201", "title": "Mais", "group_title": "Mais"}],
        "code_title_conflicts": {},
    }
    paths["codelist"].write_text(json.dumps(codelist), encoding="utf-8")
    baseline_dir = args.output_root / args.scope / "2025"
    baseline_dir.mkdir(parents=True, exist_ok=True)
    (baseline_dir / "agricultural_use_parcels_2025_kempen-transport-region.manifest.json").write_text(
        json.dumps(
            {
                "year": 2025,
                "scope_key": args.scope,
                "feature_count": 120_000,
                "clipped_area_ha": 62_000.0,
            }
        ),
        encoding="utf-8",
    )
    manifest = {
        "schema_version": 1,
        "year": release.year,
        "scope_key": args.scope,
        "member_nis_codes": ["13025", "13003"],
        "source_url": release.archive_url,
        "source_crs": OPERATOR.SOURCE_CRS,
        "output_crs": OPERATOR.OUTPUT_CRS,
        "source_archive_sha256": OPERATOR.sha256_file(paths["archive"]),
        "source_archive_size_bytes": paths["archive"].stat().st_size,
        "source_feature_count": 180_000,
        "source_fields": sorted(OPERATOR.STABLE_REQUIRED_FIELDS),
        "crop_code_list_sha256": OPERATOR.sha256_file(paths["codelist"]),
        "artifact_sha256": OPERATOR.sha256_file(paths["artifact"]),
        "feature_count": 121_500,
        "clipped_feature_count": 800,
        "clipped_area_ha": 62_500.0,
    }
    paths["manifest"].write_text(json.dumps(manifest), encoding="utf-8")
    return paths


def staged_plan(args: argparse.Namespace, release, release_decision: dict) -> tuple[Path, dict]:
    write_staged_artifacts(args, release)
    result = {
        "status": "ok",
        "scope": args.scope,
        "years": [{"year": release.year, "status": "prepared", "feature_count": 121_500}],
    }
    plan = MANAGER.build_staged_plan(args, release_decision, release, result)
    path = MANAGER.default_plan_path(args, release.year)
    MANAGER.write_json(path, plan)
    return path, plan


def test_future_release_config_accepts_only_one_exact_official_edition() -> None:
    url = "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2026_2027-03-15_public.zip"
    release = OPERATOR.resolve_release_config(2026, archive_url=url)

    assert release.definitive_version == "2026-v3"
    assert release.archive_url == url
    future = OPERATOR.resolve_release_config(2027, archive_url=url.replace("2026", "2027"))
    assert future.definitive_version == "2027-v3"
    with pytest.raises(ValueError, match="official ALZ URL"):
        OPERATOR.resolve_release_config(2026, archive_url=url.replace("www.landbouwvlaanderen.be", "example.com"))
    with pytest.raises(ValueError, match="may not be overridden"):
        OPERATOR.resolve_release_config(2025, archive_url=url.replace("2026", "2025"))


def test_future_archive_requires_one_selected_year() -> None:
    url = "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2026_2027-03-15_public.zip"
    with pytest.raises(ValueError, match="exactly one"):
        OPERATOR.resolve_release_configs("2025,2026", archive_url=url)


def test_agriculture_workspace_pagination_is_complete_and_total_consistent() -> None:
    rows = [{"id": index} for index in range(401)]

    class Response:
        ok = True
        status_code = 200
        text = ""

        def __init__(self, payload: dict) -> None:
            self.payload = payload

        def json(self) -> dict:
            return {"data": self.payload}

    class Session:
        drift = False

        def get(self, _url: str, *, params: dict, timeout: int):
            assert timeout == 30
            offset = int(params["offset"])
            total = len(rows) + (1 if self.drift and offset else 0)
            return Response({"items": rows[offset : offset + int(params["limit"])], "total": total})

    assert OPERATOR.api_items(Session(), "http://backend/datasets", 30) == rows
    drifting = Session()
    drifting.drift = True
    with pytest.raises(RuntimeError, match="total changed"):
        OPERATOR.api_items(drifting, "http://backend/datasets", 30)


def test_archive_download_rejects_oversize_before_streaming(tmp_path: Path) -> None:
    url = "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2026_2027-03-15_public.zip"

    class Response:
        headers = {"content-length": "101"}

        def __init__(self) -> None:
            self.url = url

        def raise_for_status(self) -> None:
            return None

        def iter_content(self, *, chunk_size: int):
            raise AssertionError(f"streaming should not start: {chunk_size}")

    class Session:
        def get(self, *_args, **_kwargs):
            return Response()

    with pytest.raises(RuntimeError, match="configured"):
        OPERATOR.download_archive(Session(), url, tmp_path / "source.zip", timeout=30, max_bytes=100, force=True)


def test_archive_rejects_extracted_size_over_limit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    archive_path = tmp_path / "source.zip"
    with zipfile.ZipFile(archive_path, "w") as archive:
        archive.writestr("agpa_2026.gpkg", b"0123456789")
    monkeypatch.setattr(OPERATOR, "MAX_EXTRACTED_BYTES", 9)

    with pytest.raises(RuntimeError, match="extracted-size safety limit"):
        OPERATOR.validate_archive(archive_path)


@pytest.mark.parametrize(
    ("remote", "local", "expected"),
    [
        ("2026-v3", "2025-definitive", "update_available"),
        ("2025-v3", "2025-definitive", "current"),
        ("2026-v3", None, "not_loaded"),
        ("2025-v3", "2026-definitive", "blocked_remote_older"),
    ],
)
def test_catalog_decision_orders_only_definitive_editions(
    tmp_path: Path,
    remote: str,
    local: str | None,
    expected: str,
) -> None:
    assert decision(arguments(tmp_path), remote=remote, local=local)["status"] == expected


def test_catalog_decision_keeps_provisional_snapshot_non_importable(tmp_path: Path) -> None:
    release_decision = decision(arguments(tmp_path))

    assert release_decision["release"]["edition"] == "2026-v3"
    assert release_decision["provisional_release"] == "2027-v1"
    assert release_decision["provisional_release_importable"] is False
    with pytest.raises(RuntimeError, match="definitive YYYY-v3"):
        MANAGER.fetch_release_decision_from_item(arguments(tmp_path), catalog_item(remote="2026-v1"))


def test_stage_and_apply_commands_are_separate_and_local_only(tmp_path: Path) -> None:
    args = arguments(tmp_path)
    release = release_2026()
    stage = MANAGER.build_operator_command(args, release, fetch_only=True)
    apply = MANAGER.build_operator_command(args, release, fetch_only=False)

    assert "--force" in stage and "--fetch-only" in stage
    assert "--force" not in apply and "--fetch-only" not in apply
    assert stage[stage.index("--archive-url") + 1] == release.archive_url
    with pytest.raises(RuntimeError, match="inside GeoIntel"):
        MANAGER.internal_base_url("http://192.168.123.45:1202/api/v1")


def test_staged_plan_binds_source_schema_codelist_scope_and_baseline(tmp_path: Path) -> None:
    args = arguments(tmp_path)
    release = release_2026()
    _, plan = staged_plan(args, release, decision(args))

    evidence = plan["evidence"]
    assert evidence["feature_count"] == 121_500
    assert evidence["crop_entry_count"] == 1
    assert evidence["member_nis_codes"] == ["13025", "13003"]
    assert evidence["baseline"]["year"] == 2025
    assert len(evidence["baseline"]["manifest_sha256"]) == 64
    assert evidence["baseline"]["feature_count_change_ratio"] == 0.0125
    assert plan["plan_sha256"] == MANAGER.canonical_sha256(plan, "plan_sha256")


def test_modified_staged_bytes_invalidate_plan(tmp_path: Path) -> None:
    args = arguments(tmp_path)
    release = release_2026()
    path, plan = staged_plan(args, release, decision(args))
    args.plan_path = path
    args.confirm_plan_sha256 = plan["plan_sha256"]
    Path(plan["evidence"]["artifact_path"]).write_text("tampered", encoding="utf-8")

    with pytest.raises(RuntimeError, match="incomplete or no longer match"):
        MANAGER.load_staged_plan(args, release)


def test_plan_and_review_paths_must_remain_governed(tmp_path: Path) -> None:
    args = arguments(tmp_path, plan_path=tmp_path / "outside.json")
    with pytest.raises(RuntimeError, match="outside the governed"):
        MANAGER.governed_evidence_path(args, args.plan_path)


def test_review_requires_named_approval_and_exact_plan_hash(tmp_path: Path) -> None:
    args = arguments(tmp_path)
    release = release_2026()
    path, plan = staged_plan(args, release, decision(args))

    with pytest.raises(RuntimeError, match="--approve"):
        MANAGER.build_review_evidence(args, path, plan)
    args.approve = True
    args.reviewer = "Jens"
    review = MANAGER.build_review_evidence(args, path, plan)
    assert review["staged_plan_sha256"] == plan["plan_sha256"]
    assert review["review_sha256"] == MANAGER.canonical_sha256(review, "review_sha256")


def test_review_tampering_and_catalog_drift_fail_closed(tmp_path: Path) -> None:
    args = arguments(tmp_path, approve=True, reviewer="Jens")
    release = release_2026()
    path, plan = staged_plan(args, release, decision(args))
    review = MANAGER.build_review_evidence(args, path, plan)
    review_path = MANAGER.default_review_path(args, release.year)
    MANAGER.write_json(review_path, review)
    payload = json.loads(review_path.read_text(encoding="utf-8"))
    payload["reviewer"] = "changed"
    MANAGER.write_json(review_path, payload)
    args.review_path = review_path
    args.confirm_review_sha256 = review["review_sha256"]

    with pytest.raises(RuntimeError, match="checksum is invalid"):
        MANAGER.load_review_evidence(args, release, plan)
    changed = MANAGER.fetch_release_decision_from_item(args, catalog_item(catalog_hash="b" * 64))
    with pytest.raises(RuntimeError, match="evidence changed"):
        MANAGER.require_catalog_unchanged(plan, changed)


def test_catalog_check_timestamp_does_not_create_false_drift(tmp_path: Path) -> None:
    args = arguments(tmp_path)
    first = MANAGER.fetch_release_decision_from_item(args, catalog_item(checked_at="2027-03-16T08:00:00Z"))
    second = MANAGER.fetch_release_decision_from_item(args, catalog_item(checked_at="2027-03-16T09:00:00Z"))
    plan = {"release": first["release"], "catalog_identity": first["catalog_identity"]}

    MANAGER.require_catalog_unchanged(plan, second)
    assert first["catalog_checked_at"] != second["catalog_checked_at"]


def test_current_edition_cannot_stage_or_write_evidence(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    args = arguments(tmp_path, action="stage", confirm_edition="2025-v3")
    current = decision(args, remote="2025-v3", local="2025-definitive")
    monkeypatch.setattr(MANAGER, "parse_args", lambda: args)
    monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None)
    monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: current)
    monkeypatch.setattr(
        MANAGER,
        "run_operator",
        lambda *_args, **_kwargs: pytest.fail("operator must not run for current edition"),
    )

    assert MANAGER.main() == 1
    assert not args.evidence_root.exists()


def test_full_apply_requires_review_and_verifies_final_dataset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    args = arguments(tmp_path, action="apply", confirm_edition="2026-v3", approve=True, reviewer="Jens")
    release = release_2026()
    update = decision(args)
    path, plan = staged_plan(args, release, update)
    review = MANAGER.build_review_evidence(args, path, plan)
    review_path = MANAGER.default_review_path(args, release.year)
    MANAGER.write_json(review_path, review)
    args.confirm_plan_sha256 = plan["plan_sha256"]
    args.confirm_review_sha256 = review["review_sha256"]
    current = decision(args, remote="2026-v3", local="2026-definitive")
    decisions = iter((update, current))
    monkeypatch.setattr(MANAGER, "parse_args", lambda: args)
    monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None)
    monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: next(decisions))
    monkeypatch.setattr(
        MANAGER,
        "run_operator",
        lambda *_args, **_kwargs: {
            "status": "ok",
            "years": [{"year": 2026, "status": "imported", "dataset_id": "dataset-2026", "feature_count": 121_500}],
        },
    )

    assert MANAGER.main() == 0
    applied = json.loads(path.with_name("applied-evidence.json").read_text(encoding="utf-8"))
    assert applied["dataset_id"] == "dataset-2026"
    assert applied["review_sha256"] == review["review_sha256"]
    assert applied["applied_evidence_sha256"] == MANAGER.canonical_sha256(applied, "applied_evidence_sha256")


def test_manager_is_packaged_and_never_writes_vector_features_directly() -> None:
    manager = (SCRIPTS / "manage_alz_agriculture_release.py").read_text(encoding="utf-8")
    operator = (SCRIPTS / "provision_agricultural_parcel_history.py").read_text(encoding="utf-8")
    dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
    readiness = (SCRIPTS / "run_readiness_check.sh").read_text(encoding="utf-8")

    assert "INSERT INTO vector_features" not in manager
    assert "INSERT INTO vector_features" not in operator
    assert "/datasets/upload" in operator
    assert "COPY scripts/manage_alz_agriculture_release.py" in dockerfile
    assert "py_compile scripts/manage_alz_agriculture_release.py" in readiness