Files
troupe-standup/wizard-work/ledger/verify_artifact.py
T

66 lines
2.4 KiB
Python

#!/usr/bin/env python3
"""Independent verification of the ledger tracked-file census artifact."""
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
CSV = "/root/troupe-standup/wizard-work/ledger/ledger-tracked-files.csv"
REPO = "/root/troupe-standup"
def git(*args):
return subprocess.run(["git", "-C", REPO] + list(args),
check=True, capture_output=True, text=True).stdout
def size_of(sha1):
return int(subprocess.run(["git", "cat-file", "-s", sha1],
check=True, capture_output=True, text=True).stdout.split()[0])
def main():
paths = git("ls-files").splitlines()
path_set = set(paths)
print(f"live tracked-file count: {len(paths)}", file=sys.stderr)
rows = open(CSV, encoding="utf-8").read().splitlines()
assert rows[0] == "row,path,mode,sha1,length_bytes,stage", rows[0]
data = rows[1:]
print(f"csv data rows: {len(data)}", file=sys.stderr)
assert len(data) == len(paths), "row count mismatch"
sha1_of_path = {}
for line in git("ls-files", "-s").splitlines():
head, sep, p = line.partition("\t")
if not sep:
continue
mode, sha1, stage = head.split()
sha1_of_path[p] = (sha1, mode, stage)
bad = []
with ThreadPoolExecutor(max_workers=16) as ex:
futs = {ex.submit(size_of, sha1_of_path[p][0]): p for p in path_set}
lengths = {futs[f]: f.result() for f in futs}
for i, line in enumerate(data, start=1):
parts = line.split(",")
if len(parts) != 6:
bad.append((i, "field-count", line)); continue
rrow, p, mode, sha1, lev, stage = parts
if int(rrow) != i: bad.append((i, "row-number", line))
if p not in sha1_of_path: bad.append((i, "path-not-in-index", line)); continue
esha1, emode, estage = sha1_of_path[p]
if sha1 != esha1 or mode != emode or stage != estage:
bad.append((i, "meta-mismatch", line))
if p not in lengths: bad.append((i, "no-length", line)); continue
if int(lev) != lengths[p]:
bad.append((i, "length-mismatch", line))
print(f"checked rows: {len(data)}", file=sys.stderr)
if bad:
print(f"FAILURES: {len(bad)}", file=sys.stderr)
for b in bad[:10]:
print(" ", b, file=sys.stderr)
sys.exit(1)
print("VERIFY_PASS: rows == git ls-files, meta and lengths match", file=sys.stderr)
if __name__ == "__main__":
main()