geointel/backend/tests/test_training_release_manifest.py

from __future__ import annotations

import hashlib
import importlib.util
import json
import sys
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))
SCRIPT = SCRIPTS / "training_release_manifest.py"
SPEC = importlib.util.spec_from_file_location("training_release_manifest", SCRIPT)
assert SPEC and SPEC.loader
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


@pytest.fixture(autouse=True)
def _synthetic_release_uses_a_static_live_registry_spy(monkeypatch: pytest.MonkeyPatch) -> None:
    """Filesystem unit fixtures cannot resolve a real database, but must ask for it."""

    original_assert = MODULE.assert_frozen_manifest_training_eligible
    original_failures = MODULE.frozen_manifest_training_eligibility_failures

    def static_assert(*args, **kwargs):
        assert kwargs.get("verify_live") is True
        kwargs["verify_live"] = False
        return original_assert(*args, **kwargs)

    def static_failures(*args, **kwargs):
        assert kwargs.get("verify_live") is True
        kwargs["verify_live"] = False
        return original_failures(*args, **kwargs)

    monkeypatch.setattr(MODULE, "assert_frozen_manifest_training_eligible", static_assert)
    monkeypatch.setattr(MODULE, "frozen_manifest_training_eligibility_failures", static_failures)


def write_corpus_manifest(tmp_path: Path, *, fixture_mode: bool = False) -> Path:
    policy = "geointel-training-source-eligibility/v1"
    def pair(sample_slug: str) -> dict:
        raster_id = f"dataset:raster:{sample_slug}"
        reference_id = f"dataset:reference:{sample_slug}"
        return {
            "policy_version": policy,
            "eligible": True,
            "fixture_mode": fixture_mode,
            "raster": {
                "eligible": True,
                "reasons": [],
                "evidence": {
                    "dataset_id": raster_id,
                    "checksum_sha256": "a" * 64,
                    "source_registry_id": "registry:orthophoto",
                    "source_snapshot_id": "snapshot:orthophoto",
                },
            },
            "reference": {
                "eligible": True,
                "reasons": [],
                "evidence": {
                    "dataset_id": reference_id,
                    "checksum_sha256": "b" * 64,
                    "source_registry_id": "registry:grb",
                    "source_snapshot_id": "snapshot:grb",
                },
            },
        }

    samples = []
    for sample_slug, split in (("fixture-train", "train"), ("fixture-val", "val")):
        samples.append(
            {
                "sample_slug": sample_slug,
                "split": split,
                "raster_dataset_id": f"dataset:raster:{sample_slug}",
                "reference_dataset_id": f"dataset:reference:{sample_slug}",
                "training_eligibility": pair(sample_slug),
            }
        )
    manifest = tmp_path / "operator_samples_manifest.json"
    manifest.write_text(
        json.dumps(
            {
                "immutable": True,
                "training_eligibility": {
                    "policy_version": policy,
                    "status": "eligible",
                    "fixture_mode": fixture_mode,
                },
                "samples": samples,
            }
        ),
        encoding="utf-8",
    )
    (tmp_path / "corpus-freeze.json").write_text(
        json.dumps(
            {
                "schema_version": 2,
                "immutable": True,
                "fixture_mode": fixture_mode,
                "training_eligibility_policy": policy,
                "manifest_sha256": sha256(manifest),
            }
        ),
        encoding="utf-8",
    )
    return manifest


def write_yolo_dataset(tmp_path: Path, *, empty_train_label: bool = False) -> Path:
    dataset_root = tmp_path / "dataset"
    for split, sample_slug in (("train", "fixture-train"), ("val", "fixture-val")):
        image = dataset_root / "images" / split / f"{sample_slug}.png"
        label = dataset_root / "labels" / split / f"{sample_slug}.txt"
        image.parent.mkdir(parents=True, exist_ok=True)
        label.parent.mkdir(parents=True, exist_ok=True)
        image.write_bytes(f"{split}-image".encode("utf-8"))
        label.write_text("" if split == "train" and empty_train_label else "0 0.5 0.5 0.2 0.2\n", encoding="utf-8")
    yaml_path = dataset_root / "dataset.yaml"
    yaml_path.write_text(
        f"path: {dataset_root}\ntrain: images/train\nval: images/val\nnames:\n  0: building\n",
        encoding="utf-8",
    )
    return yaml_path


