fix: make maintainer tool output actionable

--emulator accepts a buildbot core name from a profile's cores:, and a
mistyped target gets the closest names instead of every target. diff_truth
shows destinations and reports unpaired entries as evidence, not removals.
The verify summary counts discrepancies, a shared destination names both
systems, a truth hash conflict names both cores, and pipeline steps carry
their place in the real run instead of N/8. Every GitHub API caller takes
the token from github_token(), which falls back to gh auth token.
This commit is contained in:
Abdessamad Derraz committed 2026-10-21 04:42:01 +02:00
1 parent 6ee138a819
commit 7095aa61e7
29 files changed
+764 -154

No files matched your search

+6 -1
View File
@@ -1241,7 +1241,12 @@ def render(findings: list[Finding]) -> str:
def main() -> int:
parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.split("\n\n")[0])
parser.add_argument("--only", help="comma-separated areas: " + ",".join(AREAS))
# Joined with ", " so argparse wraps between names: one comma-joined
# token was cut inside "coreinfo".
parser.add_argument(
"--only", metavar="AREAS",
help="comma-separated areas, among: " + ", ".join(AREAS),
)
parser.add_argument("--json", action="store_true", help="machine-readable output")
parser.add_argument(
"--profiles", action="store_true",
+76 -13
View File
@@ -442,21 +442,72 @@ def list_registered_platforms(
SHORT_LIST = 12
def suggest_names(typed: list[str], available: list[str], listing_flag: str) -> str:
@functools.lru_cache(maxsize=1)
def _gh_cli_token() -> str:
"""What `gh auth token` prints, or "" when gh is absent or logged out."""
import logging # noqa: PLC0415
import subprocess # noqa: PLC0415
log = logging.getLogger(__name__)
try:
proc = subprocess.run(
["gh", "auth", "token"], capture_output=True, text=True,
timeout=10, check=False,
)
except FileNotFoundError:
log.debug("gh is not installed, GitHub API calls go anonymous")
return ""
except (subprocess.TimeoutExpired, OSError) as exc:
log.warning("gh auth token did not answer: %s", exc)
return ""
if proc.returncode != 0:
log.debug("gh auth token: %s", (proc.stderr or "").strip())
return ""
return (proc.stdout or "").strip()
def github_token() -> str:
"""The token for api.github.com: GITHUB_TOKEN, GH_TOKEN, then gh's own.
A maintainer logged in to gh had to export GITHUB_TOKEN by hand before
profile_sync, the scrapers or an online pipeline, which refused the run
or burnt the anonymous quota of 60 calls an hour. The token gh returns
is put in the environment, so the subprocesses a run starts and the
readers of os.environ get it too. RETROBIOS_OFFLINE runs make no API
call and never ask gh.
"""
token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") or ""
if token or os.environ.get("RETROBIOS_OFFLINE"):
return token
token = _gh_cli_token()
if token:
os.environ["GITHUB_TOKEN"] = token
return token
def suggest_names(
typed: list[str],
available: list[str],
listing_flag: str,
listed: int | None = None,
) -> str:
"""The closest known names to what was typed, and how to see them all.
The hints showed the first ten or twenty names of the alphabet behind a
"..." whatever was typed: sony-playstaton listed 3do to apple-iie.
`listed` is how many entries the listing shows when `available` also
holds aliases of them.
"""
import difflib # noqa: PLC0415
total = len(available) if listed is None else listed
close = sorted({m for name in typed
for m in difflib.get_close_matches(name, available, n=3, cutoff=0.6)})
if close:
return f"Did you mean: {', '.join(close)}? {listing_flag} lists all {len(available)}."
return f"Did you mean: {', '.join(close)}? {listing_flag} lists all {total}."
if len(available) <= SHORT_LIST:
return f"Known: {', '.join(available)}."
return f"{listing_flag} lists all {len(available)}."
return f"{listing_flag} lists all {total}."
def platform_arg(value: str) -> str:
@@ -603,16 +654,14 @@ def load_target_config(
canonical = alias_index.get(target)
if canonical is None:
available = sorted(targets.keys())
aliases = []
for tname, ovr in overrides.items():
for a in ovr.get("aliases", []):
aliases.append(f"{a} -> {tname}")
msg = f"Unknown target '{target}' for platform '{platform_name}'.\n"
msg += f"Available targets: {', '.join(available)}"
if aliases:
msg += f"\nAliases: {', '.join(sorted(aliases))}"
raise ValueError(msg)
# A typo got the 30 targets and 18 aliases in two lines to read
# through, where a platform or emulator name gets the closest ones.
raise ValueError(
f"Unknown target '{target}' for platform '{platform_name}'.\n"
+ suggest_names(
[target], sorted(alias_index), "--list-targets", len(targets)
)
)
declared = (targets[canonical] or {}).get("cores")
if declared is None:
@@ -1728,6 +1777,20 @@ def select_emulator_profiles(
"""
selected: list[tuple[str, dict]] = []
for name in profile_names:
if name not in all_profiles:
# A buildbot name (pcfx, advmame) lives in a profile's cores:,
# and the spelling hints sent pcfx to two PlayStation cores.
claimants = sorted(
key for key in upstream_profile_index(all_profiles).get(name, ())
if all_profiles[key].get("type") not in ("launcher", "test")
)
if len(claimants) == 1:
name = claimants[0]
elif claimants:
raise ProfileSelectionError(
f"{name} is a core name of {', '.join(claimants)}: "
f"use --emulator with one of them"
)
if name not in all_profiles:
available = sorted(
key
+10 -9
View File
@@ -855,16 +855,17 @@ def main() -> None:
require_platform(args.platform, args.platforms_dir)
profiles = load_emulator_profiles(args.emulators_dir)
if args.emulator:
emulator = args.emulator
if emulator:
# A typo answered "No emulator profiles found" with exit 0.
try:
select_emulator_profiles(
((emulator, _),) = select_emulator_profiles(
[args.emulator],
load_emulator_profiles(args.emulators_dir, skip_aliases=False),
)
except ProfileSelectionError as exc:
raise SystemExit(f"ERROR: {exc}") from None
profiles = {k: v for k, v in profiles.items() if k == args.emulator}
profiles = {k: v for k, v in profiles.items() if k == emulator}
if args.target and not args.platform:
parser.error("--target requires --platform")
@@ -897,14 +898,14 @@ def main() -> None:
))
if not profiles:
if args.emulator and args.target and set(args.emulator.split(",")) & run_anywhere:
if emulator and args.target and emulator in run_anywhere:
raise SystemExit(
f"ERROR: {args.emulator} is not built for {args.platform}'s "
f"ERROR: {emulator} is not built for {args.platform}'s "
f"{args.target} target; {args.platform} runs it elsewhere "
f"(drop --target to see its files)"
)
if args.emulator:
raise SystemExit(f"ERROR: {args.emulator} is not run by {args.platform}")
if emulator:
raise SystemExit(f"ERROR: {emulator} is not run by {args.platform}")
print(f"{args.platform or 'The tree'} runs no profiled emulator.", file=sys.stderr)
return
@@ -921,8 +922,8 @@ def main() -> None:
if args.json:
print(json.dumps(report, indent=2))
else:
if args.emulator:
scope = args.emulator
if emulator:
scope = emulator
elif args.platform:
scope = f"the emulators {args.platform} runs"
else:
+27 -13
View File
@@ -39,6 +39,11 @@ def _load_truth(truth_dir: str, platform: str) -> dict | None:
return yaml_load(f) or {}
def _where(entry: dict) -> str:
"""Where a diffed entry sits: one name can be declared at several paths."""
return entry.get("destination") or entry["name"]
def _format_terminal(report: dict) -> str:
lines: list[str] = []
platform = report.get("platform", "unknown")
@@ -56,7 +61,7 @@ def _format_terminal(report: dict) -> str:
if s["total_missing"]:
totals.append(f"{s['total_missing']} missing")
if s["total_extra_phantom"]:
totals.append(f"{s['total_extra_phantom']} phantom")
totals.append(f"{s['total_extra_phantom']} unpaired")
if s["total_extra_unprofiled"]:
totals.append(f"{s['total_extra_unprofiled']} unprofiled")
if s["total_hash_mismatch"]:
@@ -73,7 +78,7 @@ def _format_terminal(report: dict) -> str:
if div.get("missing"):
labels.append(f"MISSING:{len(div['missing'])}")
if div.get("extra_phantom"):
labels.append(f"PHANTOM:{len(div['extra_phantom'])}")
labels.append(f"UNPAIRED:{len(div['extra_phantom'])}")
if div.get("extra_unprofiled"):
labels.append(f"UNPROF:{len(div['extra_unprofiled'])}")
if div.get("hash_mismatch"):
@@ -84,19 +89,19 @@ def _format_terminal(report: dict) -> str:
for m in div.get("missing", []):
cores = ", ".join(m.get("cores", []))
lines.append(f" + {m['name']} [{cores}]")
lines.append(f" + {_where(m)} [{cores}]")
for h in div.get("hash_mismatch", []):
ht = h["hash_type"]
lines.append(
f" ~ {h['name']} {ht}: {h[f'truth_{ht}']} != {h[f'scraped_{ht}']}"
f" ~ {_where(h)} {ht}: {h[f'truth_{ht}']} != {h[f'scraped_{ht}']}"
)
for p in div.get("extra_phantom", []):
lines.append(f" - {p['name']} (phantom)")
lines.append(f" - {_where(p)} (no profile entry of this system matches it)")
for u in div.get("extra_unprofiled", []):
lines.append(f" ? {u['name']} (unprofiled)")
lines.append(f" ? {_where(u)} (unprofiled cores)")
for r in div.get("required_mismatch", []):
lines.append(
f" ! {r['name']} required: {r['truth_required']} != {r['scraped_required']}"
f" ! {_where(r)} required: {r['truth_required']} != {r['scraped_required']}"
)
uncovered = report.get("uncovered_systems", [])
@@ -121,7 +126,7 @@ def _format_markdown(report: dict) -> str:
)
lines.append(
f"**{s['total_missing']}** missing | "
f"**{s['total_extra_phantom']}** phantom | "
f"**{s['total_extra_phantom']}** unpaired | "
f"**{s['total_extra_unprofiled']}** unprofiled | "
f"**{s['total_hash_mismatch']}** hash | "
f"**{s['total_required_mismatch']}** required"
@@ -135,19 +140,28 @@ def _format_markdown(report: dict) -> str:
refs = ""
if m.get("source_refs"):
refs = " " + " ".join(f"`{r}`" for r in m["source_refs"])
lines.append(f"- **Add** `{m['name']}`{refs}")
lines.append(f"- **Add** `{_where(m)}`{refs}")
for h in div.get("hash_mismatch", []):
ht = h["hash_type"]
lines.append(
f"- **Fix hash** `{h['name']}` {ht}: `{h[f'truth_{ht}']}` != `{h[f'scraped_{ht}']}`"
f"- **Fix hash** `{_where(h)}` {ht}: `{h[f'truth_{ht}']}` != `{h[f'scraped_{ht}']}`"
)
# Evidence, not a request: an entry leaves a platform's file only on
# proof (export_native's removal_candidates). A core can read a
# file its profile does not pair with this declaration.
for p in div.get("extra_phantom", []):
lines.append(f"- **Remove** `{p['name']}` (phantom)")
lines.append(
f"- **Unpaired** `{_where(p)}`: no profile entry of this system matches it "
"by path, name or hash"
)
for u in div.get("extra_unprofiled", []):
lines.append(f"- **Check** `{u['name']}` (unprofiled cores)")
lines.append(
f"- **Unpaired** `{_where(u)}`: some cores of this platform "
"have no profile yet"
)
for r in div.get("required_mismatch", []):
lines.append(
f"- **Fix required** `{r['name']}`: truth={r['truth_required']}, scraped={r['scraped_required']}"
f"- **Fix required** `{_where(r)}`: truth={r['truth_required']}, scraped={r['scraped_required']}"
)
lines.append("")
+3 -1
View File
@@ -3087,8 +3087,10 @@ def _main(held: contextlib.ExitStack) -> None:
)
return
if args.list:
# Archived platforms build like the others but have no release pack.
active = set(list_registered_platforms(args.platforms_dir))
for p in list_platforms(args.platforms_dir):
print(p)
print(p if p in active else f"{p} (archived)")
return
if args.list_emulators:
list_emulator_profiles(args.emulators_dir)
+2 -1
View File
@@ -22,6 +22,7 @@ sys.path.insert(0, os.path.dirname(__file__))
from common import (
compute_composition,
count_catalog_matched,
github_token,
list_registered_platforms,
load_database,
load_emulator_profiles,
@@ -286,7 +287,7 @@ def fetch_contributors() -> list[dict]:
url = f"https://api.github.com/repos/{REPO}/contributors"
headers = {"User-Agent": "retrobios-readme/1.0"}
token = os.environ.get("GITHUB_TOKEN", "")
token = github_token()
if token:
headers["Authorization"] = f"token {token}"
try:
+5 -2
View File
@@ -285,8 +285,11 @@ def _served_digests() -> dict[str, str]:
f"{LARGE_FILES_RELEASE}"
)
headers = {"User-Agent": "retrobios/1.0", "Accept": "application/vnd.github+json"}
if os.environ.get("GITHUB_TOKEN"):
headers["Authorization"] = f"Bearer {os.environ['GITHUB_TOKEN']}"
from common import github_token # noqa: PLC0415 (common imports this module)
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
digests: dict[str, str] = {}
try:
with urllib.request.urlopen(
+8 -3
View File
@@ -7,7 +7,7 @@ Respects the `status` field in _registry.yml:
Usage:
python scripts/list_platforms.py # Active platforms (for CI)
python scripts/list_platforms.py --all # All platforms including archived
python scripts/list_platforms.py --include-archived # archived ones too
python scripts/list_platforms.py >> "$GITHUB_OUTPUT"
"""
@@ -31,10 +31,15 @@ def list_platforms(include_archived: bool = False) -> list[str]:
def main() -> None:
parser = argparse.ArgumentParser(allow_abbrev=False, description="List available platforms")
parser.add_argument("--all", action="store_true", help="Include archived platforms")
# The name verify, generate_pack, slots and pipeline use. --all is the
# older spelling, still read so existing commands keep working.
parser.add_argument(
"--include-archived", "--all", dest="include_archived", action="store_true",
help="include archived platforms",
)
args = parser.parse_args()
platforms = list_platforms(include_archived=args.all)
platforms = list_platforms(include_archived=args.include_archived)
if not platforms:
print("No platform configs found", file=sys.stderr)
+115 -47
View File
@@ -61,6 +61,7 @@ sys.path.insert(0, str(Path(__file__).parent))
from common import (
ArtifactLockBusy,
artifact_lock,
github_token,
hold_artifact_lock,
load_data_dir_registry,
)
@@ -92,6 +93,62 @@ class StepResults(dict):
# Steps that report without failing the run.
NON_BLOCKING = frozenset({"check_buildbot"})
# Every step in run order, by its results key. The position and the total
# printed during the run come from this list: hand-written "N/8" labels
# counted 8 steps in a run of 24, and the eighth came before 8b. The summary
# prints the same labels the run printed, so a FAILED row is found in the
# log by its number.
STEP_LABELS: dict[str, str] = {
"validate_schemas": "0 validate schemas",
"generate_db": "1 generate database",
"provenance": "1b provenance report",
"romset_recipes": "1c romset recipes",
"refresh_data": "2 refresh data directories",
"mame_hashes": "2a refresh MAME hashes",
"fbneo_hashes": "2a2 refresh FBNeo hashes",
"check_buildbot": "2b check buildbot system",
"check_release_assets": "2b2 check release assets",
"generate_truth": "2c generate truth",
"diff_truth": "2d diff truth",
"export_native": "2e export native",
"verify": "3 verify all platforms",
"identity": "3a file identity",
"slots": "3b slot arbitration",
"generate_packs": "4 generate packs",
"generate_manifests": "4b generate install manifests",
"generate_target_manifests": "4c generate target manifests",
"bundled_only": "4d copies only an emulator install provides",
"consistency": "5 consistency check",
"pack_integrity": "6 pack integrity",
"generate_readme": "7 generate readme",
"generate_site": "8 generate site",
"render_site": "8b render site",
}
_KEY_BY_LABEL = {label: key for key, label in STEP_LABELS.items()}
def step_number(key: str) -> str:
"""The number a step's progress lines carry ("2b2", "4d")."""
return STEP_LABELS[key].split(" ", 1)[0]
def step_title(label: str) -> str:
"""What the run prints for a label: its name and its place in the run.
A second command of one step ("2c generate truth (export model)") is
printed as written.
"""
key = _KEY_BY_LABEL.get(label)
if key is None:
return label
return f"{label} [{list(STEP_LABELS).index(key) + 1}/{len(STEP_LABELS)}]"
def skip_notice(key: str, reason: str = "") -> None:
"""The line a step that does not run prints in place of its output."""
because = f" ({reason})" if reason else ""
print(f"\n--- {step_title(STEP_LABELS[key])}: SKIPPED{because} ---")
def run(cmd: list[str], label: str) -> tuple[bool, str]:
"""Run a command, echoing its output as it comes. Returns (success, output).
@@ -99,7 +156,8 @@ def run(cmd: list[str], label: str) -> tuple[bool, str]:
after stdout: a twenty-minute pack build showed nothing, and a warning
landed far from the line it was about. One stream, read line by line.
"""
print(f"\n--- {label} ---", flush=True)
title = step_title(label)
print(f"\n--- {title} ---", flush=True)
start = time.monotonic()
repo_root = str(Path(__file__).resolve().parent.parent)
lines: list[str] = []
@@ -114,7 +172,7 @@ def run(cmd: list[str], label: str) -> tuple[bool, str]:
elapsed = time.monotonic() - start
ok = returncode == 0
print(f"--- {label}: {'OK' if ok else 'FAILED'} ({elapsed:.1f}s) ---")
print(f"--- {title}: {'OK' if ok else 'FAILED'} ({elapsed:.1f}s) ---")
return ok, "".join(lines)
@@ -248,7 +306,7 @@ def check_consistency(verify_output: str, pack_output: str) -> bool:
verify_extras = parse_verify_extras(verify_output)
pack_extras = parse_pack_extras(pack_output)
print("\n--- 5/8 consistency check ---")
print(f"\n--- {step_title(STEP_LABELS['consistency'])} ---")
all_ok = True
if not v or not p:
# Nothing parsed is nothing compared: a changed output format would
@@ -358,7 +416,7 @@ def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
]
if args.target:
truth_cmd.extend(["--target", args.target])
ok, _ = run(truth_cmd, "2c generate truth")
ok, _ = run(truth_cmd, STEP_LABELS["generate_truth"])
# A native file is one per platform, never per target: the export
# corrects it against the full model, which a targeted run does not
# write.
@@ -375,7 +433,7 @@ def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
diff_cmd.extend(["--truth-dir", str(truth_dir)])
if args.target:
diff_cmd.extend(["--target", args.target])
ok, _ = run(diff_cmd, "2d diff truth")
ok, _ = run(diff_cmd, STEP_LABELS["diff_truth"])
results["diff_truth"] = ok
all_ok = all_ok and ok
else:
@@ -397,7 +455,7 @@ def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
# file has to be in the cache already.
if not args.offline:
export_cmd.append("--fetch")
ok, _ = run(export_cmd, "2e export native")
ok, _ = run(export_cmd, STEP_LABELS["export_native"])
results["export_native"] = ok
all_ok = all_ok and ok
else:
@@ -455,6 +513,11 @@ def main() -> None:
)
args = parser.parse_args()
_refuse_unapplied_flags(parser, args)
if not args.offline:
# Read once and put in the environment every step inherits: the
# scrapers and refresh_data_dirs went anonymous unless the user had
# exported GITHUB_TOKEN by hand, though gh was logged in.
github_token()
# A second run on the same output directory is refused before any work:
# the database rebuild alone takes minutes, and the reader holding the
@@ -478,12 +541,12 @@ def main() -> None:
# `cores: [81]` unquoted passed the whole pipeline and failed CI on push.
# jsonschema is tooling only, so its absence skips the step.
if importlib.util.find_spec("jsonschema") is None:
print("\n--- 0/8 validate schemas: SKIPPED (jsonschema not installed) ---")
skip_notice("validate_schemas", "jsonschema not installed")
results["validate_schemas"] = SKIPPED
else:
ok, _ = run(
[sys.executable, "scripts/validate_schemas.py", "--source-only"],
"0/8 validate schemas",
STEP_LABELS["validate_schemas"],
)
results["validate_schemas"] = ok
all_ok = all_ok and ok
@@ -499,7 +562,7 @@ def main() -> None:
"--output",
"database.json",
],
"1/8 generate database",
STEP_LABELS["generate_db"],
)
results["generate_db"] = ok
if not ok:
@@ -512,7 +575,7 @@ def main() -> None:
ok, _ = run(
[sys.executable, "scripts/provenance_report.py", "--derive",
"--write-derivable", "derivable.json"],
"1b provenance report",
STEP_LABELS["provenance"],
)
results["provenance"] = ok
all_ok = all_ok and ok
@@ -522,7 +585,7 @@ def main() -> None:
# Read-only: writing reconstructions is an explicit --write invocation.
ok, out = run(
[sys.executable, "scripts/romset_recipes.py"],
"1c romset recipes",
STEP_LABELS["romset_recipes"],
)
results["romset_recipes"] = ok
all_ok = all_ok and ok
@@ -531,47 +594,49 @@ def main() -> None:
if not args.offline:
ok, out = run(
[sys.executable, "scripts/refresh_data_dirs.py"],
"2/8 refresh data directories",
STEP_LABELS["refresh_data"],
)
results["refresh_data"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2/8 refresh data directories: SKIPPED (--offline) ---")
skip_notice("refresh_data", "--offline")
results["refresh_data"] = SKIPPED
# Step 2a: Refresh MAME BIOS hashes
if not args.offline:
ok, _ = run(
[sys.executable, "-m", "scripts.scraper.mame_hash_scraper"],
"2a refresh MAME hashes",
STEP_LABELS["mame_hashes"],
)
results["mame_hashes"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2a refresh MAME hashes: SKIPPED (--offline) ---")
skip_notice("mame_hashes", "--offline")
results["mame_hashes"] = SKIPPED
# Step 2a2: Refresh FBNeo BIOS hashes
if not args.offline:
ok, _ = run(
[sys.executable, "-m", "scripts.scraper.fbneo_hash_scraper"],
"2a2 refresh FBNeo hashes",
STEP_LABELS["fbneo_hashes"],
)
results["fbneo_hashes"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2a2 refresh FBNeo hashes: SKIPPED (--offline) ---")
skip_notice("fbneo_hashes", "--offline")
results["fbneo_hashes"] = SKIPPED
# Step 2b: Check buildbot system directory (non-blocking)
if args.check_buildbot and not args.offline:
ok, _ = run(
[sys.executable, "scripts/check_buildbot_system.py"],
"2b check buildbot system",
STEP_LABELS["check_buildbot"],
)
results["check_buildbot"] = ok
elif args.check_buildbot:
print("\n--- 2b check buildbot system: SKIPPED (--offline) ---")
else:
if args.check_buildbot:
skip_notice("check_buildbot", "--offline")
results["check_buildbot"] = SKIPPED
# Step 2b2: The release must serve the bytes the manifests describe. The
# installer compares Content-Length with the manifest size, so an asset
@@ -579,12 +644,12 @@ def main() -> None:
if not args.offline:
ok, _ = run(
[sys.executable, "scripts/check_release_assets.py"],
"2b2 check release assets",
STEP_LABELS["check_release_assets"],
)
results["check_release_assets"] = ok
all_ok = all_ok and ok
else:
print("\n--- 2b2 check release assets: SKIPPED (--offline) ---")
skip_notice("check_release_assets", "--offline")
results["check_release_assets"] = SKIPPED
all_ok = _truth_steps(args, results) and all_ok
@@ -603,7 +668,7 @@ def main() -> None:
verify_cmd.append("--include-archived")
if args.target:
verify_cmd.extend(["--target", args.target])
ok, verify_output = run(verify_cmd, "3/8 verify all platforms")
ok, verify_output = run(verify_cmd, STEP_LABELS["verify"])
results["verify"] = ok
all_ok = all_ok and ok
@@ -612,7 +677,7 @@ def main() -> None:
# cannot see a file copied under another machine's name.
ok, _ = run(
[sys.executable, "scripts/identity.py", "--strict"],
"3a/8 file identity",
STEP_LABELS["identity"],
)
results["identity"] = ok
all_ok = all_ok and ok
@@ -629,7 +694,7 @@ def main() -> None:
slots_cmd.append("--include-archived")
if args.target:
slots_cmd.extend(["--target", args.target])
ok, _ = run(slots_cmd, "3b/8 slot arbitration")
ok, _ = run(slots_cmd, "3b slot arbitration")
results["slots"] = ok
all_ok &= ok
@@ -679,11 +744,11 @@ def main() -> None:
pack_cmd.extend(["--source", args.source])
if args.all_variants:
pack_cmd.append("--all-variants")
ok, pack_output = run(pack_cmd, "4/8 generate packs")
ok, pack_output = run(pack_cmd, STEP_LABELS["generate_packs"])
results["generate_packs"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4/8 generate packs: SKIPPED (--skip-packs) ---")
skip_notice("generate_packs", "--skip-packs")
results["generate_packs"] = SKIPPED
# Step 4b: Generate install manifests
@@ -706,11 +771,11 @@ def main() -> None:
manifest_cmd.append("--no-refresh-data")
if args.target:
manifest_cmd.extend(["--target", args.target])
ok, _ = run(manifest_cmd, "4b/8 generate install manifests")
ok, _ = run(manifest_cmd, STEP_LABELS["generate_manifests"])
results["generate_manifests"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4b/8 generate install manifests: SKIPPED (--skip-packs) ---")
skip_notice("generate_manifests", "--skip-packs")
results["generate_manifests"] = SKIPPED
# Step 4c: Generate target manifests
@@ -722,11 +787,11 @@ def main() -> None:
"--output-dir",
"install/targets",
]
ok, _ = run(target_cmd, "4c/8 generate target manifests")
ok, _ = run(target_cmd, STEP_LABELS["generate_target_manifests"])
results["generate_target_manifests"] = ok
all_ok = all_ok and ok
else:
print("\n--- 4c/8 generate target manifests: SKIPPED (--skip-packs) ---")
skip_notice("generate_target_manifests", "--skip-packs")
results["generate_target_manifests"] = SKIPPED
# Step 4d: A collected copy that only an emulator's own install provides.
@@ -735,7 +800,7 @@ def main() -> None:
# what an installation ships.
ok, _ = run(
[sys.executable, "scripts/bundled_only.py", "--strict"],
"4d/8 copies only an emulator install provides",
STEP_LABELS["bundled_only"],
)
results["bundled_only"] = ok
all_ok &= ok
@@ -745,21 +810,21 @@ def main() -> None:
# compare it with, and says so instead of failing on an empty parse.
full_pack_built = args.source == "full" or args.all_variants
if pack_output and verify_output and not full_pack_built:
print("\n--- 5/8 consistency check: SKIPPED (no full pack built) ---")
skip_notice("consistency", "no full pack built")
results["consistency"] = SKIPPED
elif pack_output and verify_output:
ok = check_consistency(verify_output, pack_output)
results["consistency"] = ok
all_ok = all_ok and ok
else:
print("\n--- 5/8 consistency check: SKIPPED ---")
skip_notice("consistency")
results["consistency"] = SKIPPED
# Step 6: Pack integrity (extract + hash verification)
if not args.skip_packs and not full_pack_built:
# The check reads the full pack's name, while a reduced pack holds fewer
# files than the platform declares by design.
print("\n--- 6/8 pack integrity: SKIPPED (no full pack built) ---")
skip_notice("pack_integrity", "no full pack built")
results["pack_integrity"] = SKIPPED
elif not args.skip_packs:
integrity_cmd = [
@@ -777,11 +842,11 @@ def main() -> None:
# the target removed as missing from a pack that was exactly right.
if args.target:
integrity_cmd.extend(["--target", args.target])
ok, _ = run(integrity_cmd, "6/8 pack integrity")
ok, _ = run(integrity_cmd, STEP_LABELS["pack_integrity"])
results["pack_integrity"] = ok
all_ok = all_ok and ok
else:
print("\n--- 6/8 pack integrity: SKIPPED (--skip-packs) ---")
skip_notice("pack_integrity", "--skip-packs")
results["pack_integrity"] = SKIPPED
# Step 7: Generate README
@@ -795,36 +860,36 @@ def main() -> None:
"--platforms-dir",
"platforms",
],
"7/8 generate readme",
STEP_LABELS["generate_readme"],
)
results["generate_readme"] = ok
all_ok = all_ok and ok
else:
print("\n--- 7/8 generate readme: SKIPPED (--skip-docs) ---")
skip_notice("generate_readme", "--skip-docs")
results["generate_readme"] = SKIPPED
# Step 8: Generate site pages
if not args.skip_docs:
ok, _ = run(
[sys.executable, "scripts/generate_site.py"],
"8/8 generate site",
STEP_LABELS["generate_site"],
)
results["generate_site"] = ok
all_ok = all_ok and ok
else:
print("\n--- 8/8 generate site: SKIPPED (--skip-docs) ---")
skip_notice("generate_site", "--skip-docs")
results["generate_site"] = SKIPPED
# Step 8b: the render the site deploy runs. Generating the pages without
# rendering them let 20 pages with duplicate ids through to CI.
if args.skip_docs:
print("\n--- 8b render site: SKIPPED (--skip-docs) ---")
skip_notice("render_site", "--skip-docs")
results["render_site"] = SKIPPED
elif shutil.which("mkdocs") is None:
print("\n--- 8b render site: SKIPPED (mkdocs not installed) ---")
skip_notice("render_site", "mkdocs not installed")
results["render_site"] = SKIPPED
else:
ok, _ = run(["mkdocs", "build", "--strict"], "8b render site")
ok, _ = run(["mkdocs", "build", "--strict"], STEP_LABELS["render_site"])
if ok:
ok, _ = run([sys.executable, "scripts/validate_site.py"], "8b validate site")
results["render_site"] = ok
@@ -835,8 +900,11 @@ def main() -> None:
print(f"\n{'=' * 60}")
warned = []
for step, outcome in results.items():
# The number finds the step's lines in the log, the key is the
# name a caller reading the results knows it by.
number = step_number(step)
if outcome is SKIPPED:
print(f" {step:.<40} SKIPPED")
print(f" {number:<4} {step:.<40} SKIPPED")
continue
label = "OK" if outcome else "FAILED"
if not outcome and step in NON_BLOCKING:
@@ -844,8 +912,8 @@ def main() -> None:
# contradiction: the step reports, it does not stop the run.
label = "WARN"
warned.append(step)
print(f" {step:.<40} {label:<7}{results.seconds[step]:>8.1f}s")
print(f" {'total':.<40} {'':<7}{total_elapsed:>8.1f}s")
print(f" {number:<4} {step:.<40} {label:<7}{results.seconds[step]:>8.1f}s")
print(f" {'':<4} {'total':.<40} {'':<7}{total_elapsed:>8.1f}s")
print(f"{'=' * 60}")
print(f" Pipeline {'COMPLETE' if all_ok else 'FINISHED WITH ERRORS'}")
if warned:
+4 -3
View File
@@ -30,7 +30,7 @@ import yaml
sys.path.insert(0, os.path.dirname(__file__))
import upstream
from common import load_emulator_profiles, suggest_names
from common import github_token, load_emulator_profiles, suggest_names
from hashing import parse_hash_list
from sourceref import ( # noqa: F401 (re-exported for callers and tests) # noqa: F401 (the prose grammar, re-exported)
_ANNOTATION_RE,
@@ -3372,12 +3372,13 @@ def select_profiles(profiles: dict, args: argparse.Namespace) -> dict:
def _check_quota(count: int, offline: bool) -> None:
"""Refuse a run the anonymous quota cannot carry to completion."""
if offline or os.environ.get("GITHUB_TOKEN"):
if offline or github_token():
return
if count * 4 > ANON_QUOTA:
print(
f"{count} profiles need roughly {count * 4} API calls, the anonymous "
f"quota is {ANON_QUOTA} per hour. Set GITHUB_TOKEN or use --offline.",
f"quota is {ANON_QUOTA} per hour. Log in with `gh auth login`, set "
"GITHUB_TOKEN, or use --offline.",
file=sys.stderr,
)
raise SystemExit(1)
+2 -2
View File
@@ -25,7 +25,7 @@ import urllib.request
import zipfile
from pathlib import Path
from artifacts import file_lock as _file_lock
from common import platform_arg, require_platform, yaml_load
from common import github_token, platform_arg, require_platform, yaml_load
try:
import yaml
@@ -159,7 +159,7 @@ def _api_request(url: str) -> dict:
"Accept": "application/json",
},
)
token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
token = github_token()
if token and "github" in url:
req.add_header("Authorization", f"token {token}")
with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
+4 -10
View File
@@ -289,17 +289,11 @@ def _github_token_env() -> dict[str, str]:
emudeck and retropie target scrapers hit the GitHub API. Without a token
the unauthenticated rate limit (60/h) is burnt within a few platforms.
"""
from common import github_token # noqa: PLC0415
env = os.environ.copy()
if env.get("GITHUB_TOKEN"):
return env
try:
proc = subprocess.run(
["gh", "auth", "token"], capture_output=True, text=True, timeout=10, check=False
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return env
token = (proc.stdout or "").strip()
if proc.returncode == 0 and token:
token = github_token()
if token:
env["GITHUB_TOKEN"] = token
return env
+5 -1
View File
@@ -467,7 +467,11 @@ def _resolve_modes(args: argparse.Namespace) -> None:
def main() -> int:
parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__)
parser = argparse.ArgumentParser(
allow_abbrev=False,
description=__doc__.replace("``", ""),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--identify", action="store_true", help="label local archives")
parser.add_argument(
"--missing", action="store_true", help="attempt pinned archives we lack"
+6 -5
View File
@@ -3,7 +3,6 @@
from __future__ import annotations
import json
import os
import re
import sys
import time
@@ -16,7 +15,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, NoReturn
from common import write_text_atomic, yaml_load
from common import github_token, write_text_atomic, yaml_load
@dataclass
@@ -273,7 +272,8 @@ def github_headers() -> dict[str, str]:
"User-Agent": "retrobios-scraper/1.0",
"Accept": "application/vnd.github.v3+json",
}
token = os.environ.get("GITHUB_TOKEN")
# GITHUB_TOKEN, else the token gh is logged in with.
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
return headers
@@ -338,10 +338,11 @@ def _exit_on_scrape_error(error: Exception) -> NoReturn:
"""Print what failed and what to try, then exit 1 without a traceback."""
print(f"Error: {error}", file=sys.stderr)
refused = re.search(r"\b(403|429)\b|rate limit", str(error), re.I)
if refused and not os.environ.get("GITHUB_TOKEN"):
if refused and not github_token():
# A gh login is read on its own, so the hint is the login itself.
print(
"Anonymous GitHub API calls are limited to 60 an hour. "
"Run again with GITHUB_TOKEN=$(gh auth token) set.",
"Log in with `gh auth login`, or set GITHUB_TOKEN, and run again.",
file=sys.stderr,
)
else:
+2 -2
View File
@@ -20,7 +20,7 @@ from typing import Any
import yaml
from common import source_yaml_paths, write_text_atomic, yaml_load
from common import github_token, source_yaml_paths, write_text_atomic, yaml_load
from scripts.scraper._hash_merge import compute_diff, merge_fbneo_profile
from scripts.scraper.fbneo_parser import parse_fbneo_source_tree
@@ -125,7 +125,7 @@ def _version_from_api() -> str:
import urllib.request
headers = {"User-Agent": "retrobios-scraper/1.0"}
token = os.environ.get("GITHUB_TOKEN")
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(
+2 -2
View File
@@ -22,7 +22,7 @@ from typing import Any
import yaml
from common import source_yaml_paths, write_text_atomic, yaml_load
from common import github_token, source_yaml_paths, write_text_atomic, yaml_load
import upstream
@@ -144,7 +144,7 @@ def _get_release_tag() -> str:
"User-Agent": "retrobios-scraper/1.0",
"Accept": "application/vnd.github.v3+json",
}
token = os.environ.get("GITHUB_TOKEN")
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
url = "https://api.github.com/repos/mamedev/mame/releases/latest"
+3 -2
View File
@@ -28,7 +28,6 @@ Verification logic (api_data_processing.sh:289-405):
from __future__ import annotations
import json
import os
import re
import sys
import urllib.error
@@ -256,7 +255,9 @@ class Scraper(BaseScraper):
def _fetch_remote_manifests(self) -> list[tuple[str, dict]]:
"""Fetch component manifests via GitHub API."""
token = os.environ.get("GITHUB_TOKEN", "")
from common import github_token # noqa: PLC0415
token = github_token()
headers = {"User-Agent": "retrobios-scraper/1.0"}
if token:
headers["Authorization"] = f"token {token}"
@@ -9,7 +9,6 @@ from __future__ import annotations
import argparse
import json
import os
import re
import sys
import urllib.error
@@ -18,7 +17,7 @@ from datetime import datetime, timezone
import yaml
from common import yaml_load
from common import github_token, yaml_load
from . import BaseTargetScraper
@@ -51,7 +50,7 @@ _SKIP = {"retroarch_maincfg", "retroarch"}
def _fetch(url: str) -> str:
headers = {"User-Agent": "retrobios-scraper/1.0"}
token = os.environ.get("GITHUB_TOKEN")
token = github_token()
if token and "api.github.com" in url:
headers["Authorization"] = f"Bearer {token}"
try:
@@ -10,7 +10,6 @@ from __future__ import annotations
import argparse
import json
import os
import re
import sys
import urllib.error
@@ -67,7 +66,9 @@ _MODULE_FLAGS_RE = re.compile(r'rp_module_flags\s*=\s*["\']([^"\']*)["\']')
def _fetch(url: str, accept: str = "text/plain") -> str:
headers = {"User-Agent": "retrobios-scraper/1.0", "Accept": accept}
token = os.environ.get("GITHUB_TOKEN")
from common import github_token # noqa: PLC0415
token = github_token()
if token and "api.github.com" in url:
headers["Authorization"] = f"Bearer {token}"
try:
+20 -3
View File
@@ -290,6 +290,7 @@ def _merge_hashes(
continue
if not ours:
existing[h] = theirs
existing.setdefault("_hash_by", {})[h] = emu_name
continue
t_list = theirs if isinstance(theirs, list) else [theirs]
o_list = ours if isinstance(ours, list) else [ours]
@@ -305,9 +306,13 @@ def _merge_hashes(
key = (file_entry["name"], h, str(ours), str(theirs), emu_name)
if key not in _CONFLICTS_SHOWN:
_CONFLICTS_SHOWN.add(key)
# Both cores and which value stays: naming only the second
# core sent the reader through ten profiles for the first.
first = existing.get("_hash_by", {}).get(h, "an earlier core")
print(
f"WARNING: hash conflict for {file_entry['name']} "
f"({h}: {ours} vs {theirs}, core {emu_name})",
f"WARNING: hash conflict for {file_entry['name']}: "
f"{h} {ours} ({first}) kept, {theirs} ({emu_name}) "
"left out of the model",
file=sys.stderr,
)
@@ -355,6 +360,9 @@ def _new_truth_entry(file_entry: dict, emu_name: str, db: dict | None) -> dict:
entry["_source_refs"] = {_serialize_source_ref(sr)} if sr is not None else set()
if db:
_enrich_hashes(entry, db)
# Which core gave each hash, for the conflict warning. Dropped before
# the model is written.
entry["_hash_by"] = {h: emu_name for h in _HASH_FIELDS if entry.get(h)}
return entry
@@ -685,6 +693,7 @@ def generate_platform_truth(
# Convert sets to sorted lists for serialization
for sys_id, sys_data in systems.items():
for fe in sys_data.get("files", []):
fe.pop("_hash_by", None)
fe["_cores"] = sorted(fe.get("_cores", set()))
fe["_required_by"] = sorted(fe.get("_required_by", set()))
fe["_named_by"] = sorted(fe.get("_named_by", set()))
@@ -796,6 +805,7 @@ def _diff_system(truth_sys: dict, scraped_sys: dict) -> dict:
missing.append(
{
"name": t_entry["name"],
"destination": t_entry.get("path") or t_entry["name"],
"cores": list(t_entry.get("_cores", [])),
"source_refs": list(t_entry.get("_source_refs", [])),
}
@@ -804,6 +814,7 @@ def _diff_system(truth_sys: dict, scraped_sys: dict) -> dict:
s_entry = scraped_files[s_position]
mismatch = _hash_mismatch(t_entry, s_entry)
if mismatch:
mismatch["destination"] = s_entry.get("destination") or s_entry["name"]
hash_mismatch.append(mismatch)
t_req = t_entry.get("required")
s_req = s_entry.get("required")
@@ -811,6 +822,7 @@ def _diff_system(truth_sys: dict, scraped_sys: dict) -> dict:
required_mismatch.append(
{
"name": s_entry["name"],
"destination": s_entry.get("destination") or s_entry["name"],
"truth_required": t_req,
"scraped_required": s_req,
}
@@ -835,7 +847,12 @@ def _diff_system(truth_sys: dict, scraped_sys: dict) -> dict:
if key in seen_extra:
continue
seen_extra.add(key)
entry = {"name": s_entry["name"]}
# The destination tells apart two declarations of one name: the
# report listed six `hiscore.dat` to add beside one to drop.
entry = {
"name": s_entry["name"],
"destination": s_entry.get("destination") or s_entry["name"],
}
if has_unprofiled:
extra_unprofiled.append(entry)
else:
+7 -3
View File
@@ -206,9 +206,13 @@ def _headers(url: str, accept_json: bool = False) -> dict[str, str]:
headers = {"User-Agent": USER_AGENT}
if accept_json:
headers["Accept"] = "application/json"
token = os.environ.get("GITHUB_TOKEN", "")
if token and urllib.parse.urlsplit(url).netloc in GITHUB_HOSTS:
headers["Authorization"] = f"token {token}"
if urllib.parse.urlsplit(url).netloc in GITHUB_HOSTS:
from common import github_token # noqa: PLC0415
# GITHUB_TOKEN, else the one gh is logged in with.
token = github_token()
if token:
headers["Authorization"] = f"token {token}"
return headers
+5 -1
View File
@@ -280,7 +280,11 @@ def validate_site(site: Path, config_path: Path) -> list[str]:
def main() -> None:
parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__)
parser = argparse.ArgumentParser(
allow_abbrev=False,
description=__doc__.replace("``", ""),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--site-dir", default="site",
help="the built site (default: %(default)s)",
+52 -4
View File
@@ -937,17 +937,34 @@ def _resolve_declaration(
return local_path, "" if held_by else resolve_status, held_by
def _twin_unmet(result: dict, held_by: str | None) -> bool:
def _twin_unmet(
result: dict,
held_by: str | None,
file_entry: dict | None = None,
sys_id: str = "",
destination: str = "",
) -> bool:
"""Mark a declaration unmet at a path another declaration holds.
The path is answered by the winner's file, which the pack counts once.
This declaration's failure is reported, never folded into the
destination's status.
destination's status. The message names the path, both systems and
what this declaration expects: "is not met at the path coco holds"
named neither the path nor the file wanted there.
"""
if held_by is None or result["status"] == Status.OK:
return False
entry = file_entry or {}
expects = next(
(f" ({h} {str(entry[h]).split(',')[0]})"
for h in ("sha1", "md5", "crc32") if entry.get(h)),
"",
)
owner = sys_id or "this"
result["discrepancy"] = (
f"{result.get('name', '')} is not met at the path {held_by} holds"
f"{destination or result.get('name', '')} is declared for {owner}"
f" and for {held_by}. The {owner} declaration{expects} is not met at"
f" the path, which holds the {held_by} file"
)
# Read by reports that list entries: the destination itself is held, and
# a count by destination does not include this one.
@@ -1180,7 +1197,10 @@ def verify_platform(
local_path,
)
shipped_keys.add(_dest_key(file_entry, sys_id))
if _twin_unmet(result, held_by):
if _twin_unmet(
result, held_by, file_entry, sys_id,
f"{base_dest}/{destination}" if base_dest else destination,
):
continue
# Aggregate by destination, keyed as the builder keys its pack.
@@ -1387,6 +1407,26 @@ def _print_ground_truth(gt: list[dict], verbose: bool) -> None:
print(f" {_format_ground_truth_aggregate(gt)}")
def _discrepancy_keys(details: list[dict]) -> set[str]:
"""The DISCREPANCY lines _print_detail_entries prints for these details.
A file already reported MISSING or UNTESTED on its own ground is not
printed again as a discrepancy, and is not counted as one either.
"""
shown = {
f"{d['system']}/{d['name']}"
for d in details
if d["status"] == Status.MISSING
or (d["status"] == Status.UNTESTED and not d.get("discrepancy"))
}
return {
key
for d in details
if d.get("discrepancy")
and (key := f"{d['system']}/{d['name']}") not in shown
}
def _print_detail_entries(details: list[dict], seen: set[str], verbose: bool) -> None:
"""Print UNTESTED, MISSING, and DISCREPANCY entries from verification details."""
for d in details:
@@ -2562,6 +2602,14 @@ def _overall_summary(group_results: list[tuple[dict, list[str]]]) -> str:
for r, group in group_results
}
parts.append(f"{sum(counts.values())} {label}{where(counts)}")
# Shipped files whose bytes contradict a declaration or a core's own
# check: the body printed 19 of them under a last line that read clean.
counts = {
" / ".join(group): len(_discrepancy_keys(r.get("details", [])))
for r, group in group_results
}
total = sum(counts.values())
parts.append(f"{total} {'discrepancy' if total == 1 else 'discrepancies'}{where(counts)}")
lines = [f"{platforms} platforms: {', '.join(parts)}"]
missing = {
+7 -9
View File
@@ -965,7 +965,7 @@ class PipelineRegressions(unittest.TestCase):
def test_every_refresh_failure_reaches_pipeline_exit_status(self):
for failed_label in (
"2/8 refresh data directories",
"2 refresh data directories",
"2a refresh MAME hashes",
"2a2 refresh FBNeo hashes",
):
@@ -2223,8 +2223,8 @@ class OneRefreshPerOnlineRun(unittest.TestCase):
contextlib.redirect_stdout(io.StringIO()), \
self.assertRaises(SystemExit):
pipeline.main()
self.assertIn("2/8 refresh data directories", commands)
for label in ("4/8 generate packs", "4b/8 generate install manifests"):
self.assertIn("2 refresh data directories", commands)
for label in ("4 generate packs", "4b generate install manifests"):
with self.subTest(step=label):
self.assertIn("--no-refresh-data", commands[label])
self.assertNotIn("--offline", commands[label])
@@ -4484,12 +4484,10 @@ class PipelineDocstringNamesThePrintedSteps(unittest.TestCase):
source = (ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
doc = ast.get_docstring(ast.parse(source))
listed = set(re.findall(r"^ (\d[a-z0-9]*)\s{2,}", doc, re.M))
# A step label is "N/8 ..." or a lettered sub-step "2b2 ...".
printed = {
step
for step, of_eight in re.findall(r'(?:"|--- )(\d+[a-z0-9]*)(/8)? [a-zA-Z]', source)
if of_eight or not step.isdigit()
}
sys.path.insert(0, str(ROOT / "scripts"))
import pipeline
printed = {pipeline.step_number(key) for key in pipeline.STEP_LABELS}
self.assertEqual(listed, printed)
+2
View File
@@ -664,6 +664,7 @@ class EveryGitHubApiCallerSendsTheToken(unittest.TestCase):
if "api.github.com" in (source := path.read_text(encoding="utf-8"))
and "GITHUB_TOKEN" not in source
and "github_headers(" not in source
and "github_token(" not in source
)
self.assertEqual(silent, [])
@@ -690,6 +691,7 @@ class EveryGitHubApiCallerSendsTheToken(unittest.TestCase):
requests
and "api.github.com" in body
and "github_headers(" not in body
and "github_token(" not in body
and "GITHUB_TOKEN" not in body
):
silent.append(f"{path.relative_to(scripts)}:{node.name}")
+1 -1
View File
@@ -818,7 +818,7 @@ class SlotsJudgeWhatVerifyJudged(unittest.TestCase):
def test_the_pipeline_passes_both_flags(self):
source = (Path(__file__).resolve().parents[1] / "scripts" / "pipeline.py").read_text()
step = source[source.index('slots_cmd = [sys.executable, "scripts/slots.py"]'):]
step = step[:step.index("3b/8")]
step = step[:step.index("3b slot arbitration")]
self.assertIn('slots_cmd.append("--include-archived")', step)
self.assertIn('slots_cmd.extend(["--target", args.target])', step)
+374
View File
@@ -0,0 +1,374 @@
"""What the maintainer tools print: names a user can act on, counts that add up."""
from __future__ import annotations
import contextlib
import io
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import common # noqa: E402
import diff_truth # noqa: E402
import pipeline # noqa: E402
import truth # noqa: E402
import verify # noqa: E402
class ABuildbotNameSelectsItsProfile(unittest.TestCase):
"""`--emulator pcfx` was refused and pointed at two PlayStation cores,
though pcfx is in the cores: of beetle_pcfx."""
PROFILES = {
"beetle_pcfx": {"type": "libretro", "cores": ["pcfx"]},
"pcsx1": {"type": "libretro"},
"pcsx2": {"type": "standalone"},
"a": {"type": "libretro", "cores": ["shared"]},
"b": {"type": "libretro", "cores": ["shared"]},
}
def test_a_core_name_serves_the_profile(self):
selected = common.select_emulator_profiles(["pcfx"], self.PROFILES)
self.assertEqual([name for name, _profile in selected], ["beetle_pcfx"])
def test_a_name_two_profiles_claim_names_both(self):
with self.assertRaises(common.ProfileSelectionError) as caught:
common.select_emulator_profiles(["shared"], self.PROFILES)
self.assertIn("a, b", str(caught.exception))
def test_an_unknown_name_still_gets_a_hint(self):
with self.assertRaises(common.ProfileSelectionError) as caught:
common.select_emulator_profiles(["pcsx3"], self.PROFILES)
self.assertIn("Did you mean", str(caught.exception))
def test_cross_reference_reports_the_resolved_profile(self):
"""cross_reference filtered on the typed name, found no profile under
it and answered `pcfx is not run by None`."""
def report(name: str) -> subprocess.CompletedProcess:
return subprocess.run(
[sys.executable, "scripts/cross_reference.py", "--emulator", name],
cwd=REPO_ROOT, capture_output=True, text=True, timeout=600,
)
by_core, by_profile = report("pcfx"), report("beetle_pcfx")
self.assertEqual(by_core.returncode, 0, by_core.stderr)
self.assertEqual(by_core.stdout, by_profile.stdout)
class AMistypedTargetGetsTheClosestName(unittest.TestCase):
"""`--target swich` printed every target and alias on two lines."""
def test_did_you_mean(self):
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as tmp:
targets = Path(tmp) / "targets"
targets.mkdir()
names = [f"board-{n}" for n in range(30)] + ["nintendo-switch"]
(targets / "p.yml").write_text(
"targets:\n" + "".join(f" {n}:\n cores: [a]\n" for n in names)
)
(targets / "_overrides.yml").write_text(
"p:\n targets:\n nintendo-switch:\n aliases: [switch]\n"
)
with self.assertRaises(ValueError) as caught:
common.load_target_config("p", "swich", tmp)
message = str(caught.exception)
self.assertIn("Did you mean: switch?", message)
self.assertIn("lists all 31", message)
self.assertNotIn("board-0", message)
class TheTruthDiffShowsWhereAndAsksNothing(unittest.TestCase):
"""The markdown wrote `**Remove** hiscore.dat (phantom)` beside six
`**Add** hiscore.dat`, none with a path, for files profiled cores read."""
def _report(self) -> dict:
truth_sys = {"files": [
{"name": "hiscore.dat", "path": "mame2003/hiscore.dat",
"md5": "a" * 32, "_cores": ["mame2003"]},
], "_coverage": {"cores_unprofiled": []}}
scraped_sys = {"files": [
{"name": "hiscore.dat", "destination": "fbneo/hiscore.dat", "md5": "b" * 32},
]}
report = truth.diff_platform_truth(
{"systems": {"arcade": truth_sys}}, {"systems": {"arcade": scraped_sys}}
)
report["platform"] = "p"
return report
def test_entries_carry_their_destination(self):
div = self._report()["divergences"]["arcade"]
self.assertEqual(div["extra_phantom"][0]["destination"], "fbneo/hiscore.dat")
self.assertEqual(div["missing"][0]["destination"], "mame2003/hiscore.dat")
def test_neither_format_asks_for_a_removal(self):
report = self._report()
for text in (diff_truth._format_markdown(report),
diff_truth._format_terminal(report)):
self.assertNotIn("Remove", text)
self.assertNotIn("phantom", text.lower())
self.assertIn("fbneo/hiscore.dat", text)
self.assertIn("mame2003/hiscore.dat", text)
class TheVerifySummaryCountsDiscrepancies(unittest.TestCase):
"""19 DISCREPANCY lines above a last line that read clean."""
def test_the_closing_line_names_them(self):
details = [
{"system": "arcade", "name": "aes.zip", "status": verify.Status.OK,
"discrepancy": "file present (OK) but hash differs"},
# Printed as MISSING, so not a second time as a discrepancy.
{"system": "arcade", "name": "x.zip", "status": verify.Status.MISSING,
"discrepancy": "unmet"},
{"system": "arcade", "name": "x.zip", "status": verify.Status.MISSING},
]
summary = verify._overall_summary([
({"severity_counts": {}, "details": details}, ["RetroArch"]),
({"severity_counts": {}, "details": []}, ["Batocera"]),
])
self.assertIn("1 discrepancy (RetroArch)", summary)
def test_a_shared_destination_names_the_path_and_both_sides(self):
result = {"name": "disk.rom", "status": verify.Status.UNTESTED}
entry = {"name": "disk.rom", "md5": "793f8678" + "0" * 24}
self.assertTrue(verify._twin_unmet(
result, "coco", entry, "nec-pc-88", "bios/disk.rom"
))
message = result["discrepancy"]
for part in ("bios/disk.rom", "nec-pc-88", "coco", "md5 793f8678"):
self.assertIn(part, message)
class ThePipelineCountsItsRealSteps(unittest.TestCase):
"""Labels read N/8 for a run of 24 steps, and the summary used keys
the log never printed."""
def test_every_recorded_step_has_a_label(self):
import re # noqa: PLC0415
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
keys = set(re.findall(r'results\["(\w+)"\] =', source))
self.assertEqual(keys - set(pipeline.STEP_LABELS), set())
def test_a_label_written_out_is_one_of_the_list(self):
import re # noqa: PLC0415
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
written = re.findall(r'run\([^()]*?, "([^"]+)"\)', source)
numbers = {label.split(" ", 1)[0] for label in pipeline.STEP_LABELS.values()}
for label in written:
with self.subTest(label=label):
# A step's own label, or a second command of a listed step.
self.assertIn(label.split(" ", 1)[0], numbers)
def test_progress_comes_from_the_list(self):
total = len(pipeline.STEP_LABELS)
title = pipeline.step_title(pipeline.STEP_LABELS["render_site"])
self.assertEqual(title, f"8b render site [{total}/{total}]")
self.assertNotIn("/8", pipeline.step_title(pipeline.STEP_LABELS["verify"]))
def test_an_offline_buildbot_check_has_a_summary_row(self):
argv = ["pipeline.py", "--offline", "--check-buildbot", "--skip-packs",
"--skip-docs"]
out = io.StringIO()
with mock.patch.object(sys, "argv", argv), mock.patch.object(
pipeline, "run", return_value=(True, "RetroArch: 1/1 OK\n")
), contextlib.redirect_stdout(out), contextlib.suppress(SystemExit):
pipeline.main()
summary = out.getvalue().split("=" * 60)[1]
self.assertIn("2b check_buildbot", summary)
# The number the run printed for the step leads its summary row.
self.assertIn("4d bundled_only", summary)
self.assertTrue(
pipeline.step_title(pipeline.STEP_LABELS["bundled_only"]).startswith("4d ")
)
class AHashConflictNamesBothCores(unittest.TestCase):
"""The warning named the second core only, and not which value stayed."""
def test_both_cores_and_the_kept_value(self):
truth._CONFLICTS_SHOWN.clear()
system: dict = {}
err = io.StringIO()
with contextlib.redirect_stderr(err):
truth._merge_file_into_system(
system, {"name": "G.yaml", "path": "r/G.yaml", "sha1": "5e" * 20},
"pcsx2", None,
)
truth._merge_file_into_system(
system, {"name": "G.yaml", "path": "r/G.yaml", "sha1": "7e" * 20},
"yaps2", None,
)
self.assertIn(f"{'5e' * 20} (pcsx2) kept", err.getvalue())
self.assertIn(f"{'7e' * 20} (yaps2)", err.getvalue())
def test_the_origin_is_not_written_to_the_model(self):
profile = {"type": "libretro", "systems": ["sony-playstation"],
"files": [{"name": "a.bin", "system": "sony-playstation",
"md5": "a" * 32}]}
model = truth.generate_platform_truth(
"p", {"cores": ["c"], "systems": {}}, {}, {"c": profile}
)
entries = [f for s in model["systems"].values() for f in s.get("files", [])]
self.assertTrue(entries)
self.assertFalse(any("_hash_by" in f for f in entries))
class HelpTextsWrapBetweenWords(unittest.TestCase):
def test_check_freshness_areas(self):
proc = subprocess.run(
[sys.executable, "scripts/check_freshness.py", "--help"],
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
env={**os.environ, "COLUMNS": "80"}, timeout=60,
)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("coreinfo", " ".join(proc.stdout.split()))
def test_docstring_descriptions_keep_their_layout(self):
"""romset_recipes ran its two questions into one paragraph and both
help texts printed reST double backquotes."""
for script, line in (("romset_recipes.py", "--identify"),
("validate_site.py", None)):
with self.subTest(script=script):
proc = subprocess.run(
[sys.executable, f"scripts/{script}", "--help"],
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
env={**os.environ, "COLUMNS": "80"}, timeout=60,
)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertNotIn("``", proc.stdout)
if line:
self.assertIn(line, proc.stdout.splitlines())
class ListPlatformsTakesTheSharedFlag(unittest.TestCase):
"""verify, generate_pack, slots and pipeline say --include-archived."""
def test_both_spellings(self):
for flag in ("--include-archived", "--all"):
with self.subTest(flag=flag):
env = {k: v for k, v in os.environ.items() if k != "GITHUB_OUTPUT"}
proc = subprocess.run(
[sys.executable, "scripts/list_platforms.py", flag],
capture_output=True, text=True, cwd=REPO_ROOT, check=False,
env=env, timeout=60,
)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("retropie", proc.stdout)
def test_the_pack_list_marks_archived_platforms(self):
"""generate_pack --list printed retropie like a platform with a
release pack."""
proc = subprocess.run(
[sys.executable, "scripts/generate_pack.py", "--list"],
capture_output=True, text=True, cwd=REPO_ROOT, check=False, timeout=120,
)
self.assertEqual(proc.returncode, 0, proc.stderr)
lines = proc.stdout.splitlines()
self.assertIn("retropie (archived)", lines)
self.assertIn("retroarch", lines)
class NoScriptReadsTheTokenOnItsOwn(unittest.TestCase):
"""Nine scripts read GITHUB_TOKEN from the environment after
github_token() existed, and went anonymous when run on their own.
download.py is the user's downloader: it runs without the repository and
a user's gh session is not its business."""
READS = ('environ.get("GITHUB_TOKEN"', "environ['GITHUB_TOKEN']",
'environ["GITHUB_TOKEN"]', 'environ.get("GH_TOKEN"')
OWNERS = {"scripts/common.py", "scripts/download.py"}
def test_every_reader_asks_github_token(self):
for path in sorted((REPO_ROOT / "scripts").rglob("*.py")):
relative = path.relative_to(REPO_ROOT).as_posix()
if relative in self.OWNERS:
continue
for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
if " = " in line and line.split(" = ")[0].strip().startswith("env["):
continue
with self.subTest(file=relative, line=number):
self.assertFalse(any(read in line for read in self.READS), line.strip())
class TheGitHubTokenComesFromGh(unittest.TestCase):
"""Only refresh_stale asked gh: profile_sync refused the run and the
scrapers went anonymous for a maintainer logged in to gh."""
def setUp(self):
common._gh_cli_token.cache_clear()
self.addCleanup(common._gh_cli_token.cache_clear)
env = {k: v for k, v in os.environ.items()
if k not in ("GITHUB_TOKEN", "GH_TOKEN", "RETROBIOS_OFFLINE")}
patcher = mock.patch.dict(os.environ, env, clear=True)
patcher.start()
self.addCleanup(patcher.stop)
@staticmethod
def _gh(token: str = "gho_fromgh", code: int = 0):
return mock.patch(
"subprocess.run",
return_value=subprocess.CompletedProcess(["gh"], code, token, ""),
)
def test_gh_answers_when_the_variable_is_unset(self):
with self._gh():
self.assertEqual(common.github_token(), "gho_fromgh")
# Subprocesses of the run inherit it.
self.assertEqual(os.environ.get("GITHUB_TOKEN"), "gho_fromgh")
def test_the_variable_wins(self):
os.environ["GITHUB_TOKEN"] = "set"
with self._gh() as run:
self.assertEqual(common.github_token(), "set")
run.assert_not_called()
def test_no_gh_means_anonymous(self):
with mock.patch("subprocess.run", side_effect=FileNotFoundError("gh")):
self.assertEqual(common.github_token(), "")
with self._gh("", 1):
common._gh_cli_token.cache_clear()
self.assertEqual(common.github_token(), "")
def test_offline_runs_never_ask(self):
os.environ["RETROBIOS_OFFLINE"] = "1"
with self._gh() as run:
self.assertEqual(common.github_token(), "")
run.assert_not_called()
def test_every_api_caller_reads_it(self):
import profile_sync # noqa: PLC0415
import upstream # noqa: PLC0415
from scraper import base_scraper # noqa: PLC0415
with self._gh():
self.assertEqual(
upstream._headers("https://api.github.com/repos/o/n")["Authorization"],
"token gho_fromgh",
)
os.environ.pop("GITHUB_TOKEN")
with self._gh():
self.assertEqual(base_scraper.github_headers()["Authorization"],
"Bearer gho_fromgh")
os.environ.pop("GITHUB_TOKEN")
with self._gh(), contextlib.redirect_stderr(io.StringIO()):
profile_sync._check_quota(606, offline=False)
def test_another_forge_never_triggers_gh(self):
import upstream # noqa: PLC0415
with self._gh() as run:
headers = upstream._headers("https://gitlab.com/api/v4/projects/1")
self.assertNotIn("Authorization", headers)
run.assert_not_called()
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -37,7 +37,7 @@ Every module in `tests/`, by the area it guards.
| Area | Fixtures | Modules |
|------|----------|---------|
| Core regression suites | synthetic | `test_e2e`, `test_audit_regressions` |
| Core regression suites | synthetic | `test_e2e`, `test_audit_regressions`, `test_tool_messages` |
| File resolution and identity | synthetic + collection | `test_adler32_resolution`, `test_agnostic_scan`, `test_agnostic_scan_mode`, `test_agnostic_scope`, `test_agnostic_substitute`, `test_archive_choice`, `test_bios_companions`, `test_bundled_identity`, `test_casefold_cache`, `test_db_aliases`, `test_declared_names`, `test_derivations`, `test_directory_entries`, `test_extra_identity`, `test_foreign_tree`, `test_foreign_tree_names`, `test_frontend_digest`, `test_game_data_homonyms`, `test_hashing`, `test_identity`, `test_md5_pack_context`, `test_name_tiebreak`, `test_nec_filing`, `test_owner_resolution`, `test_path_suffix`, `test_resolution_evidence`, `test_shipped_content`, `test_twin_declarations`, `test_validated_choice`, `test_variant_choice`, `test_variant_identity`, `test_zipped_member` |
| Verification, cross-reference and profile checks | synthetic + profiles | `test_crypto_aes`, `test_crypto_unique_data`, `test_crypto_verify`, `test_case_insensitive_platforms`, `test_checks_have_values`, `test_core_file_fills_declaration`, `test_cross_reference_sources`, `test_file_consumers`, `test_fileless_audit`, `test_gap_settlement`, `test_hidden_extras`, `test_launcher_layer`, `test_native_mode`, `test_no_case_collisions`, `test_path_conflict_status`, `test_path_null_copies`, `test_profile_checks`, `test_romm_hash_rule`, `test_shared_groups`, `test_standalone_mode`, `test_undeclared_dedup`, `test_unplaced_extras`, `test_validation_scope`, `test_verify_all_dir`, `test_verify_emulator_evidence`, `test_verify_listing`, `test_verify_pack_choice` |
| Packs and manifests | synthetic + real packs | `test_artifact_lock`, `test_atomic_artifacts`, `test_batocera_cores`, `test_deterministic_zip`, `test_emulator_pack_counts`, `test_finalize_pack_names`, `test_large_file_cache`, `test_libretro_layout`, `test_manifest_entry`, `test_pack_conformance_dirs`, `test_pack_counts`, `test_pack_failures`, `test_pack_file_names`, `test_pack_grouping`, `test_pack_integrity`, `test_pack_member_identity`, `test_pack_readme`, `test_platform_inheritance`, `test_platform_keys`, `test_required_only_archives`, `test_required_only_twins`, `test_split_extras`, `test_split_names`, `test_split_pack`, `test_system_pack_extras`, `test_system_pack_parity`, `test_torrentzip`, `test_which_pack` |
+10 -10
View File
@@ -23,11 +23,11 @@ Pipeline steps:
| Step | Description | Skipped by |
|------|-------------|------------|
| 0/8 | Validate profiles and platform YAMLs against the schemas | - (no `jsonschema` installed) |
| 1/8 | Generate database | - |
| 0 | Validate profiles and platform YAMLs against the schemas | - (no `jsonschema` installed) |
| 1 | Generate database | - |
| 1b | Dump-catalog coverage report | - (offline, reads `provenance/`) |
| 1c | Romset recipes: identify archives, list rebuild targets | - (offline, reads `recipes/`) |
| 2/8 | Refresh data directories | `--offline` |
| 2 | Refresh data directories | `--offline` |
| 2a | Refresh MAME BIOS hashes | `--offline` |
| 2a2 | Refresh FBNeo BIOS hashes | `--offline` |
| 2b | Check buildbot staleness | only with `--check-buildbot`, skipped by `--offline` |
@@ -35,17 +35,17 @@ Pipeline steps:
| 2c | Generate truth YAMLs | only with `--with-truth` / `--with-export` |
| 2d | Diff truth vs scraped | only with `--with-truth` / `--with-export` |
| 2e | Export native formats | only with `--with-export` |
| 3/8 | Verify all platforms (`--strict`) | - |
| 3 | Verify all platforms (`--strict`) | - |
| 3a | File identity: one content under two identities (`identity.py --strict`, gates the run) | - |
| 3b | Slot arbitration: destinations both layers claim (`slots.py`, reports only) | - |
| 4/8 | Generate packs | `--skip-packs` |
| 4 | Generate packs | `--skip-packs` |
| 4b | Generate install manifests | `--skip-packs` |
| 4c | Generate target manifests | `--skip-packs` |
| 4d | Collected copies only an emulator install provides (`bundled_only.py --strict`, gates the run) | - |
| 5/8 | Consistency check | if verify or pack skipped |
| 6/8 | Pack integrity (extract + hash) | `--skip-packs` |
| 7/8 | Generate README | `--skip-docs` |
| 8/8 | Generate site | `--skip-docs` |
| 5 | Consistency check | if verify or pack skipped |
| 6 | Pack integrity (extract + hash) | `--skip-packs` |
| 7 | Generate README | `--skip-docs` |
| 8 | Generate site | `--skip-docs` |
| 8b | Render the site with `mkdocs build --strict`, then `validate_site.py`, as the deploy does | `--skip-docs`, or no `mkdocs` installed |
Pipeline flags:
@@ -614,7 +614,7 @@ python scripts/refresh_stale.py --only platforms,targets --jobs 6
| `validate_pr.py` | Validate BIOS files in pull requests, post markdown report |
| `validate_schemas.py` | Validate the data contracts: schemas and semantic invariants. `--source-only` checks `emulators/` and `platforms/` alone, which is what PR validation runs |
| `auto_fetch.py` | Fetch missing BIOS files from known sources (4-step pipeline) |
| `list_platforms.py` | List active platforms (`--all` includes archived, used by CI) |
| `list_platforms.py` | List active platforms (`--include-archived` adds the archived ones, used by CI) |
| `download.py` | Download a pack from GitHub releases, every part checked (Python, stdlib only) |
| `download.sh` | Same, as a shell one-liner (`curl` + `unzip`) |
| `split_pack.py` | Cut a pack over the release asset limit into parts that are each a ZIP of whole files, read back before the pack is removed |