diff --git a/CHECK-ledger-dir-census b/CHECK-ledger-dir-census index 9e5eb46..bcde7e6 100755 --- a/CHECK-ledger-dir-census +++ b/CHECK-ledger-dir-census @@ -1,9 +1,9 @@ #!/usr/bin/env bash -# CHECK -- ledger dir census (forge: extract lane). Exit 0 iff committed -# wizard-work/ledger/2026-10-31-ledger-dir-census.csv matches a fresh -# re-derivation from the live filesystem. +# CHECK -- ledger dir census (extract lane). Exit 0 iff the committed +# wizard-work/ledger/2026-10-31-ledger-dir-census.csv is the re-derivation of +# the TRACKED population at the rev its manifest pins -- read from git, never +# from the working tree, so a clean clone and the host that wrote it agree. +# 0 pass, 1 fault, 2 could not grade (the checker prints which). set -u -cd "$(dirname "${BASH_SOURCE[0]}")" || exit 1 -REPO_ROOT="$(pwd)" -cd "$REPO_ROOT" || exit 1 -exec python3 wizard-work/ledger/verify_ledger_dir_census.py +cd "$(dirname "${BASH_SOURCE[0]}")" || exit 2 +exec python3 wizard-work/ledger/verify_ledger_dir_census.py --check --repo "$(pwd)" diff --git a/wizard-work/ledger/build_2026-10-31_ledger_dir_census.py b/wizard-work/ledger/build_2026-10-31_ledger_dir_census.py index 743198c..979b212 100644 --- a/wizard-work/ledger/build_2026-10-31_ledger_dir_census.py +++ b/wizard-work/ledger/build_2026-10-31_ledger_dir_census.py @@ -1,38 +1,27 @@ #!/usr/bin/env python3 # build_2026-10-31_ledger_dir_census.py -# Extract: rebuild the committed artifact '2026-10-31-ledger-dir-census.csv' -# from the LIVE filesystem of wizard-work/ledger/ (rel_path, size_bytes, sha256), -# byte-for-byte matching verify_ledger_dir_census.py's re_derive(). -# Usage: python3 build_2026-10-31_ledger_dir_census.py [out_csv] -import os, sys, csv, hashlib +# Extract: rebuild the committed artifact '2026-10-31-ledger-dir-census.csv'. +# +# Same derivation as build_dir_census.py -- a wrapper over +# verify_ledger_dir_census.py --write, so the two builders and the checker all +# read one implementation. The census is the TRACKED population of +# wizard-work/ledger at a named rev (HEAD unless --rev names one). +# +# Usage: +# python3 build_2026-10-31_ledger_dir_census.py # in place +# python3 build_2026-10-31_ledger_dir_census.py /tmp/census.csv # elsewhere +import os +import sys -HERE = os.path.dirname(os.path.abspath(__file__)) # wizard-work/ledger -REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) # troupe-standup root -ARTIFACT = os.path.join(REPO_ROOT, "wizard-work", "ledger", "2026-10-31-ledger-dir-census.csv") -OUT = sys.argv[1] if len(sys.argv) > 1 else ARTIFACT +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import verify_ledger_dir_census as census # noqa: E402 (sibling module) -def sha256_of(path): - h = hashlib.sha256() - with open(path, "rb") as f: - for chunk in iter(lambda: f.read(1024 * 1024), b""): - h.update(chunk) - return h.hexdigest() - -rows = [] -for dirpath, dirnames, filenames in os.walk(HERE): - dirnames.sort() - for fn in sorted(filenames): - full = os.path.join(dirpath, fn) - if os.path.abspath(full) == os.path.abspath(OUT): - continue # do not include the artifact being built - rel = os.path.relpath(full, REPO_ROOT) - rows.append((rel, str(os.path.getsize(full)), sha256_of(full))) -rows.sort(key=lambda r: r[0]) - -os.makedirs(os.path.dirname(OUT), exist_ok=True) -with open(OUT, "w", newline="", encoding="utf-8") as f: - w = csv.writer(f) - w.writerow(["rel_path", "size_bytes", "sha256"]) - w.writerows(rows) - -print(f"wrote {len(rows)} rows -> {OUT}") +if __name__ == "__main__": + argv = list(sys.argv[1:]) + where = [] + if argv and not argv[0].startswith("-"): + out = argv.pop(0) + where = ["--csv", out] + if out.endswith(".csv"): + where += ["--manifest", out[:-4] + ".manifest.json"] + sys.exit(census.main(["--write", *where, *argv])) diff --git a/wizard-work/ledger/build_dir_census.py b/wizard-work/ledger/build_dir_census.py index a97ddd8..610a609 100644 --- a/wizard-work/ledger/build_dir_census.py +++ b/wizard-work/ledger/build_dir_census.py @@ -1,60 +1,30 @@ #!/usr/bin/env python3 -# build_dir_census.py -- regenerate the ledger dir census CSV (extract lane). +# build_dir_census.py -- regenerate the ledger dir census (extract lane). # -# Schema (must match verify_ledger_dir_census.py): -# rel_path,size_bytes,sha256 -# rel_path is relative to the troupe-standup repo root, sorted ascending. -# The artifact itself is excluded from BOTH the re-derivation and the -# committed file, so the check is self-consistent across regeneration. +# This is a wrapper, deliberately: the derivation lives in +# verify_ledger_dir_census.py so the builder and the checker cannot disagree +# about what the census contains. The census is the TRACKED population of +# wizard-work/ledger at a NAMED REV -- not the working tree, not the host -- so +# the same rows in any clone. Ignored +# `__pycache__/*.pyc` files are outside the population by construction; they +# exist on the machine that ran Python, not in the repository. # -# Consumes (real tool output): the live filesystem of wizard-work/ledger/. -# Produces: the committed artifact at +# Writes: # wizard-work/ledger/2026-10-31-ledger-dir-census.csv -# and the primary deliverable at -# /root/.hermes/wizard-outputs/ledger/2026-10-31-ledger-dir-census.csv +# rel_path,size_bytes,sha256, sorted by rel_path, the CSV and manifest +# themselves excluded +# wizard-work/ledger/2026-10-31-ledger-dir-census.manifest.json +# the pinned rev, its subject, the schema, the population rule, row count +# +# Usage: +# python3 wizard-work/ledger/build_dir_census.py # pin at HEAD +# python3 wizard-work/ledger/build_dir_census.py --rev import os -import csv -import hashlib +import sys -HERE = os.path.dirname(os.path.abspath(__file__)) # wizard-work/ledger -REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) # troupe-standup root -SRC = HERE -OUT = "/root/troupe-standup/wizard-work/ledger/2026-10-31-ledger-dir-census.csv" - - -def sha256_of(path): - h = hashlib.sha256() - with open(path, "rb") as f: - for chunk in iter(lambda: f.read(1024 * 1024), b""): - h.update(chunk) - return h.hexdigest() - - -def re_derive(): - rows = [] - for dirpath, dirnames, filenames in os.walk(SRC): - dirnames.sort() - for fn in sorted(filenames): - full = os.path.join(dirpath, fn) - if os.path.abspath(full) == os.path.abspath(OUT): - continue - rel = os.path.relpath(full, REPO_ROOT) - size = os.path.getsize(full) - rows.append((rel, size, sha256_of(full))) - rows.sort(key=lambda r: r[0]) - return rows - - -def main(): - rows = re_derive() - os.makedirs(os.path.dirname(OUT), exist_ok=True) - with open(OUT, "w", newline="", encoding="utf-8") as f: - w = csv.writer(f) - w.writerow(["rel_path", "size_bytes", "sha256"]) - w.writerows(rows) - print(f"rows: {len(rows)}") - print(f"wrote: {OUT}") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import verify_ledger_dir_census as census # noqa: E402 (sibling module) if __name__ == "__main__": - main() + sys.exit(census.main(["--write", *sys.argv[1:]])) diff --git a/wizard-work/ledger/verify_ledger_dir_census.py b/wizard-work/ledger/verify_ledger_dir_census.py index e9d3cc1..1902111 100644 --- a/wizard-work/ledger/verify_ledger_dir_census.py +++ b/wizard-work/ledger/verify_ledger_dir_census.py @@ -1,62 +1,488 @@ #!/usr/bin/env python3 -# verify_ledger_dir_census.py -# Real test: re-derive the live inventory of wizard-work/ledger from the -# filesystem and compare byte-for-byte with the committed artifact. Exits 0 -# iff identical. The artifact CSV is excluded from both sides so the check is -# self-consistent across regeneration. -import hashlib, os, csv, sys +"""verify_ledger_dir_census.py -- grade the ledger directory census against a +pinned rev, over the TRACKED population, never the working tree. -HERE = os.path.dirname(os.path.abspath(__file__)) # wizard-work/ledger -REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) # troupe-standup root -ARTIFACT = os.path.join(REPO_ROOT, "wizard-work", "ledger", "2026-10-31-ledger-dir-census.csv") -SRC = HERE +WHAT IT GRADES + `wizard-work/ledger/2026-10-31-ledger-dir-census.csv` claims to be an + inventory of the ledger directory: `rel_path,size_bytes,sha256`, sorted by + path. This checker re-derives that inventory from the repository at the rev + named in `2026-10-31-ledger-dir-census.manifest.json` and compares row for + row. The subject is the blob at `:`, read through `git + cat-file`; the working tree and the host are not consulted, so a clean clone + and the machine that wrote the CSV read the same bytes. -def re_derive(): - rows = [] - for dirpath, dirnames, filenames in os.walk(SRC): - dirnames.sort() - for fn in sorted(filenames): - full = os.path.join(dirpath, fn) - if os.path.abspath(full) == os.path.abspath(ARTIFACT): - continue - rel = os.path.relpath(full, REPO_ROOT) - size = str(os.path.getsize(full)) - h = hashlib.sha256() - with open(full, "rb") as f: - for chunk in iter(lambda: f.read(1024 * 1024), b""): - h.update(chunk) - rows.append((rel, size, h.hexdigest())) +WHY THE REV, AND WHY TRACKED ONLY + The first version of this checker walked `os.walk("wizard-work/ledger")` and + compared against the live filesystem. Two consequences, both measured on + 2026-10-31 (Q140/Q141): the population contained 20 ignored + `__pycache__/*.pyc` files, so a clean clone re-derived 574 to 580 rows + against 594 committed; and the derivation moved whenever any Python run + recompiled bytecode or any lane rewrote the CSV, so the same commit returned + two verdicts. A census a reader cannot reproduce from the repository is not + evidence. Tracked blobs at a named rev are reproducible by anyone who can + clone it. + +WHAT "PINNED" MEANS + The manifest's `rev` is the tree the inventory describes. It is taken when + the CSV is written (`--write`) and it does not follow HEAD: other lanes add + files to `wizard-work/ledger/` constantly, and a census that reds on every + such commit grades the committer, not the census. Staleness is reported as + an ADVISORY line (added / removed / changed rows between the pin and HEAD) + and never as a fault. Moving the pin is a deliberate act: re-run `--write` + and commit the CSV, the manifest and the queue record together. + +EXIT CODES + 0 the committed CSV is the derivation at the pinned rev + 1 it is not (drift, missing rows, extra rows, or a manifest/CSV + disagreement; every count is printed) + 2 could not grade (rev unresolvable, CSV or manifest unreadable or + malformed, git failed) -- ungradable is not a pass and is not printed as + one + +USAGE + python3 wizard-work/ledger/verify_ledger_dir_census.py # = --check + python3 wizard-work/ledger/verify_ledger_dir_census.py --check + python3 wizard-work/ledger/verify_ledger_dir_census.py --write [--rev REV] + python3 wizard-work/ledger/verify_ledger_dir_census.py --json + python3 wizard-work/ledger/verify_ledger_dir_census.py --self-test +""" +import argparse +import csv +import hashlib +import json +import os +import subprocess +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_REPO = os.path.abspath(os.path.join(HERE, "..", "..")) +CENSUS_DIR = "wizard-work/ledger" +CSV_REL = CENSUS_DIR + "/2026-10-31-ledger-dir-census.csv" +MANIFEST_REL = CENSUS_DIR + "/2026-10-31-ledger-dir-census.manifest.json" +SCHEMA = ["rel_path", "size_bytes", "sha256"] +EXCLUDE = {CSV_REL, MANIFEST_REL} + + +class Ungradable(Exception): + """Raised when the rev or an artifact cannot be read: exit 2, not exit 1.""" + + +def git(repo, *args, stdin=None): + proc = subprocess.run(["git", "-C", repo, *args], input=stdin, + stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if proc.returncode != 0: + raise Ungradable("git %s -> rc %d: %s" % ( + " ".join(args), proc.returncode, + proc.stderr.decode("utf-8", "replace").strip()[:400])) + return proc.stdout + + +def resolve_rev(repo, rev): + """Full commit sha for `rev`, or Ungradable. Never silently falls back.""" + if rev is None or not str(rev).strip(): + raise Ungradable("no rev given and the manifest names none") + try: + out = git(repo, "rev-parse", "--verify", "%s^{commit}" % rev) + except Ungradable as exc: + raise Ungradable("rev %r does not resolve to a commit: %s" % (rev, exc)) + sha = out.decode().strip() + if len(sha) != 40: + raise Ungradable("rev %r resolved to %r, not a full sha" % (rev, sha)) + return sha + + +def blob_rows(repo, rev): + """[(rel_path, size, sha256)] for tracked blobs under CENSUS_DIR at rev.""" + raw = git(repo, "ls-tree", "-r", "-l", "-z", rev, "--", CENSUS_DIR) + wanted = [] + for chunk in raw.split(b"\x00"): + if not chunk.strip(): + continue + meta, _, path = chunk.partition(b"\t") + parts = meta.decode().split() + if len(parts) < 4: # mode type sha size + continue + if parts[1] != "blob": # a gitlink is not a file + continue + rel = path.decode("utf-8", "surrogateescape") + if rel in EXCLUDE: + continue + wanted.append((parts[2], rel)) + if not wanted: + raise Ungradable("no tracked blobs under %s at %s" % (CENSUS_DIR, rev)) + # one git cat-file --batch for the whole population: byte-exact content + query = ("\n".join(sha for sha, _ in wanted) + "\n").encode() + out = git(repo, "cat-file", "--batch", stdin=query) + rows, pos = [], 0 + for sha, rel in wanted: + nl = out.index(b"\n", pos) + header = out[pos:nl].decode("utf-8", "replace").split() + if len(header) < 3 or header[1] != "blob": + raise Ungradable("cat-file did not return a blob for %s (%s)" + % (rel, header)) + n = int(header[2]) + body = out[nl + 1:nl + 1 + n] + rows.append((rel, len(body), hashlib.sha256(body).hexdigest())) + pos = nl + 1 + n + 1 # content, then the batch separator newline rows.sort(key=lambda r: r[0]) return rows -def read_committed(path): - with open(path, newline="", encoding="utf-8") as f: - r = csv.reader(f) - header = next(r) - assert header == ["rel_path", "size_bytes", "sha256"], f"bad header {header}" - return [tuple(x) for x in r] -committed = read_committed(ARTIFACT) -derived = re_derive() +def read_csv(path): + if not os.path.exists(path): + raise Ungradable("no CSV at %s" % path) + with open(path, "r", newline="", encoding="utf-8") as fh: + reader = csv.reader(fh) + try: + header = next(reader) + except StopIteration: + raise Ungradable("CSV at %s is empty" % path) + if header != SCHEMA: + raise Ungradable("bad header %r in %s (want %r)" + % (header, path, SCHEMA)) + rows = [] + for n, row in enumerate(reader, start=2): + if not row: + continue + if len(row) != 3: + raise Ungradable("line %d of %s has %d fields, want 3" + % (n, path, len(row))) + try: + size = int(row[1]) + except ValueError: + raise Ungradable("line %d of %s has a non-integer size %r" + % (n, path, row[1])) + rows.append((row[0], size, row[2])) + return rows -print(f"artifact: {ARTIFACT}") -print(f"rows committed: {len(committed)} rows re-derived: {len(derived)}") -if committed == derived: - print("PASS: committed artifact matches live re-derivation.") - sys.exit(0) +def write_csv(path, rows): + tmp = path + ".tmp" + with open(tmp, "w", newline="", encoding="utf-8") as fh: + writer = csv.writer(fh) + writer.writerow(SCHEMA) + writer.writerows(rows) + os.replace(tmp, path) -print("FAIL: mismatch (committed vs re-derived).") -n = min(len(committed), len(derived)) -for i in range(n): - if committed[i] != derived[i]: - print(f" first diff at row {i}:") - print(f" committed: {committed[i]}") - print(f" derived: {derived[i]}") - break -else: - if len(committed) < len(derived): - print(f" derived has {len(derived)-len(committed)} extra rows") + +def read_manifest(path): + if not os.path.exists(path): + raise Ungradable("no manifest at %s" % path) + with open(path, "r", encoding="utf-8") as fh: + try: + man = json.load(fh) + except json.JSONDecodeError as exc: + raise Ungradable("manifest at %s is not JSON: %s" % (path, exc)) + if not isinstance(man, dict): + raise Ungradable("manifest at %s is not an object" % path) + for key in ("rev", "rows", "schema", "population"): + if key not in man: + raise Ungradable("manifest at %s has no %r" % (path, key)) + return man + + +def write_manifest(path, rev, subject, n_rows): + man = { + "artifact": CSV_REL, + "rev": rev, + "rev_subject": subject, + "schema": SCHEMA, + "population": "tracked blobs under %s at rev, self-excluded " + "(the CSV and this manifest)" % CENSUS_DIR, + "rows": n_rows, + "generated_by": CENSUS_DIR + "/verify_ledger_dir_census.py --write", + } + with open(path + ".tmp", "w", encoding="utf-8") as fh: + json.dump(man, fh, indent=2, sort_keys=True) + fh.write("\n") + os.replace(path + ".tmp", path) + + +def compare(committed, derived): + """Counts in both directions, plus the first names in each fault class.""" + by_c = {p: (s, h) for p, s, h in committed} + by_d = {p: (s, h) for p, s, h in derived} + drift, extra, missing = [], [], [] + match = 0 + for path, value in by_c.items(): + if path not in by_d: + extra.append(path) + elif by_d[path] == value: + match += 1 + else: + drift.append(path) + missing = [path for path in by_d if path not in by_c] + return { + "match": match, "drift": len(drift), "extra": len(extra), + "missing": len(missing), "committed": len(committed), + "derived": len(derived), + "drift_paths": sorted(drift)[:5], "extra_paths": sorted(extra)[:5], + "missing_paths": sorted(missing)[:5], + } + + +def do_check(repo, csv_path, manifest_path, rev_override): + manifest = read_manifest(manifest_path) + committed = read_csv(csv_path) + rev = resolve_rev(repo, rev_override or manifest["rev"]) + derived = blob_rows(repo, rev) + counts = compare(committed, derived) + + faults = [] + if counts["drift"]: + faults.append("count drift moved 0 -> %d" % counts["drift"]) + if counts["extra"]: + faults.append("count extra moved 0 -> %d" % counts["extra"]) + if counts["missing"]: + faults.append("count missing moved 0 -> %d" % counts["missing"]) + if manifest["rows"] != len(committed): + faults.append("manifest rows %s != CSV rows %d" + % (manifest["rows"], len(committed))) + if list(manifest["schema"]) != SCHEMA: + faults.append("manifest schema %r != %r" % (manifest["schema"], SCHEMA)) + pin = resolve_rev(repo, manifest["rev"]) + if pin != rev: + faults.append("rev moved: manifest pins %s, graded %s" + % (pin[:12], rev[:12])) + + try: + head = resolve_rev(repo, "HEAD") + if head == rev: + advisory = ("currency: HEAD %s is the pinned rev -- 0 added, " + "0 removed, 0 changed" % head[:12]) + else: + delta = compare(committed, blob_rows(repo, head)) + advisory = ("currency: HEAD %s is past the pin %s -- %d added, " + "%d removed, %d changed (advisory only: a census " + "describes the rev it names)" + % (head[:12], rev[:12], delta["missing"], + delta["extra"], delta["drift"])) + except Ungradable as exc: + advisory = "currency: SKIP (%s)" % exc + + return {"rev": rev, "counts": counts, "faults": faults, + "advisory": advisory} + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--repo", default=DEFAULT_REPO) + ap.add_argument("--rev", default=None) + ap.add_argument("--csv", default=None) + ap.add_argument("--manifest", default=None) + ap.add_argument("--check", action="store_true") + ap.add_argument("--write", action="store_true") + ap.add_argument("--json", action="store_true") + ap.add_argument("--self-test", action="store_true") + args = ap.parse_args(argv) + + if args.self_test: + return self_test() + + repo = os.path.abspath(args.repo) + csv_path = os.path.abspath(args.csv or os.path.join(repo, CSV_REL)) + manifest_path = os.path.abspath(args.manifest or os.path.join(repo, MANIFEST_REL)) + + if args.write: + try: + rev = resolve_rev(repo, args.rev or "HEAD") + rows = blob_rows(repo, rev) + subject = git(repo, "log", "-1", "--format=%s", rev).decode( + "utf-8", "replace").strip() + os.makedirs(os.path.dirname(csv_path), exist_ok=True) + write_csv(csv_path, rows) + write_manifest(manifest_path, rev, subject, len(rows)) + except Ungradable as exc: + print("UNGRADABLE: %s" % exc) + return 2 + print("wrote %d rows -> %s" % (len(rows), csv_path)) + print("pinned rev %s (%s)" % (rev, subject[:60])) + print("wrote %s" % manifest_path) + return 0 + + try: + result = do_check(repo, csv_path, manifest_path, args.rev) + except Ungradable as exc: + print("UNGRADABLE: %s" % exc) + return 2 + + counts = result["counts"] + if args.json: + print(json.dumps({ + "rev": result["rev"], + "committed": counts["committed"], "derived": counts["derived"], + "match": counts["match"], "drift": counts["drift"], + "extra": counts["extra"], "missing": counts["missing"], + "faults": result["faults"], "advisory": result["advisory"], + }, indent=2)) else: - print(f" committed has {len(committed)-len(derived)} extra rows") -sys.exit(1) + print("subject: %s at %s (tracked blobs, CSV and manifest excluded)" + % (CENSUS_DIR, result["rev"])) + print("rows committed: %d rows derived: %d" + % (counts["committed"], counts["derived"])) + print("match %d drift %d extra %d missing %d" + % (counts["match"], counts["drift"], counts["extra"], + counts["missing"])) + for label in ("drift", "extra", "missing"): + for path in counts[label + "_paths"]: + print(" %s: %s" % (label, path)) + if result["advisory"]: + print(result["advisory"]) + + if result["faults"]: + for fault in result["faults"]: + print("FAIL: %s" % fault) + print("VERDICT: rc 1 -- %d fault(s)" % len(result["faults"])) + return 1 + print("VERDICT: rc 0 -- the committed CSV is the derivation at the pinned rev") + return 0 + + +# --------------------------------------------------------------------------- +# self-test: cells over synthetic repositories, so every leg is reachable +# --------------------------------------------------------------------------- +def _run(repo, *argv): + proc = subprocess.run([sys.executable, os.path.abspath(__file__), *argv], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + return proc.returncode, proc.stdout.decode("utf-8", "replace") + + +def _init_repo(tmp): + env = dict(os.environ) + env.update({"GIT_AUTHOR_NAME": "selftest", + "GIT_AUTHOR_EMAIL": "selftest@example.invalid", + "GIT_COMMITTER_NAME": "selftest", + "GIT_COMMITTER_EMAIL": "selftest@example.invalid"}) + subprocess.run(["git", "init", "-q", tmp], check=True, env=env) + return env + + +def _commit(tmp, files, msg, env): + for rel, body in files.items(): + full = os.path.join(tmp, rel) + os.makedirs(os.path.dirname(full), exist_ok=True) + with open(full, "wb") as fh: + fh.write(body) + subprocess.run(["git", "-C", tmp, "add", "-A"], check=True, env=env) + subprocess.run(["git", "-C", tmp, "commit", "-q", "-m", msg], + check=True, env=env) + return subprocess.run(["git", "-C", tmp, "rev-parse", "HEAD"], + capture_output=True, text=True, + env=env).stdout.strip() + + +def self_test(): + import shutil + import tempfile + + cells = [] + + def cell(name, ok, detail=""): + cells.append((name, ok, detail)) + + base = tempfile.mkdtemp(prefix="ledger-census-selftest-") + try: + tmp = os.path.join(base, "repo") + env = _init_repo(tmp) + rev_a = _commit(tmp, { + "wizard-work/ledger/a.txt": b"alpha\n", + "wizard-work/ledger/sub/b.txt": b"bravo\n", + "outside/ignore.txt": b"not censused\n", + }, "A", env) + csvp = os.path.join(tmp, CSV_REL) + man = os.path.join(tmp, MANIFEST_REL) + + rc, out = _run(tmp, "--repo", tmp, "--write") + cell("write-two-rows", rc == 0 and "wrote 2 rows" in out, + out.strip().replace("\n", " | ")[-160:]) + cell("pin-names-rev-a", rev_a in open(man, encoding="utf-8").read(), + "manifest rev is %s" % rev_a[:12]) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("green-control", rc == 0, "rc=%d" % rc) + + # the Q140 property: ignored bytecode that exists only on this host + pyc = os.path.join(tmp, "wizard-work/ledger/__pycache__/a.cpython-314.pyc") + os.makedirs(os.path.dirname(pyc), exist_ok=True) + with open(pyc, "wb") as fh: + fh.write(b"\x00bytecode on this host only\x00") + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("host-bytecode-invisible", rc == 0 and "rows derived: 2" in out, + "rc=%d" % rc) + + # an uncommitted edit of a graded file is not the subject either + with open(os.path.join(tmp, "wizard-work/ledger/a.txt"), "wb") as fh: + fh.write(b"dirty worktree\n") + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("dirty-worktree-invisible", rc == 0, "rc=%d" % rc) + + # a committed byte change under a moved pin is green ... + keep_man = open(man, "rb").read() + keep_csv = open(csvp, "rb").read() + rev_b = _commit(tmp, {"wizard-work/ledger/a.txt": b"alpha changed\n"}, + "B", env) + rc, out = _run(tmp, "--repo", tmp, "--write", "--rev", rev_b) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("repinned-green", rc == 0, "pin moved to %s, rc=%d" + % (rev_b[:12], rc)) + + # ... and the same CSV under the old pin is drift + with open(man, "wb") as fh: + fh.write(keep_man) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("pinned-rev-drift-reds", + rc == 1 and "drift moved 0 -> 1" in out, "rc=%d" % rc) + + # a row naming a path that is not at the rev + with open(csvp, "wb") as fh: + fh.write(keep_csv) + with open(csvp, "a", encoding="utf-8") as fh: + fh.write("wizard-work/ledger/ghost.txt,7,%s\n" % ("0" * 64)) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("extra-row-reds", rc == 1 and "extra moved 0 -> 1" in out, + "rc=%d" % rc) + + # a blob at the rev with no row + with open(csvp, "wb") as fh: + fh.write(keep_csv) + lines = keep_csv.decode().splitlines() + with open(csvp, "w", encoding="utf-8") as fh: + fh.write("\n".join([lines[0]] + lines[2:]) + "\n") + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("missing-row-reds", rc == 1 and "missing moved 0 -> 1" in out, + "rc=%d" % rc) + + # a manifest whose row count disagrees with the CSV it names + with open(csvp, "wb") as fh: + fh.write(keep_csv) + lied = json.loads(keep_man.decode()) + lied["rows"] = lied["rows"] + 1 + with open(man, "w", encoding="utf-8") as fh: + json.dump(lied, fh) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("manifest-count-lie-reds", rc == 1 and "manifest rows" in out, + "rc=%d" % rc) + + # a rev that does not exist is ungradable, never a pass + bogus = json.loads(keep_man.decode()) + bogus["rev"] = "0" * 40 + with open(man, "w", encoding="utf-8") as fh: + json.dump(bogus, fh) + rc, out = _run(tmp, "--repo", tmp, "--check") + cell("unknown-rev-ungradable", rc == 2 and "UNGRADABLE" in out, + "rc=%d" % rc) + except Exception as exc: # noqa: BLE001 + cell("selftest-harness", False, "%s: %s" % (type(exc).__name__, exc)) + finally: + shutil.rmtree(base, ignore_errors=True) + + failed = [c for c in cells if not c[1]] + for name, ok, detail in cells: + print("%s %s%s" % ("PASS" if ok else "FAIL", name, + "" if ok else " -- " + detail)) + print("self-test: %d cells, %d passed, %d failed" + % (len(cells), len(cells) - len(failed), len(failed))) + return 0 if not failed else 1 + + +if __name__ == "__main__": + sys.exit(main())