tests/test_model_accuracy_summary.py

from __future__ import annotations

import csv
import json
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
FIELDS = ["model_id", "model_name", "metric", "fp32", "quantized", "delta", "published_comparison"]


def build(tmp_path: Path) -> tuple[Path, list[dict[str, str]]]:
    output = tmp_path / "accuracy.csv"
    subprocess.run(
        [sys.executable, str(ROOT / "scripts/build_model_accuracy_summary.py"), "--repo-root", str(ROOT), "--output", str(output)],
        check=True,
        capture_output=True,
        text=True,
    )
    with output.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        assert reader.fieldnames == FIELDS
        rows = list(reader)
    return output, rows


def test_accuracy_csv_is_compact_and_covers_active_models(tmp_path: Path) -> None:
    _, rows = build(tmp_path)
    registry = {
        row["model_id"]
        for row in csv.DictReader((ROOT / "model_registry.csv").open(newline="", encoding="utf-8"))
        if row["eligibility"] == "ELIGIBLE"
    }
    assert len(rows) == 21
    assert {row["model_id"] for row in rows} == registry
    assert all(list(row) == FIELDS for row in rows)


def test_accuracy_values_and_known_upstream_corrections(tmp_path: Path) -> None:
    _, rows = build(tmp_path)
    by_id = {row["model_id"]: row for row in rows}
    assert by_id["AD01"]["fp32"] == "0.876001 / 0.764121"
    assert by_id["VC02"]["fp32"] == "87.00%"
    assert by_id["VC09"]["fp32"] == "56.85% / 79.87%"
    assert by_id["VC09"]["quantized"] == "56.48% / 79.76%"
    assert by_id["VC12"]["fp32"] == "69.48% / 89.26%"
    assert by_id["VC12"]["quantized"] == "68.30% / 88.44%"
    assert by_id["VC13"]["metric"] == "Top-1 / Top-5 error"


def test_task_specific_quality_metrics_remain_explicit(tmp_path: Path) -> None:
    _, rows = build(tmp_path)
    by_id = {row["model_id"]: row for row in rows}
    assert by_id["SG06"]["metric"] == "mIoU"
    assert by_id["SP02"]["metric"] == "FP / FN (1 s)"
    assert by_id["SP02"]["fp32"] == "5 / 6"
    assert by_id["SP02"]["quantized"] == "4 / 6"
    assert by_id["SG08"]["metric"] == "mIoU (CamVid cross-dataset)"
    assert by_id["SG08"]["fp32"] == "50.6498%"
    assert by_id["SG08"]["quantized"] == "51.1600%"


def test_published_comparison_is_sparse_and_only_used_for_direct_comparisons(tmp_path: Path) -> None:
    _, rows = build(tmp_path)
    by_id = {row["model_id"]: row for row in rows}
    assert set(
        model_id for model_id, row in by_id.items()
        if row["published_comparison"] != "공개 수치 없음."
    ) == {
        "AD01", "SG06", "SG07", "SP01"
    }
    assert "+0.3198 pp" in by_id["SG06"]["published_comparison"]
    assert by_id["SG08"]["published_comparison"] == "공개 수치 없음."
    assert by_id["OD06"]["published_comparison"] == "공개 수치 없음."


def test_vc11_compute_graph_equivalence_is_persisted() -> None:
    audit = json.loads((ROOT / "research/evidence/vision/vc11_compute_graph_equivalence.json").read_text(encoding="utf-8"))
    assert audit["status"] == "PASS"
    for value in audit["variants"].values():
        assert value["status"] == "PASS"
        assert value["official_compute_graph_sha256"] == value["acquired_compute_graph_sha256"]


def test_validator_accepts_and_rejects_value_tamper(tmp_path: Path) -> None:
    output, rows = build(tmp_path)
    validation = tmp_path / "validation.json"
    command = [sys.executable, str(ROOT / "scripts/validate_model_accuracy_summary.py"), "--repo-root", str(ROOT), "--csv", str(output), "--output", str(validation)]
    accepted = subprocess.run(command, capture_output=True, text=True)
    assert accepted.returncode == 0, accepted.stderr
    assert json.loads(validation.read_text(encoding="utf-8"))["status"] == "PASS"

    next(row for row in rows if row["model_id"] == "VC09")["fp32"] = "56.48% / 79.76%"
    with output.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=FIELDS, lineterminator="\n")
        writer.writeheader()
        writer.writerows(rows)
    rejected = subprocess.run(command, capture_output=True, text=True)
    assert rejected.returncode == 1
    report = json.loads(validation.read_text(encoding="utf-8"))
    assert report["status"] == "FAIL"
    assert any(item["category"] in {"metric_value", "known_correction"} for item in report["failures"])


def test_previously_missing_models_now_have_measured_values(tmp_path: Path) -> None:
    _, rows = build(tmp_path)
    by_id = {row["model_id"]: row for row in rows}
    assert by_id["LM04"]["fp32"] == "0.667078 / 0.318983 / 0.233873"
    assert by_id["LM04"]["quantized"] == "0.668862 / 0.322615 / 0.235853"
    assert by_id["OD06"]["fp32"] == "24.8751%"
    assert by_id["OD06"]["quantized"] == "24.2822%"
    assert by_id["OD07"]["fp32"] == "31.8594%"
    assert by_id["OD07"]["quantized"] == "31.4191%"
    assert by_id["SP08"]["fp32"] == by_id["SP08"]["quantized"] == "94.05%"
    assert all(row[field] != "N/A" for row in rows for field in ("metric", "fp32", "quantized", "delta"))