103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify the finished GitHub `python` org repos table (evidence test).
|
|
|
|
Input: /root/.hermes/wizard-outputs/ledger/python_org_repos_top50.csv
|
|
Output: PASS line + exit 0, or FAIL + exit 1.
|
|
|
|
Checks:
|
|
A. Schema/handshake of the stored CSV (exact header, 50 rows, unique
|
|
repo names, ranks 1..50, integer counts >= 0, stars non-increasing).
|
|
B. Stored top-50 equals a fresh fetch of the live GitHub API
|
|
(independent source provenance, not the extractor's own output).
|
|
"""
|
|
import csv
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
|
|
CSV_PATH = "/root/.hermes/wizard-outputs/ledger/python_org_repos_top50.csv"
|
|
ORG = "python"
|
|
PER_PAGE = 100
|
|
N_TOP = 50
|
|
EXPECTED_HEADER = ["rank", "repo", "owner", "description", "language",
|
|
"stars", "forks", "open_issues", "updated_at"]
|
|
HEADERS = {
|
|
"Accept": "application/vnd.github+json",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
"User-Agent": "ledger-verify-python-org-repos/1.0",
|
|
}
|
|
|
|
|
|
def fetch_repos():
|
|
page, all_repos = 1, []
|
|
while True:
|
|
url = "https://api.github.com/orgs/%s/repos?per_page=%d&page=%d" % (ORG, PER_PAGE, page)
|
|
req = urllib.request.Request(url, headers=HEADERS)
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
all_repos.extend(json.loads(r.read()))
|
|
link = r.getheader("Link") or ""
|
|
if 'rel="next"' not in link:
|
|
break
|
|
page += 1
|
|
return all_repos
|
|
|
|
|
|
def main():
|
|
try:
|
|
with open(CSV_PATH, newline="", encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
header = reader.fieldnames
|
|
rows = list(reader)
|
|
except Exception as e:
|
|
print("FAIL A: cannot read CSV: %s" % e)
|
|
sys.exit(1)
|
|
|
|
if header != EXPECTED_HEADER:
|
|
print("FAIL A: header mismatch: %r" % (header,))
|
|
sys.exit(1)
|
|
if len(rows) != N_TOP:
|
|
print("FAIL A: expected %d rows, got %d" % (N_TOP, len(rows)))
|
|
sys.exit(1)
|
|
|
|
names = [r["repo"] for r in rows]
|
|
if len(set(names)) != N_TOP:
|
|
print("FAIL A: duplicate repo names")
|
|
sys.exit(1)
|
|
if [int(r["rank"]) for r in rows] != list(range(1, N_TOP + 1)):
|
|
print("FAIL A: ranks not contiguous 1..%d" % N_TOP)
|
|
sys.exit(1)
|
|
for r in rows:
|
|
for c in ("stars", "forks", "open_issues"):
|
|
try:
|
|
v = int(r[c])
|
|
except (TypeError, ValueError):
|
|
print("FAIL A: %s not an integer: %r" % (c, r[c]))
|
|
sys.exit(1)
|
|
if v < 0:
|
|
print("FAIL A: %s < 0: %s" % (c, v))
|
|
sys.exit(1)
|
|
stars = [int(r["stars"]) for r in rows]
|
|
if stars != sorted(stars, reverse=True):
|
|
print("FAIL A: stars not sorted non-increasing")
|
|
sys.exit(1)
|
|
|
|
# Source provenance: recompute the top-50 from a fresh live fetch.
|
|
live = fetch_repos()
|
|
live_sorted = sorted(live, key=lambda r: (-int(r.get("stargazers_count") or 0),
|
|
r["full_name"] or ""))
|
|
live_top = live_sorted[:N_TOP]
|
|
stored_top = [(r["repo"], int(r["stars"]), r["updated_at"]) for r in rows]
|
|
live_top_t = [(r["name"], int(r["stargazers_count"] or 0), r["updated_at"])
|
|
for r in live_top]
|
|
if stored_top != live_top_t:
|
|
print("FAIL B: stored table differs from a fresh live API fetch")
|
|
sys.exit(1)
|
|
|
|
print("OK: %d rows verified against a fresh live GitHub API fetch (%d total repos fetched)"
|
|
% (len(rows), len(live)))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|