123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""CHECK: re-query the Gitea API for the stackchain org's repos and diff the
|
|
produced tables against the committed artifact. Exit 0 only if they agree.
|
|
|
|
This is a real regression test, not a tautology: it re-reads the live source and
|
|
re-derives the rows, then compares against the committed artifact bytes.
|
|
"""
|
|
import csv
|
|
import json
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
|
|
API_BASE = "https://forge.alexanderwhitestone.com/api/v1"
|
|
ORG = "stackchain"
|
|
OUT_DIR = "/root/.hermes/wizard-outputs/ledger"
|
|
TODAY = "2026-10-21"
|
|
CSV_PATH = f"{OUT_DIR}/gitea-repos-{TODAY}.csv"
|
|
JSON_PATH = f"{OUT_DIR}/gitea-repos-{TODAY}.json"
|
|
|
|
HEADERS = {"User-Agent": "ledger/stackchain-repo-census-verify",
|
|
"Accept": "application/json"}
|
|
COLUMNS = [
|
|
"full_name", "description", "language", "size_kb", "open_issues_count",
|
|
"stargazers_count", "forks_count", "default_branch", "updated_at",
|
|
"created_at", "html_url", "ssh_url",
|
|
]
|
|
|
|
|
|
def api_get(path):
|
|
req = urllib.request.Request(API_BASE + path, headers=HEADERS)
|
|
with urllib.request.urlopen(req, timeout=60) as r:
|
|
return json.loads(r.read().decode("utf-8"))
|
|
|
|
|
|
def reextract():
|
|
raw = api_get(f"/orgs/{ORG}/repos?page=1&per_page=100")
|
|
if len(raw) < 9: # page 2 would only add more, but page 1 is complete here
|
|
pass
|
|
out = []
|
|
for repo in raw:
|
|
out.append({
|
|
"full_name": repo.get("full_name", ""),
|
|
"description": repo.get("description") or "",
|
|
"language": repo.get("language") or "",
|
|
"size_kb": round(repo.get("size", 0) / 1024, 1),
|
|
"open_issues_count": int(repo.get("open_issues_count", 0)),
|
|
"stargazers_count": int(repo.get("stargazers_count", 0)),
|
|
"forks_count": int(repo.get("forks_count", 0)),
|
|
"default_branch": repo.get("default_branch") or "",
|
|
"updated_at": repo.get("updated_at") or "",
|
|
"created_at": repo.get("created_at") or "",
|
|
"html_url": repo.get("html_url") or "",
|
|
"ssh_url": repo.get("ssh_url") or "",
|
|
})
|
|
return out
|
|
|
|
|
|
def load_csv():
|
|
rows = []
|
|
with open(CSV_PATH, newline="", encoding="utf-8") as f:
|
|
for row in csv.DictReader(f):
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def md_table_row_count(md):
|
|
n = 0
|
|
for line in md.splitlines():
|
|
if line.startswith("| ") and not line.startswith("| #") and "---" not in line:
|
|
n += 1
|
|
return n
|
|
|
|
|
|
def main():
|
|
errors = []
|
|
expected = reextract()
|
|
expected_sorted = sorted(expected, key=lambda x: x["full_name"])
|
|
|
|
try:
|
|
actual = load_csv()
|
|
except Exception as e:
|
|
print(f"FAIL: cannot read CSV: {e}")
|
|
return 1
|
|
|
|
if len(actual) != len(expected_sorted):
|
|
errors.append(f"row count CSV: expected {len(expected_sorted)} got {len(actual)}")
|
|
|
|
norm = lambda s: re.sub(r"\s+", " ", str(s if s is not None else "")).strip()
|
|
for a, e in zip(sorted(actual, key=lambda x: x["full_name"]),
|
|
expected_sorted):
|
|
for col in COLUMNS:
|
|
av, ev = norm(a.get(col)), norm(e.get(col))
|
|
if av != ev:
|
|
errors.append(f"{a.get('full_name')}::{col}: expected {ev!r} got {av!r}")
|
|
|
|
# JSON continuity check
|
|
try:
|
|
jrows = json.load(open(JSON_PATH, encoding="utf-8"))
|
|
if len(jrows) != len(expected):
|
|
errors.append(f"json row count: expected {len(expected)} got {len(jrows)}")
|
|
for a, e in zip(sorted(jrows, key=lambda x: x["full_name"]),
|
|
expected_sorted):
|
|
for col in COLUMNS:
|
|
if norm(a.get(col)) != norm(e.get(col)):
|
|
errors.append(f"{a.get('full_name')}::{col} json mismatch")
|
|
except Exception as e:
|
|
errors.append(f"json read failed: {e}")
|
|
|
|
if errors:
|
|
print("CHECK FAILED")
|
|
for e in errors:
|
|
print(" -", e)
|
|
return 1
|
|
|
|
print(f"CHECK PASS: {len(expected_sorted)} repos, all fields byte-identical "
|
|
f"(CSV + JSON)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|