stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ledger: grade the dir census at a pinned rev over tracked blobs (Q141 tooling)

Browse Source
This commit is contained in:
volt committed 2026-10-31 21:23:41 +00:00
1 parent 3747f17817
commit 40921f3ff6
4 files changed
+528 -143
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHECK-ledger-dir-census
+7 -7
View File
@@ -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)"
wizard-work/ledger/build_2026-10-31_ledger_dir_census.py
+23 -34
View File
@@ -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]))
wizard-work/ledger/build_dir_census.py
+21 -51
View File
@@ -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 <sha>
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:]]))
wizard-work/ledger/verify_ledger_dir_census.py
+477 -51
View File
@@ -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 `<rev>:<path>`, 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())