geointel/tests/test_evaluate_yolo_checkpoint_matrix.py

import json
import sys
from pathlib import Path

from scripts.evaluate_yolo_checkpoint_matrix import (
    dataset_overlap_evidence,
    main,
    model_lineage_independence_evidence,
    validate_pure_background_prefixes,
    write_blocked_manifest,
)


def test_dataset_overlap_evidence_marks_repeated_validation_rows(
    tmp_path: Path,
) -> None:
    dataset_yaml = tmp_path / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    (tmp_path / "yolo_tile_dataset_summary.json").write_text(
        json.dumps({"tile_size": 512, "stride": 256}), encoding="utf-8"
    )

    evidence = dataset_overlap_evidence(dataset_yaml)

    assert evidence["status"] == "overlapping"
    assert evidence["overlap_pixels"] == 256
    assert evidence["validation_tiles_non_overlapping"] is False
    assert evidence["statistical_independence_established"] is False


def test_dataset_overlap_evidence_is_explicit_when_summary_missing(
    tmp_path: Path,
) -> None:
    evidence = dataset_overlap_evidence(tmp_path / "dataset.yaml")

    assert evidence["status"] == "unavailable"
    assert evidence["validation_tiles_non_overlapping"] is None
    assert evidence["statistical_independence_established"] is False


def test_nonoverlap_does_not_overclaim_statistical_independence(
    tmp_path: Path,
) -> None:
    dataset_yaml = tmp_path / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    (tmp_path / "yolo_tile_dataset_summary.json").write_text(
        json.dumps({"tile_size": 512, "stride": 512}), encoding="utf-8"
    )

    evidence = dataset_overlap_evidence(dataset_yaml)

    assert evidence["validation_tiles_non_overlapping"] is True
    assert evidence["statistical_independence_established"] is False
    assert "not established" in evidence["interpretation"]


def _write_summary(path: Path, rows: list[tuple[str, str]]) -> None:
    path.write_text(
        json.dumps(
            {
                "tile_size": 512,
                "stride": 512,
                "tiles": [
                    {"sample_slug": sample_slug, "split": split}
                    for sample_slug, split in rows
                ],
            }
        ),
        encoding="utf-8",
    )


def test_model_lineage_independence_detects_exposure_in_every_split(
    tmp_path: Path,
) -> None:
    evaluation = tmp_path / "evaluation"
    evaluation.mkdir()
    dataset_yaml = evaluation / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    _write_summary(
        evaluation / "yolo_tile_dataset_summary.json",
        [("turnhout", "val"), ("postel_bos", "val")],
    )
    active = tmp_path / "active.json"
    challenger = tmp_path / "challenger.json"
    _write_summary(active, [("postel_bos", "train"), ("turnhout", "val")])
    _write_summary(
        challenger, [("postel_bos", "train"), ("dessel", "train")]
    )

    evidence = model_lineage_independence_evidence(
        dataset_yaml, [active, challenger]
    )

    assert evidence["status"] == "overlap"
    assert evidence["overlapping_evaluation_samples"] == ["postel_bos", "turnhout"]
    assert evidence["independent_for_all_supplied_lineage_corpora"] is False
    assert [
        row["overlapping_evaluation_samples"] for row in evidence["lineage_corpora"]
    ] == [["postel_bos", "turnhout"], ["postel_bos"]]
    assert evidence["lineage_corpora"][0]["exposure_roles"]["turnhout"] == ["val"]


def test_model_lineage_independence_accepts_disjoint_samples(tmp_path: Path) -> None:
    evaluation = tmp_path / "evaluation"
    evaluation.mkdir()
    dataset_yaml = evaluation / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    _write_summary(
        evaluation / "yolo_tile_dataset_summary.json", [("turnhout", "val")]
    )
    training = tmp_path / "training.json"
    _write_summary(training, [("mol", "train")])

    evidence = model_lineage_independence_evidence(dataset_yaml, [training])

    assert evidence["status"] == "independent"
    assert evidence["overlapping_evaluation_samples"] == []
    assert evidence["independent_for_all_supplied_lineage_corpora"] is True