def write_accepted_review_audit(tmp_path: Path, corpus_manifest: Path) -> Path:
    artifacts = {}
    for sample_slug in ("fixture-train", "fixture-val"):
        artifact = tmp_path / f"{sample_slug}-contact-sheet.png"
        artifact.write_bytes(f"reviewed {sample_slug}".encode("utf-8"))
        artifacts[sample_slug] = artifact
    decisions = tmp_path / "review-decisions.json"
    decisions.write_text(
        json.dumps(
            {
                "decisions": [
                    {
                        "sample_slug": sample_slug,
                        "decision": "accepted",
                        "reviewer": "reviewer@example.test",
                        "reviewed_at": "2026-08-01T12:00:00+00:00",
                        "reviewed_artifact_path": str(artifact.resolve()),
                        "reviewed_artifact_sha256": sha256(artifact),
                    }
                    for sample_slug, artifact in artifacts.items()
                ]
            }
        ),
        encoding="utf-8",
    )
    evidence = {
        "review_decisions_path": str(decisions.resolve()),
        "review_decisions_sha256": sha256(decisions),
        "required_sample_count": 2,
        "accepted_sample_count": 2,
        "accepted_sample_slugs": ["fixture-train", "fixture-val"],
        "reviewer_ids": ["reviewer@example.test"],
        "reviewed_at_by_sample": {
            "fixture-train": "2026-08-01T12:00:00+00:00",
            "fixture-val": "2026-08-01T12:00:00+00:00",
        },
        "reviewed_artifact_path_by_sample": {
            sample_slug: str(artifact.resolve()) for sample_slug, artifact in artifacts.items()
        },
        "reviewed_artifact_sha256_by_sample": {
            sample_slug: sha256(artifact) for sample_slug, artifact in artifacts.items()
        },
    }
    audit = tmp_path / "belgium-building-corpus-audit.json"
    audit.write_text(
        json.dumps(
            {
                "status": "ok",
                "manifest_immutable": True,
                "spatial_leakage_status": "ok",
                "review_complete": True,
                "corpus_manifest_path": str(corpus_manifest.resolve()),
                "corpus_manifest_sha256": sha256(corpus_manifest),
                "human_review_evidence": evidence,
            }
        ),
        encoding="utf-8",
    )
    return audit


def test_valid_operational_release_binds_yaml_assets_corpus_and_human_review(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)

    paths = MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
        review_audit_path=review_audit,
    )

    assert all(path.is_file() for path in paths.values())
    assert MODULE.training_release_failures(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
    ) == []


def test_tile_summary_must_be_an_exact_view_of_the_live_verified_release(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)
    paths = MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
        review_audit_path=review_audit,
    )
    assets = json.loads(paths["asset_manifest"].read_text(encoding="utf-8"))
    summary = yaml_path.parent / "yolo_tile_dataset_summary.json"
    summary.write_text(
        json.dumps(
            {
                "dataset_yaml": str(yaml_path.resolve()),
                "training_release_manifest": str(paths["release_manifest"].resolve()),
                "training_release_manifest_sha256": sha256(paths["release_manifest"]),
                "training_asset_manifest": str(paths["asset_manifest"].resolve()),
                "source_manifest_sha256": sha256(corpus_manifest),
                "tiles": [
                    {
                        "split": entry["split"],
                        "image_path": entry["image_path"],
                        "label_path": entry["label_path"],
                    }
                    for entry in assets["entries"]
                ],
            }
        ),
        encoding="utf-8",
    )

    MODULE.assert_yolo_summary_bound_to_training_release(
        summary_path=summary,
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
    )
    payload = json.loads(summary.read_text(encoding="utf-8"))
    payload["tiles"] = payload["tiles"][:1]
    summary.write_text(json.dumps(payload), encoding="utf-8")

    with pytest.raises(MODULE.TrainingReleaseError, match="complete immutable view"):
        MODULE.assert_yolo_summary_bound_to_training_release(
            summary_path=summary,
            train_yaml=yaml_path,
            corpus_manifest=corpus_manifest,
        )


def test_unbound_or_changed_yaml_is_rejected_before_training(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)

    assert MODULE.training_release_failures(train_yaml=yaml_path) == ["training_release_manifest_missing"]
    MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
        review_audit_path=review_audit,
    )
    yaml_path.write_text(yaml_path.read_text(encoding="utf-8") + "# tampered\n", encoding="utf-8")

    assert "training_release_yaml_checksum_mismatch" in MODULE.training_release_failures(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
    )


