photoanalyzer/tests/characterization/test_donor_ledger.py

"""Ledger lint (US01-01, extended by US07-01).

Every donor-ledger row must carry a real source reference — now inside the frozen
archive — a target location, and either existing characterization test IDs or a
real pending backlog story. Since archival (US07-01) a row may also be ``resolved``:
its replacement has shipped, ``parity`` names tests that exist and prove it, and
``delta`` states every intentional difference. Nothing may quietly become
"finished" without one of those two."""
import re
from pathlib import Path

import yaml

REPO = Path(__file__).resolve().parents[2]
ARCHIVE = REPO / "legacy_cli_archive"
LEDGER = ARCHIVE / "donor_ledger.yaml"
ARCHIVED_SOURCES = ARCHIVE / "src"
STORIES = REPO / "delivery_backlog" / "stories"
TESTS_DIR = Path(__file__).resolve().parent

CLASSIFICATIONS = {"reuse", "extract", "refactor", "replace"}
REQUIRED_AREAS = {"discovery", "hashing", "imaging", "nsfw", "vision", "exif",
                  "database", "ui", "configuration", "logging", "cancellation",
                  "error"}
STATUSES = {"characterized", "resolved", "pending"}


def load_rows():
    rows = yaml.safe_load(LEDGER.read_text(encoding="utf-8"))["rows"]
    assert rows, "empty ledger"
    return rows


def collect_test_ids() -> set:
    """module::function for every test in this suite."""
    ids = set()
    for f in TESTS_DIR.glob("test_*.py"):
        for m in re.finditer(r"^def (test_\w+)", f.read_text(encoding="utf-8"),
                             re.MULTILINE):
            ids.add(f"{f.stem}::{m.group(1)}")
    return ids


def test_rows_have_required_fields_and_unique_ids():
    rows = load_rows()
    ids = [r["id"] for r in rows]
    assert len(ids) == len(set(ids)), "duplicate row ids"
    for r in rows:
        for field in ("id", "area", "source", "classification", "rationale",
                      "target", "status"):
            assert r.get(field), f"{r.get('id', '?')}: missing {field}"
        assert r["classification"] in CLASSIFICATIONS, r["id"]
        assert r["status"] in STATUSES, r["id"]
        assert len(str(r["rationale"]).strip()) >= 20, \
            f"{r['id']}: rationale too thin to count as a documented reason"


def test_source_references_resolve():
    """Source paths are relative to the archive: the donors moved there, whole."""
    for r in load_rows():
        src = r["source"]
        f = ARCHIVED_SOURCES / src["file"]
        assert f.is_file(), f"{r['id']}: source file {src['file']} missing from the archive"
        text = f.read_text(encoding="utf-8")
        for sym in src["symbols"]:
            assert sym in text, f"{r['id']}: symbol {sym!r} not found in {src['file']}"


def test_rows_have_tests_or_pending_story():
    known_tests = collect_test_ids()
    for r in load_rows():
        tests = r.get("tests", [])
        pending = r.get("pending_story")
        parity = r.get("parity", [])
        assert tests or pending or parity, \
            f"{r['id']}: neither characterization tests, parity tests, nor a pending story"
        for t in tests:
            assert t in known_tests, f"{r['id']}: unknown test id {t}"
        if pending:
            matches = list(STORIES.glob(f"{pending}-*.md"))
            assert matches, f"{r['id']}: pending_story {pending} has no story file"
        if r["status"] == "characterized":
            assert tests, f"{r['id']}: characterized rows need test ids"


def test_resolved_rows_name_their_parity_or_their_delta():
    """A resolved row is a claim that the behavior is handled. It has to say how:
    tests that prove the replacement, or a stated difference — usually both."""
    for r in load_rows():
        if r["status"] != "resolved":
            assert "parity" not in r, f"{r['id']}: parity on a non-resolved row"
            continue
        parity = r.get("parity", [])
        delta = r.get("delta")
        assert parity or delta, f"{r['id']}: resolved without parity tests or a delta"
        if delta:
            assert len(str(delta).strip()) >= 20, f"{r['id']}: delta too thin to be a reason"
        for ref in parity:
            rel, _, func = ref.partition("::")
            path = REPO / rel
            assert path.is_file(), f"{r['id']}: parity test file {rel} missing"
            assert f"def {func}" in path.read_text(encoding="utf-8"), \
                f"{r['id']}: parity test {ref} not found"


def test_pending_rows_are_the_only_unfinished_work():
    """The ledger is the honest list of what has not been carried over: a pending
    row names the story that will, and that story must still be open work."""
    for r in load_rows():
        if r["status"] != "pending":
            continue
        story = r.get("pending_story")
        assert story, f"{r['id']}: pending without a story"
        assert list(STORIES.glob(f"{story}-*.md")), f"{r['id']}: unknown story {story}"


def test_every_target_module_exists():
    """A row is only finished if the thing it points at is really there."""
    for r in load_rows():
        if r["status"] == "pending":
            continue
        modules = re.findall(r"photo_pipeline/[\w/]+\.py", str(r["target"]))
        for module in modules:
            assert (REPO / module).is_file(), f"{r['id']}: target {module} does not exist"


def test_all_required_areas_covered():
    covered = {r["area"] for r in load_rows()}
    assert REQUIRED_AREAS <= covered, f"uncovered areas: {REQUIRED_AREAS - covered}"
    assert covered <= REQUIRED_AREAS, f"unknown areas: {covered - REQUIRED_AREAS}"


def test_every_donor_is_in_the_archive_and_nowhere_else():
    """US01-01 forbade archiving before characterization; US07-01 requires it after.
    Each donor exists exactly once — frozen, in the archive."""
    for donor in ("photo_analyzer.py", "nsfwtag/scoring.py", "nsfwtag/exif.py",
                  "nsfwtag/server.py", "webapp/query.py", "webapp/runner.py",
                  "webapp/server.py"):
        assert (ARCHIVED_SOURCES / donor).is_file(), f"donor missing from archive: {donor}"
        assert not (REPO / donor).exists(), f"donor still live at the repo root: {donor}"