85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify the Hermes profile census artifact (Extract lane).
|
|
|
|
Checks:
|
|
1. The CSV header is exactly: profile,archives,is_ledger
|
|
2. Every data row has 3 fields, numeric archives/is_ledger
|
|
3. The profile set and counts match a fresh walk of /root/.hermes/profiles
|
|
4. is_ledger flag is correct for every row
|
|
|
|
Exit code 0 = correct, nonzero = wrong.
|
|
"""
|
|
import csv
|
|
import os
|
|
import sys
|
|
|
|
DEFAULT_CSV = "/root/.hermes/wizard-outputs/ledger/profiles-census.csv"
|
|
PROFILES_DIR = "/root/.hermes/profiles"
|
|
|
|
|
|
def expected_rows():
|
|
rows = {}
|
|
for name in os.listdir(PROFILES_DIR):
|
|
pdir = os.path.join(PROFILES_DIR, name)
|
|
if not os.path.isdir(pdir):
|
|
continue
|
|
n = 0
|
|
for root, dirs, files in os.walk(pdir):
|
|
for f in files:
|
|
if f.endswith(".txt"):
|
|
n += 1
|
|
rows[name] = (n, 1 if name == "ledger" else 0)
|
|
return rows
|
|
|
|
|
|
def main():
|
|
csv_path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CSV
|
|
with open(csv_path, newline="") as fh:
|
|
reader = csv.reader(fh)
|
|
rows = list(reader)
|
|
|
|
errors = []
|
|
|
|
if not rows or rows[0] != ["profile", "archives", "is_ledger"]:
|
|
errors.append("bad header: %r" % (rows[0] if rows else None))
|
|
|
|
data = rows[1:]
|
|
if not data:
|
|
errors.append("no data rows")
|
|
|
|
expected = expected_rows()
|
|
if len(data) != len(expected):
|
|
errors.append("row count %d != expected %d" % (len(data), len(expected)))
|
|
|
|
for r in data:
|
|
if len(r) != 3:
|
|
errors.append("row has %d fields: %r" % (len(r), r))
|
|
continue
|
|
name, a, l = r
|
|
if not (a.isdigit() and l.isdigit()):
|
|
errors.append("non-numeric fields in row: %r" % r)
|
|
continue
|
|
want = expected.get(name)
|
|
if want is None:
|
|
errors.append("unexpected profile %r" % name)
|
|
elif (int(a), int(l)) != want:
|
|
errors.append("mismatch for %r: got (%s,%s) want %s" % (name, a, l, want))
|
|
|
|
extra = set(expected) - {r[0] for r in data}
|
|
for name in sorted(extra):
|
|
errors.append("missing profile %r" % name)
|
|
|
|
if errors:
|
|
print("FAIL")
|
|
for e in errors:
|
|
print(" -", e)
|
|
sys.exit(1)
|
|
|
|
total = sum(int(r[1]) for r in data)
|
|
print("OK: %d profiles, %d total txts, ledger=%s" % (len(data), total, [r for r in data if r[2] == "1"]))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|