127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Ledger — independent re-derivation check for the git commit census.
|
|
|
|
Re-derives commits-census.csv and commits-by-day.csv cell-by-cell from git
|
|
at HEAD (not from the committed tables), then asserts the committed tables
|
|
match exactly. Also checks commits-full.csv has one row per commit.
|
|
|
|
Exit code is the verdict:
|
|
0 tables match git
|
|
1 mismatch
|
|
2 ungradable (git unreadable)
|
|
"""
|
|
import csv
|
|
import subprocess
|
|
import sys
|
|
|
|
REPO = "/root/troupe-standup"
|
|
OUT = "/root/troupe-standup/wizard-work/ledger"
|
|
|
|
# Pinned revision (must match extract_commits_census.py PIN).
|
|
PIN = "d8d8e3292db6c435b174e39a3ca18792874f9c19"
|
|
|
|
SEP = "\x1f"
|
|
FMT = "%H" + SEP + "%an" + SEP + "%ae" + SEP + "%aI" + SEP + "%s"
|
|
|
|
CENSUS = "commits-census.csv"
|
|
BY_DAY = "commits-by-day.csv"
|
|
FULL = "commits-full.csv"
|
|
|
|
|
|
def git_log():
|
|
cmd = ["git", "-C", REPO, "log",
|
|
"--pretty=format:" + FMT, "--date=iso-strict", PIN]
|
|
out = subprocess.run(cmd, check=True, capture_output=True, text=True).stdout
|
|
rows = []
|
|
for c in out.splitlines():
|
|
if not c:
|
|
continue
|
|
h, an, ae, ad, s = c.split(SEP)
|
|
rows.append({"hash": h, "author": an, "email": ae,
|
|
"author_date": ad, "subject": s})
|
|
return rows
|
|
|
|
|
|
def read_csv(name):
|
|
path = f"{OUT}/{name}"
|
|
with open(path, newline="", encoding="utf-8") as f:
|
|
return list(csv.reader(f))
|
|
|
|
|
|
def close(a, b):
|
|
try:
|
|
return float(a) == float(b)
|
|
except (TypeError, ValueError):
|
|
return a == b
|
|
|
|
|
|
def main():
|
|
rows = git_log()
|
|
total = len(rows)
|
|
by_author = {}
|
|
for r in rows:
|
|
by_author[r["author"]] = by_author.get(r["author"], 0) + 1
|
|
by_day = {}
|
|
for r in rows:
|
|
d = r["author_date"][:10]
|
|
by_day[d] = by_day.get(d, 0) + 1
|
|
|
|
problems = []
|
|
|
|
# --- commits-census.csv ---
|
|
try:
|
|
census = read_csv(CENSUS)
|
|
except Exception as e:
|
|
print("UNGRADABLE: cannot read commits-census.csv:", e)
|
|
return 2
|
|
if census[0] != ["author", "commits"]:
|
|
problems.append("census header: %r" % census[0])
|
|
expected_census = sorted(by_author.items(), key=lambda kv: (-kv[1], kv[0]))
|
|
if len(census) - 1 != len(expected_census):
|
|
problems.append("census rows %d != expected %d"
|
|
% (len(census) - 1, len(expected_census)))
|
|
for (act, exp) in zip(census[1:], expected_census):
|
|
if not close(act[1], str(exp[1])):
|
|
problems.append("census author %r commits %r != %r"
|
|
% (act[0], act[1], exp[1]))
|
|
|
|
# --- commits-by-day.csv ---
|
|
try:
|
|
days = read_csv(BY_DAY)
|
|
except Exception as e:
|
|
print("UNGRADABLE: cannot read commits-by-day.csv:", e)
|
|
return 2
|
|
if days[0] != ["date", "commits"]:
|
|
problems.append("by-day header: %r" % days[0])
|
|
expected_days = sorted(by_day.items())
|
|
if len(days) - 1 != len(expected_days):
|
|
problems.append("by-day rows %d != expected %d"
|
|
% (len(days) - 1, len(expected_days)))
|
|
for (act, exp) in zip(days[1:], expected_days):
|
|
if not close(act[1], str(exp[1])):
|
|
problems.append("by-day %s commits %r != %r"
|
|
% (act[0], act[1], exp[1]))
|
|
|
|
# --- commits-full.csv row count ---
|
|
try:
|
|
full = read_csv(FULL)
|
|
except Exception as e:
|
|
print("UNGRADABLE: cannot read commits-full.csv:", e)
|
|
return 2
|
|
if len(full) - 1 != total:
|
|
problems.append("full rows %d != commit count %d" % (len(full) - 1, total))
|
|
|
|
if problems:
|
|
print("FAIL (%d problems)" % len(problems))
|
|
for p in problems[:30]:
|
|
print(" -", p)
|
|
return 1
|
|
|
|
print("PASS: commits-census.csv and commits-by-day.csv match git at HEAD; "
|
|
"commits-full.csv has %d rows; total %d commits" % (len(full) - 1, total))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|