volt
2556dbb0aa
Measured at 67f94cc: 19 of the 20 tracked tools answered a --json target inside a nonexistent directory with an uncaught FileNotFoundError at rc 1. Only queue_append_audit.py already refused (rc 2). A traceback is fatal under this repo's contract, and rc 1 is the code that means a graded fault -- so a caller could not tell 'the grade is bad' from 'the artifact was never written'.
Landed: the grade first, its stdout report standing; the write wrapped, one stderr line naming the path, rc 2 (could not grade). gate_blame.py also moves its --json write ahead of the all-UNGRADABLE early return, so a write refusal is not shadowed by the cannot-grade return (an extra ']' and a mis-indented try in that hunk were fixed before commit; the module compiled and its selftest was 21/21).
tools/write_probe.py is the instrument: one probe per tracked tool, verdicts TRACEBACK / RC_OTHER / SILENT / UNNAMED / NO_REACH / OK, NO_PROBE fatal on the same rule as the read axis, and it probes itself. It is leg 6 of verify.py, and selftest_sweep.py now carries both a read probe and a malformed probe for it.
Verification at this commit follows in the next commit with docs/TOOLS.md and the queue record.
490 lines
22 KiB
Python
490 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
append_only_audit.py -- measure the repo-wide append-only invariant from git objects.
|
|
|
|
THE INVARIANT, stated so it can be falsified. Two tests are applied to every
|
|
(commit, path) change, because the ledger's rule ("append to another party's file,
|
|
never overwrite it") is one claim and a stricter byte rule is a different one:
|
|
|
|
BYTE TEST pure append iff the parent blob's bytes are an exact prefix of the
|
|
child blob's bytes and the child is strictly longer.
|
|
classes: NEW / IDENTICAL / APPEND / EDIT
|
|
|
|
LINE TEST non-destructive iff every parent line still occurs in the child, in
|
|
order (the parent line sequence is an order-preserving subsequence of
|
|
the child's). This is the rule the ledger actually states: no existing
|
|
text is lost. An insertion at a mid-file anchor breaks the BYTE test
|
|
and passes the LINE test; a rewrite or deletion fails both.
|
|
classes: NO_PARENT / IDENTICAL / SUBSEQUENCE / LOSS
|
|
LOSS rows report the 1-based index of the first parent line that is
|
|
gone, how many parent lines are unmatched, and that line's text.
|
|
|
|
UNITS AND WHAT COUNTS AS EVIDENCE. Blob bytes are read through
|
|
`git cat-file --batch` (binary-safe); sizes come from the object, never from a
|
|
text-mode `git show`, which folds CRLF and under-reports size by the number of
|
|
CR pairs. Line counts are of the blob's own lines.
|
|
|
|
ONE INSTRUMENT, ONE HALF OF THE RULE. This tool answers "was an existing line
|
|
replaced?" and prints that count as CANDIDATE ROWS, never as violations: the
|
|
rule forbids ANOTHER PARTY's line, and a LOSS row does not carry an author. The
|
|
second half -- whose line it was -- is `tools/append_only_owners.py`, which reads
|
|
this tool's --json and adjudicates each row on the remover's identity. At
|
|
f5403b2 the candidates were 74 and the adjudication read 65 same-party / 9
|
|
other-party. Neither number is the other's restatement.
|
|
|
|
OUTPUT: --json PATH writes the full per-change record; stdout is a per-file table
|
|
plus the LOSS list restricted to append-sanctioned paths. Read-only: the only write
|
|
is the --json file.
|
|
|
|
USAGE:
|
|
python3 append_only_audit.py --repo /root/troupe-standup --json out.json
|
|
python3 append_only_audit.py --selftest # synthetic repo, 18 cells
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from collections import Counter, defaultdict
|
|
|
|
EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" # git's well-known empty tree
|
|
ESCAPE_RE = re.compile(rb"\\u([0-9a-fA-F]{4})")
|
|
|
|
# Paths the ledger's sharing rule covers (append to another party's file, never
|
|
# overwrite it). Everything else is a file its owner may rewrite freely.
|
|
APPEND_SANCTIONED = ("WORK-QUEUE.md", "checkins/", "replies/", "recoveries/",
|
|
"audits/", "inbox/")
|
|
|
|
|
|
def is_sanctioned(path):
|
|
return any(path == p or path.startswith(p) for p in APPEND_SANCTIONED)
|
|
|
|
|
|
def run(cmd, cwd, binary=False, check=True):
|
|
kw = {} if binary else {"text": True}
|
|
p = subprocess.run(cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kw)
|
|
if check and p.returncode != 0:
|
|
raise RuntimeError("%r failed rc=%d: %s" % (cmd, p.returncode, p.stderr[:400]))
|
|
return p.stdout if p.returncode == 0 else b""
|
|
|
|
|
|
class BlobReader:
|
|
"""Binary-safe streaming blob reader over a persistent `git cat-file --batch`."""
|
|
|
|
def __init__(self, repo):
|
|
self.proc = subprocess.Popen(
|
|
["git", "cat-file", "--batch"],
|
|
cwd=repo, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
)
|
|
|
|
def get(self, oid):
|
|
if oid is None:
|
|
return b""
|
|
self.proc.stdin.write((oid + "\n").encode())
|
|
self.proc.stdin.flush()
|
|
header = self.proc.stdout.readline().decode().strip()
|
|
parts = header.split()
|
|
if len(parts) == 2 and parts[1] == "missing":
|
|
raise RuntimeError("blob %s missing" % oid)
|
|
if len(parts) < 3:
|
|
raise RuntimeError("bad batch header: %r" % header)
|
|
size = int(parts[2])
|
|
data = self.proc.stdout.read(size)
|
|
self.proc.stdout.read(1) # trailing newline
|
|
return data
|
|
|
|
def close(self):
|
|
try:
|
|
self.proc.stdin.close()
|
|
except Exception:
|
|
pass
|
|
self.proc.terminate()
|
|
|
|
|
|
def first_parent(repo, sha):
|
|
out = run(["git", "rev-parse", "--verify", "--quiet", sha + "^1"], repo,
|
|
binary=True, check=False)
|
|
return out.decode().strip() or None
|
|
|
|
|
|
def changed_paths(repo, parent, sha):
|
|
"""[(status, path)] for the first-parent diff; a root commit makes every path A."""
|
|
base = parent if parent else EMPTY_TREE
|
|
out = run(["git", "diff", "--name-status", "-M", "-z", base, sha], repo, binary=True)
|
|
fields = [f.decode("utf-8", "replace") for f in out.split(b"\x00") if f]
|
|
rows, i = [], 0
|
|
while i < len(fields):
|
|
st = fields[i]
|
|
if st.startswith("R") or st.startswith("C"):
|
|
rows.append((st, fields[i + 1], fields[i + 2]))
|
|
i += 3
|
|
else:
|
|
rows.append((st, fields[i + 1]))
|
|
i += 2
|
|
return rows
|
|
|
|
|
|
def blob_oid(repo, sha, path):
|
|
out = run(["git", "rev-parse", "--verify", "--quiet", "%s:%s" % (sha, path)],
|
|
repo, binary=True, check=False)
|
|
return out.decode().strip() or None
|
|
|
|
|
|
def byte_class(parent_bytes, child_bytes):
|
|
"""Pure byte append iff parent is a strict byte prefix of child."""
|
|
if not parent_bytes:
|
|
return "NEW", 0
|
|
if parent_bytes == child_bytes:
|
|
return "IDENTICAL", len(parent_bytes)
|
|
n = min(len(parent_bytes), len(child_bytes))
|
|
cpl = 0
|
|
while cpl < n and parent_bytes[cpl] == child_bytes[cpl]:
|
|
cpl += 1
|
|
if cpl == len(parent_bytes) and len(child_bytes) > len(parent_bytes):
|
|
return "APPEND", cpl
|
|
return "EDIT", cpl
|
|
|
|
|
|
def line_test(parent_bytes, child_bytes):
|
|
"""(class, first_order_break_1based, that_line_text, missing_parent_lines, added_child_lines)
|
|
|
|
class LOSS means the parent's line sequence is not an order-preserving
|
|
subsequence of the child's: a line that existed is no longer there in order.
|
|
first_order_break is where the greedy scan lost the thread (a locator).
|
|
missing_parent_lines is the exact multiset difference |parent - child|: the
|
|
number of parent lines that do not occur anywhere in the child. The two are
|
|
different units and are reported separately (the 7th line after a break can
|
|
still be present in the child).
|
|
"""
|
|
if not parent_bytes:
|
|
return "NO_PARENT", None, "", 0, 0
|
|
pl = parent_bytes.split(b"\n")
|
|
cl = child_bytes.split(b"\n")
|
|
if pl == cl:
|
|
return "IDENTICAL", None, "", 0, 0
|
|
missing = sum((Counter(pl) - Counter(cl)).values())
|
|
added = sum((Counter(cl) - Counter(pl)).values())
|
|
i = 0
|
|
for ln in cl:
|
|
if i < len(pl) and ln == pl[i]:
|
|
i += 1
|
|
if i == len(pl):
|
|
return "SUBSEQUENCE", None, "", 0, added
|
|
return "LOSS", i + 1, pl[i].decode("utf-8", "replace")[:120], missing, added
|
|
|
|
|
|
def escape_unescape(b):
|
|
"""Replace literal backslash-uXXXX escape text with the character it denotes.
|
|
|
|
One commit in this history (c597a133, Merlin 2026-10-15) is exactly this: 29
|
|
literal `\\u2014`-style escapes in WORK-QUEUE.md and 43 in a check-in were
|
|
rewritten as the characters they denote. That replaces existing lines without
|
|
changing what they say, so it must be told apart from text actually lost.
|
|
"""
|
|
return ESCAPE_RE.sub(lambda m: chr(int(m.group(1), 16)).encode("utf-8"), b)
|
|
|
|
|
|
def is_line_subsequence(a_bytes, b_bytes):
|
|
al = a_bytes.split(b"\n")
|
|
bl = b_bytes.split(b"\n")
|
|
i = 0
|
|
for ln in bl:
|
|
if i < len(al) and ln == al[i]:
|
|
i += 1
|
|
return i == len(al)
|
|
|
|
|
|
def loss_kind(parent_bytes, child_bytes):
|
|
"""ESCAPE_NORMALISATION if unescaping the parent alone explains the change,
|
|
else REWRITE (text that existed is not there in the same form for any other
|
|
reason). Only meaningful for rows the line test classed LOSS."""
|
|
if is_line_subsequence(escape_unescape(parent_bytes), child_bytes):
|
|
return "ESCAPE_NORMALISATION"
|
|
return "REWRITE"
|
|
|
|
|
|
def audit(repo, progress=False):
|
|
head = run(["git", "rev-parse", "HEAD"], repo).strip()
|
|
shas = run(["git", "rev-list", "--reverse", "HEAD"], repo).split()
|
|
reader = BlobReader(repo)
|
|
rows, per_file = [], defaultdict(lambda: {
|
|
"changes": 0, "NEW": 0, "APPEND": 0, "IDENTICAL": 0, "EDIT": 0,
|
|
"NO_PARENT": 0, "SUBSEQUENCE": 0, "LOSS": 0, "renames": 0, "losses": [], "edits": [],
|
|
"first_seen": None, "last_seen": None,
|
|
})
|
|
cache = {}
|
|
try:
|
|
for n, sha in enumerate(shas):
|
|
parent = first_parent(repo, sha)
|
|
meta = run(["git", "log", "-1", "--format=%H%x09%an%x09%ad%x09%s",
|
|
"--date=iso-strict", sha], repo).rstrip("\n").split("\t")
|
|
for row in changed_paths(repo, parent, sha):
|
|
if row[0] == "D":
|
|
continue
|
|
newpath = row[-1]
|
|
is_rename = row[0].startswith("R")
|
|
p_oid = blob_oid(repo, parent, newpath) if parent else None
|
|
c_oid = blob_oid(repo, sha, newpath)
|
|
if p_oid and p_oid not in cache:
|
|
cache[p_oid] = reader.get(p_oid)
|
|
if c_oid and c_oid not in cache:
|
|
cache[c_oid] = reader.get(c_oid)
|
|
pb = cache.get(p_oid, b"")
|
|
cb = cache.get(c_oid, b"")
|
|
bcls, cpl = byte_class(pb, cb)
|
|
if is_rename:
|
|
bcls = "RENAME"
|
|
lcls, brk_no, brk_txt, miss_p, add_c = line_test(pb, cb)
|
|
kind = loss_kind(pb, cb) if lcls == "LOSS" else ""
|
|
rec = {
|
|
"commit": sha, "author": meta[1], "date": meta[2],
|
|
"subject": meta[3] if len(meta) > 3 else "",
|
|
"path": newpath, "status": row[0],
|
|
"byte_class": bcls, "common_prefix": cpl,
|
|
"line_class": lcls, "loss_kind": kind,
|
|
"first_order_break_line": brk_no,
|
|
"order_break_line_text": brk_txt,
|
|
"missing_parent_lines": miss_p, "added_child_lines": add_c,
|
|
"parent_bytes": len(pb), "child_bytes": len(cb),
|
|
}
|
|
rows.append(rec)
|
|
agg = per_file[newpath]
|
|
agg["changes"] += 1
|
|
if bcls == "RENAME":
|
|
agg["renames"] += 1
|
|
else:
|
|
agg[bcls] += 1
|
|
agg[lcls] += 1
|
|
if bcls == "EDIT":
|
|
agg["edits"].append({k: rec[k] for k in
|
|
("commit", "author", "date", "parent_bytes",
|
|
"child_bytes", "common_prefix", "status")})
|
|
if lcls == "LOSS":
|
|
agg["LOSS_" + kind] = agg.get("LOSS_" + kind, 0) + 1
|
|
agg["losses"].append({k: rec[k] for k in
|
|
("commit", "author", "date", "parent_bytes",
|
|
"child_bytes", "loss_kind",
|
|
"first_order_break_line",
|
|
"order_break_line_text", "missing_parent_lines",
|
|
"added_child_lines", "status")})
|
|
agg["first_seen"] = agg["first_seen"] or meta[2]
|
|
agg["last_seen"] = meta[2]
|
|
if progress and (n + 1) % 200 == 0:
|
|
print(" ... %d/%d commits" % (n + 1, len(shas)), file=sys.stderr)
|
|
if len(cache) > 4000:
|
|
cache = {k: v for k, v in cache.items() if len(v) < 200000}
|
|
finally:
|
|
reader.close()
|
|
return {"repo": repo, "head": head, "commits": len(shas), "rows": rows,
|
|
"per_file": dict(per_file),
|
|
"sanctioned_definition": list(APPEND_SANCTIONED)}
|
|
|
|
|
|
def report(res, out=sys.stdout):
|
|
rows, per_file = res["rows"], res["per_file"]
|
|
bcount, lcount = defaultdict(int), defaultdict(int)
|
|
for r in rows:
|
|
bcount[r["byte_class"]] += 1
|
|
lcount[r["line_class"]] += 1
|
|
print("append-only audit -- repo %s" % res["repo"], file=out)
|
|
print("HEAD %s ; %d commits ; %d (commit,path) changes" %
|
|
(res["head"], res["commits"], len(rows)), file=out)
|
|
print("BYTE test : " + " ".join("%s=%d" % (k, bcount[k]) for k in
|
|
("NEW", "APPEND", "IDENTICAL", "EDIT", "RENAME")), file=out)
|
|
print("LINE test : " + " ".join("%s=%d" % (k, lcount[k]) for k in
|
|
("NO_PARENT", "SUBSEQUENCE", "IDENTICAL", "LOSS")), file=out)
|
|
print(" (NEW in the byte test == NO_PARENT in the line test, two names for one rule)",
|
|
file=out)
|
|
print("", file=out)
|
|
clean = sorted(p for p, a in per_file.items() if a["LOSS"] == 0)
|
|
lossy = sorted((p, a) for p, a in per_file.items() if a["LOSS"] > 0)
|
|
print("files with ZERO LOSS rows (no existing line ever lost): %d of %d"
|
|
% (len(clean), len(per_file)), file=out)
|
|
print("", file=out)
|
|
print("files whose history contains LOSS rows (an existing line changed or removed):",
|
|
file=out)
|
|
for p, a in sorted(lossy, key=lambda kv: (-kv[1]["LOSS"], kv[0])):
|
|
flag = " <-- APPEND-SANCTIONED" if is_sanctioned(p) else ""
|
|
print(" %-52s LOSS=%-3d (of %d changes) byte-EDIT=%-3d%s"
|
|
% (p, a["LOSS"], a["changes"], a["EDIT"], flag), file=out)
|
|
print("", file=out)
|
|
bad = [(p, a) for p, a in lossy if is_sanctioned(p)]
|
|
if bad:
|
|
total = sum(a["LOSS"] for _, a in bad)
|
|
nesc = sum(a.get("LOSS_ESCAPE_NORMALISATION", 0) for _, a in bad)
|
|
nrew = sum(a.get("LOSS_REWRITE", 0) for _, a in bad)
|
|
print("CANDIDATE ROWS on append-sanctioned paths (%d file(s), %d LOSS rows)." % (len(bad), total), file=out)
|
|
print(" A CANDIDATE is not a violation: the rule forbids ANOTHER PARTY's line, and a", file=out)
|
|
print(" LOSS row only says a line was replaced. tools/append_only_owners.py adjudicates", file=out)
|
|
print(" each row on the remover's identity; at f5403b2 it read 65 same-party / 9 other.", file=out)
|
|
print(" of those: %d fully explained by unicode-escape normalisation (the parent's"
|
|
% nesc, file=out)
|
|
print(" lines reappear unchanged once literal \\uXXXX text is decoded); %d are REWRITE —"
|
|
% nrew, file=out)
|
|
print(" text that existed does not reappear in the same form for any encoding reason.",
|
|
file=out)
|
|
for p, a in sorted(bad, key=lambda kv: -kv[1]["LOSS"]):
|
|
print("", file=out)
|
|
print(" %s (%d changes, %d byte-APPEND, %d byte-EDIT, %d LOSS)"
|
|
% (p, a["changes"], a["APPEND"], a["EDIT"], a["LOSS"]), file=out)
|
|
for e in a["losses"]:
|
|
print(" %s %s %s [%s]" % (e["commit"][:8], e["date"][:19],
|
|
e["author"], e["loss_kind"]), file=out)
|
|
print(" %d -> %d B ; parent lines gone: %d ; parent lines added: %d ; order break at line #%s"
|
|
% (e["parent_bytes"], e["child_bytes"], e["missing_parent_lines"],
|
|
e["added_child_lines"], e["first_order_break_line"]), file=out)
|
|
if e["order_break_line_text"]:
|
|
print(" line where the order scan broke: %r" % e["order_break_line_text"],
|
|
file=out)
|
|
else:
|
|
print("CANDIDATE ROWS on append-sanctioned paths: none", file=out)
|
|
print("", file=out)
|
|
print("terms: BYTE-APPEND = parent blob is a byte-exact prefix of child and child is",
|
|
file=out)
|
|
print("longer. LOSS = the parent's line sequence is not an order-preserving subsequence",
|
|
file=out)
|
|
print("of the child's, i.e. a line that existed is no longer there in order. A mid-file",
|
|
file=out)
|
|
print("insertion is byte-EDIT and line-SUBSEQUENCE: it moves later text without losing",
|
|
file=out)
|
|
print("any. Sizes come from git objects via `git cat-file --batch`.", file=out)
|
|
|
|
|
|
# --------------------------------------------------------------------------- selftest
|
|
|
|
def _g(repo, *args):
|
|
return subprocess.run(["git", "-C", repo] + list(args), check=True,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE).stdout.decode()
|
|
|
|
|
|
def selftest():
|
|
tmp = tempfile.mkdtemp(prefix="aoa-selftest-")
|
|
try:
|
|
repo = os.path.join(tmp, "r")
|
|
os.makedirs(repo)
|
|
_g(repo, "init", "-q", "-b", "main")
|
|
_g(repo, "config", "user.email", "t@test.invalid")
|
|
_g(repo, "config", "user.name", "T")
|
|
p = lambda *a: os.path.join(repo, *a)
|
|
cells = []
|
|
|
|
# --- synthetic history: append, mid-file insertion, in-place edit, deletion, rename
|
|
with open(p("a.txt"), "w") as f:
|
|
f.write("one\ntwo\n")
|
|
with open(p("b.txt"), "w") as f:
|
|
f.write("keep\n")
|
|
with open(p("mid.txt"), "w") as f:
|
|
f.write("head\nbody\ntail\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c1 new")
|
|
with open(p("a.txt"), "a") as f:
|
|
f.write("three\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c2 append")
|
|
with open(p("mid.txt"), "w") as f: # insert in the middle
|
|
f.write("head\nINSERTED\nbody\ntail\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c3 mid insertion")
|
|
with open(p("mid.txt"), "w") as f: # lose the word 'body'
|
|
f.write("head\nINSERTED\ntail\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c4 line loss")
|
|
with open(p("a.txt"), "w") as f: # in-place byte edit, same length
|
|
f.write("ONE\ntwo\nthree\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c5 in-place edit")
|
|
with open(p("c.txt"), "w") as f:
|
|
f.write("new file\n")
|
|
with open(p("d.txt"), "w") as f:
|
|
f.write("keep\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c6 new files")
|
|
os.rename(p("b.txt"), p("b2.txt"))
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c7 rename")
|
|
with open(p("esc.txt"), "w") as f: # literal escape text
|
|
f.write("dash \\u2014 here\nplain\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c8 escape text")
|
|
with open(p("esc.txt"), "w") as f: # rewritten as the character
|
|
f.write("dash \u2014 here\nplain\n")
|
|
_g(repo, "add", "-A"); _g(repo, "commit", "-qm", "c9 escape normalised")
|
|
|
|
res = audit(repo)
|
|
rows = res["rows"]
|
|
by = defaultdict(list)
|
|
for r in rows:
|
|
by[r["path"]].append(r)
|
|
|
|
def cls(path, commit_subject_prefix, key):
|
|
for r in rows:
|
|
if r["path"] == path and r["subject"].startswith(commit_subject_prefix):
|
|
return r[key]
|
|
return None
|
|
|
|
cells = [
|
|
("c1 a NEW byte", cls("a.txt", "c1", "byte_class") == "NEW"),
|
|
("c1 b NEW byte", cls("b.txt", "c1", "byte_class") == "NEW"),
|
|
("c2 append byte", cls("a.txt", "c2", "byte_class") == "APPEND"),
|
|
("c3 insertion is byte-EDIT", cls("mid.txt", "c3", "byte_class") == "EDIT"),
|
|
("c3 insertion is line-SUBSEQUENCE", cls("mid.txt", "c3", "line_class") == "SUBSEQUENCE"),
|
|
("c4 loss is byte-EDIT", cls("mid.txt", "c4", "byte_class") == "EDIT"),
|
|
("c4 loss flagged LOSS", cls("mid.txt", "c4", "line_class") == "LOSS"),
|
|
("c4 loss names break at line 3", cls("mid.txt", "c4", "first_order_break_line") == 3),
|
|
("c4 loss counts 1 missing parent line",
|
|
cls("mid.txt", "c4", "missing_parent_lines") == 1),
|
|
("c4 loss adds 0 child lines", cls("mid.txt", "c4", "added_child_lines") == 0),
|
|
("c4 break line text quoted", cls("mid.txt", "c4", "order_break_line_text") == "body"),
|
|
("c3 insertion adds 1 child line",
|
|
cls("mid.txt", "c3", "added_child_lines") == 1),
|
|
("c5 in-place byte edit (cpl 0)", any(
|
|
r["path"] == "a.txt" and r["byte_class"] == "EDIT" and r["common_prefix"] == 0
|
|
for r in rows)),
|
|
("c5 in-place edit is line-LOSS", cls("a.txt", "c5", "line_class") == "LOSS"),
|
|
("c7 rename detected", cls("b2.txt", "c7", "byte_class") == "RENAME"),
|
|
("c9 escape rewrite is LOSS", cls("esc.txt", "c9", "line_class") == "LOSS"),
|
|
("c9 classified ESCAPE_NORMALISATION",
|
|
cls("esc.txt", "c9", "loss_kind") == "ESCAPE_NORMALISATION"),
|
|
("c4 real loss is not called normalisation",
|
|
cls("mid.txt", "c4", "loss_kind") == "REWRITE"),
|
|
]
|
|
ok = 0
|
|
for name, passed in cells:
|
|
print("%-40s %s" % (name, "PASS" if passed else "FAIL"))
|
|
ok += bool(passed)
|
|
print("selftest: %d cells, %d passed, %d failed" % (len(cells), ok, len(cells) - ok))
|
|
return 0 if ok == len(cells) else 1
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--repo", default=".")
|
|
ap.add_argument("--json")
|
|
ap.add_argument("--selftest", action="store_true")
|
|
ap.add_argument("--progress", action="store_true")
|
|
a = ap.parse_args()
|
|
if a.selftest:
|
|
return selftest()
|
|
# A repo that is not there, or is not a git work tree, is a PRECONDITION
|
|
# failure: rc 2 with a named message. Before 2026-10-17 this path escaped
|
|
# as a FileNotFoundError traceback with rc 1 out of the first `git` call.
|
|
if not os.path.isdir(a.repo):
|
|
print("cannot read repo: %s is not a directory" % a.repo, file=sys.stderr)
|
|
return 2
|
|
try:
|
|
res = audit(a.repo, progress=a.progress)
|
|
except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
|
|
print("cannot read repo %s: %s" % (a.repo, exc), file=sys.stderr)
|
|
return 2
|
|
report(res)
|
|
if a.json:
|
|
try:
|
|
with open(a.json, "w") as f:
|
|
json.dump(res, f, indent=1, sort_keys=True)
|
|
except OSError as exc:
|
|
print("cannot write --json " + a.json + ": " + str(exc), file=sys.stderr)
|
|
return 2
|
|
print("\njson written: %s (%d bytes)" % (a.json, os.path.getsize(a.json)))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|