def test_model_lineage_independence_is_unavailable_without_ancestral_corpus(
    tmp_path: Path,
) -> None:
    evaluation = tmp_path / "evaluation"
    evaluation.mkdir()
    dataset_yaml = evaluation / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    _write_summary(
        evaluation / "yolo_tile_dataset_summary.json", [("turnhout", "val")]
    )

    evidence = model_lineage_independence_evidence(dataset_yaml, [])

    assert evidence["status"] == "unavailable"
    assert evidence["overlapping_evaluation_samples"] == []
    assert evidence["independent_for_all_supplied_lineage_corpora"] is False


def test_blocked_manifest_records_that_models_and_gpu_were_not_used(
    tmp_path: Path,
) -> None:
    dataset_yaml = tmp_path / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    output = tmp_path / "evidence" / "matrix.json"

    write_blocked_manifest(
        output,
        dataset_yaml,
        {
            "status": "overlap",
            "independent_for_all_supplied_lineage_corpora": False,
        },
    )

    payload = json.loads(output.read_text(encoding="utf-8"))
    assert payload["status"] == "blocked_model_lineage_sample_exposure"
    assert payload["model_loading_attempted"] is False
    assert payload["gpu_inference_attempted"] is False


def test_cli_blocks_lineage_validation_exposure_before_model_resolution(
    tmp_path: Path, monkeypatch
) -> None:
    evaluation = tmp_path / "evaluation"
    images = evaluation / "images" / "val"
    images.mkdir(parents=True)
    (images / "turnhout_r0_c0.png").write_bytes(b"not opened before gate")
    dataset_yaml = evaluation / "dataset.yaml"
    dataset_yaml.write_text("path: .\nval: images/val\n", encoding="utf-8")
    _write_summary(
        evaluation / "yolo_tile_dataset_summary.json", [("turnhout", "val")]
    )
    ancestor = tmp_path / "ancestor.json"
    _write_summary(ancestor, [("turnhout", "val"), ("mol", "train")])
    output = tmp_path / "blocked.json"
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "evaluate_yolo_checkpoint_matrix.py",
            "--dataset-yaml",
            str(dataset_yaml),
            "--model",
            str(tmp_path / "model-is-never-resolved.pt"),
            "--output",
            str(output),
            "--background-prefix",
            "background",
            "--lineage-summary",
            str(ancestor),
            "--require-lineage-sample-independence",
        ],
    )

    assert main() == 3
    payload = json.loads(output.read_text(encoding="utf-8"))
    assert payload["status"] == "blocked_model_lineage_sample_exposure"
    assert payload["model_loading_attempted"] is False


def test_pure_background_prefix_rejects_nonempty_labels(tmp_path: Path) -> None:
    image_dir = tmp_path / "images" / "val"
    label_dir = tmp_path / "labels" / "val"
    image_dir.mkdir(parents=True)
    label_dir.mkdir(parents=True)
    image = image_dir / "sparse_bg_0001.png"
    image.write_bytes(b"image")
    (label_dir / "sparse_bg_0001.txt").write_text(
        "0 0.5 0.5 0.1 0.1\n", encoding="utf-8"
    )

    import pytest

    with pytest.raises(ValueError, match="non-empty labels"):
        validate_pure_background_prefixes([image], ("sparse_bg",))


def test_pure_background_prefix_accepts_empty_labels(tmp_path: Path) -> None:
    image_dir = tmp_path / "images" / "val"
    label_dir = tmp_path / "labels" / "val"
    image_dir.mkdir(parents=True)
    label_dir.mkdir(parents=True)
    image = image_dir / "pure_bg_0001.png"
    image.write_bytes(b"image")
    (label_dir / "pure_bg_0001.txt").write_text("", encoding="utf-8")

    assert validate_pure_background_prefixes([image], ("pure_bg",)) == [image]