clay-borg/tools/edition-check.py

#!/usr/bin/env python3
"""Is the vendored edition still what ground-game published? (ADR-0011 D2)

`editions/ground-darvo-r0/` is a copy of content owned by another repo.
A stale copy is worse than no copy, so the digest is committed and this
compares it.

**An absent upstream is reported absent, never as a pass.** That is the
shape ADR-0009 used for `node`: a check that cannot run says so, because
a silent skip is the class this project has found seven times.
"""
import hashlib
import os
import re
import sys

from repo import ROOT, enter_root

EDITION = "editions/ground-darvo-r0"
PROVENANCE = f"{EDITION}/PROVENANCE.md"
UPSTREAM_DIR = os.path.join(os.path.dirname(ROOT), "ground-game", EDITION)


def digest(path):
    return hashlib.sha256(open(path, "rb").read()).hexdigest()


def recorded():
    """Every recorded digest, by filename.

    The first version matched ONE `sha256 <hex>` and assumed it described
    `Problems.csv`. ADR-0015 vendored three more files, and the check
    reported the first digest against the wrong file -- a gate written for
    a single-file world silently comparing across files.
    """
    text = open(os.path.join(ROOT, PROVENANCE)).read()
    found = dict(
        (name, d)
        for d, name in re.findall(r"sha256\s+([0-9a-f]{64})\s+(\S+)", text)
    )
    if not found:
        raise ValueError(f"{PROVENANCE} records no `sha256 <hex> <file>` digests")
    return found


def vendored_files():
    """Every `.csv` actually present, so a file added without a digest is
    caught rather than skipped."""
    d = os.path.join(ROOT, EDITION)
    return sorted(f for f in os.listdir(d) if f.endswith(".csv"))


def check():
    want = recorded()
    print("edition-check — vendored data against its provenance")
    rc = 0

    present = vendored_files()
    undocumented = [f for f in present if f not in want]
    if undocumented:
        print(f"  [FAIL] vendored with no recorded digest: {', '.join(undocumented)}")
        rc = 1
    missing = [f for f in want if f not in present]
    if missing:
        print(f"  [FAIL] a digest is recorded for a file that is not here: {', '.join(missing)}")
        rc = 1

    for name in present:
        if name not in want:
            continue
        have = digest(os.path.join(ROOT, EDITION, name))
        if have != want[name]:
            print(f"  [FAIL] {name} does not match its recorded digest")
            print(f"         recorded {want[name]}\n         actual   {have}")
            rc = 1
        else:
            print(f"  [ok  ] {name} matches its recorded digest")

    # ADR-0015 D3's falsifier, checked rather than asserted: the hand
    # reader handles commas inside quotes and NOTHING ELSE. A doubled
    # quote or an embedded newline means `csv` is the answer after all.
    for name in present:
        raw = open(os.path.join(ROOT, EDITION, name), encoding="utf-8-sig").read()
        if '""' in raw:
            print(f"  [FAIL] {name} contains a doubled quote — ADR-0011's revisit")
            print("         condition has fired; the hand reader cannot parse it")
            rc = 1
        # An embedded newline shows up as an odd quote count on a line.
        for i, line in enumerate(raw.splitlines(), 1):
            if line.count('"') % 2:
                print(f"  [FAIL] {name}:{i} has an unbalanced quote — embedded newline?")
                rc = 1
                break
    if rc == 0:
        print("  [ok  ] no doubled quotes or embedded newlines (ADR-0015 D3)")

    if not os.path.isdir(UPSTREAM_DIR):
        # NOT a pass and NOT a failure: the question could not be asked.
        print("  [----] upstream not checked out — freshness UNVERIFIED")
        print(f"         expected {UPSTREAM_DIR}")
        return rc
    for name in present:
        up = os.path.join(UPSTREAM_DIR, name)
        if not os.path.exists(up):
            print(f"  [FAIL] {name} is not in upstream — where did it come from?")
            rc = 1
        elif digest(up) != digest(os.path.join(ROOT, EDITION, name)):
            print(f"  [FAIL] upstream {name} has changed since it was vendored")
            rc = 1
    if rc == 0:
        print("  [ok  ] every vendored copy is current with ../ground-game")
    return rc


def self_test():
    """A checker that cannot detect a mismatch is decoration."""
    results = []

    def chk(name, ok, detail=""):
        results.append((name, ok, detail))

    present = vendored_files()
    want = recorded()
    chk("vendored files exist", len(present) >= 4, ", ".join(present))
    chk("every vendored file has a recorded digest",
        all(f in want for f in present),
        "a file added without a digest must fail, not be skipped")
    chk("every digest names a file that is here",
        all(f in present for f in want),
        "a stale digest is a lie with a filename")
    chk("digests match the real files",
        all(digest(os.path.join(ROOT, EDITION, f)) == want[f] for f in present))

    # The control that matters: a changed byte must be detected.
    import tempfile
    one = present[0]
    with tempfile.NamedTemporaryFile("wb", delete=False) as fh:
        fh.write(open(os.path.join(ROOT, EDITION, one), "rb").read() + b"\n#tamper\n")
        tampered = fh.name
    chk("a tampered copy has a different digest",
        digest(tampered) != want[one], "otherwise the check is decoration")
    os.unlink(tampered)

    # ADR-0015 D3's condition must be DETECTABLE, or asserting its absence
    # in `check()` proves nothing.
    chk("a doubled quote would be detected", '""' in 'a,""b""', "the pattern check itself")
    chk("an unbalanced quote would be detected", 'a,"b'.count('"') % 2 == 1)

    print("edition-check self-test (positive control)")
    ok = True
    for name, passed, det in results:
        print(f"  [{'ok ' if passed else 'FAIL'}] {name}" + (f"  — {det}" if det else ""))
        ok &= passed
    return 0 if ok else 1


if __name__ == "__main__":
    enter_root()
    raise SystemExit(self_test() if "--self-test" in sys.argv else check())