forge: landing_check.py DRY consolidation refactor (Counter single census pass) + equivalence verifier

This commit is contained in:
forge committed 2026-10-23 09:33:18 +00:00
1 parent af1542ca93
commit cf80507de7
3 files changed
+393 -11

No files matched your search

+14 -11
View File
@@ -65,6 +65,7 @@ import shutil
import subprocess
import sys
import tempfile
from collections import Counter
DEFAULT_BASE = ("https://forge.alexanderwhitestone.com/git/stackchain/"
"troupe-standup/raw")
@@ -193,14 +194,15 @@ def build_rows(repo, rev, branch, forge, paths=None, max_bytes=4 * 1024 * 1024):
def summarise(rows):
counts = {}
for row in rows:
counts[row["verdict"]] = counts.get(row["verdict"], 0) + 1
return counts
"""Verdict census as a {verdict: n} mapping -- one pass via Counter,
replacing the original's per-call dict.get(k,0)+1 loop."""
return Counter(row["verdict"] for row in rows)
def summary_line(rows):
counts = summarise(rows)
def summary_line(rows, counts=None):
"""One-line tally. Pass a precomputed census via `counts` to avoid
recomputing it; run() reuses its own census instead of re-scanning."""
counts = counts if counts is not None else summarise(rows)
order = ["OK", "DIFFERS", "MISSING", "HTTP_ERROR", "UNREACHABLE",
"READ_FAILED", "SKIPPED_LARGE"]
parts = ["%d %s" % (counts[k], k) for k in order if counts.get(k)]
@@ -212,6 +214,8 @@ def run(args):
rows = build_rows(args.repo, args.rev, args.branch,
Forge(args.base_url, args.timeout),
args.path, args.max_bytes)
counts = summarise(rows) # single census pass
fatal = sum(counts.get(k, 0) for k in FATAL) # derived once
if not args.quiet:
width = max([len("path")] + [len(r["path"]) for r in rows]) if rows else 4
print("%-*s %-12s %s" % (width, "path", "verdict", "detail"))
@@ -219,13 +223,12 @@ def run(args):
print("%-*s %-12s %s" % (width, row["path"], row["verdict"],
row["detail"]))
print()
line = summary_line(rows)
print(line)
print(summary_line(rows, counts)) # reuse the census
if args.json:
payload = {"rev": args.rev, "branch": args.branch,
"base_url": args.base_url, "rows": rows,
"summary": summarise(rows),
"fatal": sum(row["verdict"] in FATAL for row in rows)}
"summary": counts,
"fatal": fatal}
try:
with open(args.json, "w", encoding="utf-8") as fh:
json.dump(payload, fh, indent=2, sort_keys=True)
@@ -233,7 +236,7 @@ def run(args):
except OSError as exc:
print("cannot write --json " + args.json + ": " + str(exc), file=sys.stderr)
return 2
return 1 if any(row["verdict"] in FATAL for row in rows) else 0
return 1 if fatal else 0
def selftest():