def test_changed_label_asset_is_rejected_even_when_yaml_bytes_are_unchanged(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)
    MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
        review_audit_path=review_audit,
    )
    label = yaml_path.parent / "labels" / "train" / "fixture-train.txt"
    label.write_text("0 0.4 0.4 0.2 0.2\n", encoding="utf-8")

    failures = MODULE.training_release_failures(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
    )
    assert "training_release_asset_manifest_content_mismatch" in failures


def test_operational_release_requires_complete_accepted_human_review(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    incomplete_audit = tmp_path / "audit.json"
    incomplete_audit.write_text(
        json.dumps(
            {
                "status": "needs_human_review",
                "manifest_immutable": True,
                "spatial_leakage_status": "ok",
                "review_complete": False,
            }
        ),
        encoding="utf-8",
    )

    try:
        MODULE.create_training_release_manifest(
            train_yaml=yaml_path,
            corpus_manifest=corpus_manifest,
            review_audit_path=incomplete_audit,
        )
    except MODULE.TrainingReleaseError as exc:
        assert "review_complete_not_true" in str(exc)
        assert "accepted_human_review_evidence_missing" in str(exc)
    else:
        raise AssertionError("operational release accepted an incomplete human review")


def test_operational_release_rejects_tampered_accepted_review_artifact(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)
    artifact = tmp_path / "fixture-train-contact-sheet.png"
    artifact.write_bytes(b"changed after review")

    try:
        MODULE.create_training_release_manifest(
            train_yaml=yaml_path,
            corpus_manifest=corpus_manifest,
            review_audit_path=review_audit,
        )
    except MODULE.TrainingReleaseError as exc:
        assert "reviewed_artifact_checksum_mismatch" in str(exc)
    else:
        raise AssertionError("tampered human-review artifact was accepted")


def test_fixture_relaxation_requires_explicit_fixture_corpus_and_is_not_operational(tmp_path: Path) -> None:
    fixture_manifest = write_corpus_manifest(tmp_path, fixture_mode=True)
    yaml_path = write_yolo_dataset(tmp_path)
    MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=fixture_manifest,
        fixture_mode=True,
    )

    assert MODULE.training_release_failures(
        train_yaml=yaml_path,
        corpus_manifest=fixture_manifest,
        fixture_mode=True,
    ) == []
    assert "training_release_fixture_mode_mismatch" in MODULE.training_release_failures(
        train_yaml=yaml_path,
        corpus_manifest=fixture_manifest,
        fixture_mode=False,
    )


def test_release_contracts_a_reviewed_empty_label_as_explicit_pure_background(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path, empty_train_label=True)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)

    paths = MODULE.create_training_release_manifest(
        train_yaml=yaml_path,
        corpus_manifest=corpus_manifest,
        review_audit_path=review_audit,
    )

    labels = json.loads(paths["label_contract_manifest"].read_text(encoding="utf-8"))
    assert labels["counts"]["pure_background"] == 1
    assert MODULE.training_release_failures(train_yaml=yaml_path, corpus_manifest=corpus_manifest) == []


def test_empty_label_without_accepted_sample_review_cannot_become_a_background_negative(tmp_path: Path) -> None:
    corpus_manifest = write_corpus_manifest(tmp_path)
    yaml_path = write_yolo_dataset(tmp_path, empty_train_label=True)
    review_audit = write_accepted_review_audit(tmp_path, corpus_manifest)
    audit_payload = json.loads(review_audit.read_text(encoding="utf-8"))
    evidence = audit_payload["human_review_evidence"]
    evidence["accepted_sample_slugs"] = ["fixture-val"]
    review_audit.write_text(json.dumps(audit_payload), encoding="utf-8")
    review = {
        "status": "accepted",
        "fixture_only": False,
        "review_complete": True,
        "evidence": evidence,
    }

    try:
        MODULE.build_training_label_contract_manifest(
            corpus_manifest_path=corpus_manifest,
            asset_manifest=MODULE.build_yolo_asset_manifest(yaml_path),
            review=review,
            fixture_mode=False,
        )
    except MODULE.TrainingReleaseError as exc:
        assert "Pure-background label sample was not accepted by review" in str(exc)
    else:
        raise AssertionError("unreviewed empty label was accepted as a background negative")