chore: merge round 2 fixes (tooling)

This commit is contained in:
Abdessamad Derraz committed 2026-10-21 10:31:04 +02:00
commit 055627f39c
22 files changed
+704 -141

No files matched your search

Binary file not shown.
+18 -3
View File
@@ -1256,6 +1256,8 @@ def check_profiles(run: bool, offline: bool) -> list[Finding]:
"--all",
"--json",
"--check-version",
"--detect-new-files",
"--watch-hashes",
]
if offline:
cmd.append("--offline")
@@ -1276,7 +1278,9 @@ def summarize_profile_sync(report: object) -> list[Finding]:
The JSON carries the ref verdicts and the pin against the tip. A profile
whose refs need review is stale, one the forge could not serve is unknown,
and one whose upstream moved while every ref still anchors is reported as
such without being counted as stale.
such without being counted as stale. A cited file that gained a file name
since the pin, or a hash the profile does not declare, is stale too: the
upstream may load a file the profile does not document.
"""
if not isinstance(report, list):
return [Finding("profiles", "profile_sync", ERROR, detail="report is not a list")]
@@ -1314,7 +1318,18 @@ def summarize_profile_sync(report: object) -> list[Finding]:
continue
review = row.get("needs_review") or 0
pin, head = str(row.get("pin") or ""), str(row.get("head") or "")
if review:
added = sorted({
str(item.get("name"))
for item in row.get("new_files") or ()
if isinstance(item, dict) and item.get("since_pin")
})
hashes = len(row.get("new_hashes") or ())
reasons = [f"{review} refs to review"] if review else []
if added:
reasons.append(f"new file names since the pin: {', '.join(added)}")
if hashes:
reasons.append(f"{hashes} undeclared hashes added since the pin")
if reasons:
findings.append(
Finding(
"profiles",
@@ -1322,7 +1337,7 @@ def summarize_profile_sync(report: object) -> list[Finding]:
STALE,
local=pin[:8],
remote=head[:8],
detail=f"{review} refs to review",
detail=", ".join(reasons),
)
)
elif pin and head and pin != head:
+4 -1
View File
@@ -657,8 +657,11 @@ def load_target_config(
# through, where a platform or emulator name gets the closest ones.
raise ValueError(
f"Unknown target '{target}' for platform '{platform_name}'.\n"
# Named as a full command: cross_reference.py and slots.py take
# --target but have no --list-targets of their own.
+ suggest_names(
[target], sorted(alias_index), "--list-targets", len(targets)
[target], sorted(alias_index),
f"verify.py --platform {platform_name} --list-targets", len(targets),
)
)
+10 -5
View File
@@ -986,11 +986,16 @@ def main() -> None:
if args.platform:
from common import load_target_config, resolve_platform_cores
target_cores = (
load_target_config(args.platform, args.target, args.platforms_dir)
if args.target
else None
)
try:
target_cores = (
load_target_config(args.platform, args.target, args.platforms_dir)
if args.target
else None
)
except (FileNotFoundError, ValueError) as exc:
# A typo ended in a traceback where verify and slots name the
# closest target.
parser.error(str(exc))
config = load_platform_config(args.platform, args.platforms_dir)
relevant = resolve_platform_cores(config, profiles, target_cores=target_cores)
run_anywhere = (
+2 -2
View File
@@ -182,7 +182,7 @@ def main() -> None:
group.add_argument(
"--all", action="store_true", help="diff all registered platforms"
)
group.add_argument("--platform", type=platform_arg, help="diff a single platform")
group.add_argument("--platform", "-p", type=platform_arg, help="diff a single platform")
# One output form: --json beside --format markdown printed JSON and
# dropped the markdown without a word.
output = parser.add_mutually_exclusive_group()
@@ -207,7 +207,7 @@ def main() -> None:
help="platform YAML directory (default: %(default)s)",
)
parser.add_argument(
"--target",
"--target", "-t",
help="read the models generate_truth --target wrote, each under the "
"target as its platform files it",
)
+1 -1
View File
@@ -858,7 +858,7 @@ def main() -> None:
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--all", action="store_true", help="every platform")
group.add_argument("--platform", type=platform_arg, help="a single platform")
group.add_argument("--platform", "-p", type=platform_arg, help="a single platform")
parser.add_argument(
"--output-dir", default="exports",
help="directory the updated platform files go to (default: %(default)s)",
+71 -75
View File
@@ -25,8 +25,6 @@ import re
import shutil
import sys
import tempfile
import urllib.error
import urllib.request
import zipfile
from pathlib import Path
@@ -345,68 +343,6 @@ def _platform_pack_stem(
return "_".join(_name_part(n) for n in names) + version_tag
def download_external(file_entry: dict, dest_path: str) -> bool:
"""Download an external BIOS file, verify hash, save to dest_path."""
url = file_entry.get("source_url")
if not url:
return False
sha256 = file_entry.get("sha256")
sha1 = file_entry.get("sha1")
md5 = file_entry.get("md5")
if not (sha256 or sha1 or md5):
print(
f" WARNING: no hash for {file_entry['name']}, skipping unverifiable download"
)
return False
try:
req = urllib.request.Request(
url, headers={"User-Agent": "retrobios-pack-gen/1.0"}
)
with urllib.request.urlopen(req, timeout=120) as resp:
data = resp.read()
except urllib.error.URLError as e:
print(f" WARNING: Failed to download {url}: {e}")
return False
if sha256:
actual = hashlib.sha256(data).hexdigest()
if actual != sha256:
print(f" WARNING: SHA256 mismatch for {file_entry['name']}")
return False
elif sha1:
actual = hashlib.sha1(data).hexdigest()
if actual != sha1:
print(f" WARNING: SHA1 mismatch for {file_entry['name']}")
return False
elif md5:
actual = hashlib.md5(data).hexdigest()
if actual != md5:
print(f" WARNING: MD5 mismatch for {file_entry['name']}")
return False
os.makedirs(os.path.dirname(dest_path), exist_ok=True)
with open(dest_path, "wb") as f:
f.write(data)
return True
# What the builder writes into a platform pack beside the files it carries.
PACK_DOCUMENTS = ("README.txt", "manifest.json")
@@ -952,7 +888,7 @@ def generate_pack(
if status == "external":
file_ext = os.path.splitext(file_entry["name"])[1] or ""
with tempfile.NamedTemporaryFile(
delete=False, suffix=file_ext
delete=False, suffix=file_ext, dir=packresolve.scratch_dir()
) as tmp:
tmp_path = tmp.name
@@ -1707,7 +1643,7 @@ def generate_emulator_pack(
if how == "external":
file_ext = os.path.splitext(fe["name"])[1] or ""
with tempfile.NamedTemporaryFile(
delete=False, suffix=file_ext
delete=False, suffix=file_ext, dir=packresolve.scratch_dir()
) as tmp:
tmp_path = tmp.name
try:
@@ -2152,6 +2088,11 @@ def generate_md5_pack(
plat_file_index.setdefault(alias.lower(), []).append(fe)
context_name = plat_display if platform_name else (emu_display or "Custom")
# The selection mode is part of the name, as for --system: BizHawk is
# both a platform and a profile display name, and the two layouts
# (Firmware/ under the platform, the profile's paths under the core)
# wrote one file, the second run replacing the first without a word.
mode_part = "_Emulator" if emulator_name and not platform_name else ""
# --standalone changes the destination layout, so a run with it must not
# take the name of a run without it.
custom_tags = "".join(
@@ -2166,7 +2107,8 @@ def generate_md5_pack(
"\n".join(sorted(f"{kind}:{value.lower()}" for kind, value in hashes)).encode()
).hexdigest()[:8]
zip_name = (
f"{_name_part(context_name, '_')}_Custom_{selection}{custom_tags}_BIOS_Pack.zip"
f"{_name_part(context_name, '_')}{mode_part}_Custom_{selection}{custom_tags}"
"_BIOS_Pack.zip"
)
zip_path = os.path.join(output_dir, zip_name)
os.makedirs(output_dir, exist_ok=True)
@@ -2370,6 +2312,39 @@ def generate_target_manifests(targets_dir: str, output_dir: str) -> None:
print(f"Generated {count} target manifest(s) in {output_dir}")
def _refuse_unknown_systems(
system_filter: list[str],
platforms: list[str],
platforms_dir: str,
parser: argparse.ArgumentParser,
) -> None:
"""Refuse a --system no selected platform declares, before any work.
The name is known once the config is loaded. Checked inside the build, a
typo cost the lock and a refresh of every data directory first, then
ended on a WARNING and exit 1.
"""
from common import _norm_system_id
wanted = {_norm_system_id(s) for s in system_filter}
declared: set[str] = set()
for name in platforms:
systems = load_platform_config(name, platforms_dir).get("systems", {})
if any(sid in system_filter or _norm_system_id(sid) in wanted for sid in systems):
return
declared.update(systems)
where = f"--platform {platforms[0]}" if len(platforms) == 1 else "--all"
hint = suggest_names(
system_filter,
sorted(declared),
f"--platform {platforms[0]} --list-systems" if len(platforms) == 1
else "--platform <name> --list-systems",
)
parser.error(
f"--system {','.join(system_filter)} matches no system of {where}. {hint}"
)
def _refuse_for_verify_packs(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
"""What --verify-packs cannot be combined with, refused before any lock.
@@ -2431,7 +2406,7 @@ def _validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
parser.error("--split requires --platform or --all")
if args.split and has_emulator:
parser.error("--split is incompatible with --emulator")
if args.group_by != "system" and not args.split:
if args.group_by is not None and not args.split:
parser.error("--group-by requires --split")
if args.target and not (has_platform or has_all):
parser.error("--target requires --platform or --all")
@@ -2448,7 +2423,9 @@ def _validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
# A mode either applies a narrowing flag or refuses it. Swallowing one
# answers about an artifact the caller did not ask for, and the name on
# disk then asserts a narrowing the contents do not carry.
source_given = getattr(args, "source", "full") != "full"
# --source and --group-by default to None so a typed default value is
# seen: compared with "full", `--emulator X --source full` went through.
source_given = getattr(args, "source", None) is not None
if source_given and (has_emulator or (has_system and not has_platform)):
parser.error("--source requires --platform or --all")
if has_from_md5:
@@ -2642,6 +2619,7 @@ def _run_verify_packs(args: argparse.Namespace) -> None:
)
verified = 0
failing: list[str] = []
for platform_name in platforms:
found = sorted(expected[platform_name] & on_disk)
verified += len(found)
@@ -2655,6 +2633,7 @@ def _run_verify_packs(args: argparse.Namespace) -> None:
f"generate_pack.py --platform {platform_name} --output-dir {args.output_dir})"
)
all_ok = False
failing.append(f"{platform_name} (no pack)")
else:
print(f" {platform_name}: SKIP (no pack in {args.output_dir})")
continue
@@ -2700,7 +2679,17 @@ def _run_verify_packs(args: argparse.Namespace) -> None:
for err in errors:
print(f" {err}")
all_ok = False
failing.append(f"{platform_name} ({pack_name})")
if failing:
# Named at the end, where the log is read, and on stderr, as the
# build path does: a failing pack sat among its own error lines.
print(
f"ERROR: packs failed verification: {', '.join(failing)}",
file=sys.stderr,
)
elif verified:
print(f"Verified {verified} pack{'s' if verified != 1 else ''}: all OK")
if not verified:
# Every platform skipped is no check at all: a mistyped or purged
# output directory read as a passed integrity check.
@@ -2918,7 +2907,7 @@ def _refuse_for_all_variants(args: argparse.Namespace, parser: argparse.Argument
# or --required-only beside it would be overridden, and the modes that
# build one pack (emulator, system, hashes) never read it.
for flag, given in (
("--source", args.source != "full"),
("--source", args.source is not None),
("--required-only", args.required_only),
("--emulator", args.emulator),
("--from-md5", args.from_md5 or args.from_md5_file),
@@ -3027,7 +3016,7 @@ def _given_flags(args: argparse.Namespace) -> set[str]:
"--system": args.system,
"--standalone": args.standalone,
"--split": args.split,
"--group-by": args.group_by != "system",
"--group-by": args.group_by is not None,
"--manifest": args.manifest,
"--manifest-targets": args.manifest_targets,
"--verify-packs": args.verify_packs,
@@ -3035,7 +3024,7 @@ def _given_flags(args: argparse.Namespace) -> set[str]:
"--from-md5": args.from_md5,
"--from-md5-file": args.from_md5_file,
"--required-only": args.required_only,
"--source": args.source != "full",
"--source": args.source is not None,
"--target": args.target,
"--region": args.region,
"--one-per-slot": args.one_per_slot,
@@ -3140,9 +3129,9 @@ def _main(held: contextlib.ExitStack) -> None:
parser.add_argument(
"--source",
choices=["platform", "truth", "full"],
default="full",
default=None,
help="File source: platform (YAML only), truth (emulator profiles), full "
"(both)",
"(both, the default)",
)
parser.add_argument(
"--all-variants",
@@ -3155,7 +3144,7 @@ def _main(held: contextlib.ExitStack) -> None:
parser.add_argument(
"--group-by",
choices=["system", "manufacturer"],
default="system",
default=None,
help="Grouping for --split (default: system)",
)
parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)")
@@ -3255,6 +3244,9 @@ def _main(held: contextlib.ExitStack) -> None:
return
_validate_args(args, parser)
# Unset until every refusal has run, so a typed default value is seen.
args.source = args.source or "full"
args.group_by = args.group_by or "system"
# Hash lookup / pack mode
# A platform run may refresh the trees below, under the exclusive lock:
@@ -3371,6 +3363,9 @@ def _main(held: contextlib.ExitStack) -> None:
parser.error("Specify --platform or --all")
return
if system_filter:
_refuse_unknown_systems(system_filter, platforms, args.platforms_dir, parser)
data_registry = load_data_dir_registry(args.platforms_dir)
# The output directory is claimed before the data directory refresh: a
# refresh waits for every run that holds a cache, so a second build
@@ -4915,6 +4910,7 @@ from packresolve import ( # noqa: E402,F401
parse_hash_file,
lookup_hashes,
resolve_file,
download_external,
)
if __name__ == "__main__":
+1 -1
View File
@@ -43,7 +43,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--all", action="store_true", help="all registered platforms")
group.add_argument("--platform", type=platform_arg, help="single platform name")
group.add_argument("--platform", "-p", type=platform_arg, help="single platform name")
parser.add_argument(
"--output-dir",
default=DEFAULT_OUTPUT_DIR,
+6 -2
View File
@@ -98,7 +98,8 @@ def asset_candidates(name: str, registered: Iterable[str]) -> list[str]:
"""Assets that may hold *name*, a registered path or a bare file name.
A bare name shared by several registered paths has one asset per path,
and the caller's hash tells them apart.
and the caller's hash tells them apart. A bare name no registered path
carries has none.
"""
registered = list(registered)
if "/" in name:
@@ -114,7 +115,10 @@ def asset_candidates(name: str, registered: Iterable[str]) -> list[str]:
names[p] for p in sorted(names)
if "/.variants/" in p and os.path.basename(p).startswith(name + ".")
]
return matches or [name]
# A bare name no registered path carries is no asset: .gitignore is the
# registry. Asking the release anyway cost a 404 per name on every
# online build and named an upload that never existed.
return matches
def fetch_large_file(
+61 -2
View File
@@ -6,9 +6,14 @@ and the offline switch that forbids reaching for the network."""
from __future__ import annotations
import re
from largefiles import fetch_large_file
import hashlib
import os
import re
import urllib.error
import urllib.request
from pathlib import Path
from largefiles import fetch_large_file
from common import (
declared_hash_verdict,
resolution_is_hash_exact,
@@ -26,6 +31,60 @@ def set_offline(value: bool) -> None:
global _OFFLINE
_OFFLINE = value
def scratch_dir() -> str:
"""The repository's tmp/, never the system /tmp (a 4 GB tmpfs on WSL)."""
path = Path(__file__).resolve().parent.parent / "tmp"
path.mkdir(exist_ok=True)
return str(path)
def download_external(
file_entry: dict, dest_path: str, offline: bool | None = None
) -> bool:
"""Download a `storage: external` file, verify its hash, write dest_path.
The run's network policy applies: under --offline nothing is fetched and
the entry stays missing, as verify reports it. The body is streamed to
dest_path and hashed on the way, and removed when the hash disagrees.
"""
url = file_entry.get("source_url")
if not url:
return False
name = file_entry.get("name", url)
declared = next(
(
(algo, str(file_entry[algo]).lower())
for algo in ("sha256", "sha1", "md5")
if file_entry.get(algo)
),
None,
)
if declared is None:
print(f" WARNING: no hash for {name}, skipping unverifiable download")
return False
if _OFFLINE if offline is None else offline:
print(f" WARNING: {name} is external and the run is offline, not fetched")
return False
algo, expected = declared
digest = hashlib.new(algo)
Path(dest_path).parent.mkdir(parents=True, exist_ok=True)
try:
req = urllib.request.Request(url, headers={"User-Agent": "retrobios-pack-gen/1.0"})
with urllib.request.urlopen(req, timeout=120) as resp, open(dest_path, "wb") as out:
for chunk in iter(lambda: resp.read(1 << 20), b""):
digest.update(chunk)
out.write(chunk)
except (urllib.error.URLError, OSError) as exc:
print(f" WARNING: Failed to download {url}: {exc}")
Path(dest_path).unlink(missing_ok=True)
return False
if digest.hexdigest() != expected:
print(f" WARNING: {algo.upper()} mismatch for {name}")
Path(dest_path).unlink(missing_ok=True)
return False
return True
def _values(raw: object) -> list[str]:
"""A hash field's values: one, a list, or a comma-separated string."""
items = raw if isinstance(raw, list) else [raw] if raw else []
+23 -9
View File
@@ -382,7 +382,7 @@ def check_consistency(verify_output: str, pack_output: str) -> bool:
all_ok = False
status = "OK" if all_ok else "FAILED"
print(f"--- consistency check: {status} ---")
print(f"--- {step_title(STEP_LABELS['consistency'])}, counts: {status} ---")
return all_ok
@@ -447,7 +447,7 @@ def check_pack_destinations(
import json # noqa: PLC0415
import zipfile # noqa: PLC0415
print("\n--- 5/8 consistency check: destinations ---")
print(f"\n--- {step_title(STEP_LABELS['consistency'])}, destinations ---")
try:
credited_by = json.loads(Path(destinations_file).read_text())
except (OSError, ValueError) as exc:
@@ -506,7 +506,7 @@ def check_emulator_packs(
select_emulator_profiles,
)
print("\n--- 5/8 consistency check: emulator packs ---")
print(f"\n--- {step_title(STEP_LABELS['consistency'])}, emulator packs ---")
db = load_database(db_path)
zip_contents = build_zip_contents_index(db)
profiles = load_emulator_profiles(emulators_dir, skip_aliases=False)
@@ -561,13 +561,18 @@ def _refuse_unapplied_flags(
--all-variants builds every source itself, so a --source beside it is
dropped; --skip-packs builds none, so both are.
"""
if args.all_variants and args.source != "full":
# --source defaults to None so a typed `--source full` is seen: compared
# with its default value, it went through beside the modes that refuse it.
source_given = args.source is not None
if args.all_variants and source_given:
parser.error("--source is incompatible with --all-variants")
if args.skip_packs:
for flag, on in (("--source", args.source != "full"),
for flag, on in (("--source", source_given),
("--all-variants", args.all_variants)):
if on:
parser.error(f"{flag} is incompatible with --skip-packs")
if args.source is None:
args.source = "full"
def _truth_steps(args: argparse.Namespace, results: dict) -> bool:
@@ -662,9 +667,9 @@ def main() -> None:
)
parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)")
parser.add_argument(
"--source", choices=["platform", "truth", "full"], default="full",
"--source", choices=["platform", "truth", "full"], default=None,
help="packs to build: the platform list, the profiles, or both (default: "
"%(default)s)",
"full)",
)
parser.add_argument(
"--all-variants", action="store_true",
@@ -724,7 +729,13 @@ 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:
skip_notice("validate_schemas", "jsonschema not installed")
# The skip names the install and what still runs the check, or a
# contributor read SKIPPED, then COMPLETE, and met the failure on push.
skip_notice(
"validate_schemas",
"jsonschema not installed. PR validation runs this check: "
'pip install "jsonschema[format-nongpl]"',
)
results["validate_schemas"] = SKIPPED
else:
ok, _ = run(
@@ -1011,7 +1022,10 @@ def main() -> None:
# Counts can agree while the paths differ, both ways at once.
ok = check_pack_destinations(destinations_file, pack_output) and ok
ok = check_emulator_packs() and ok
print(f"--- 5/8 consistency check, counts and paths: {'OK' if ok else 'FAILED'} ---")
print(
f"--- {step_title(STEP_LABELS['consistency'])}, counts and paths: "
f"{'OK' if ok else 'FAILED'} ---"
)
results["consistency"] = ok
all_ok = all_ok and ok
else:
+100 -29
View File
@@ -30,7 +30,13 @@ import yaml
sys.path.insert(0, os.path.dirname(__file__))
import upstream
from common import github_token, load_emulator_profiles, suggest_names
from common import (
_norm_system_id,
github_token,
load_emulator_profiles,
suggest_names,
system_index,
)
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,
@@ -488,9 +494,17 @@ def _find_all(haystack: list[str], needle: list[str]) -> list[int]:
def _map_index(
opcodes: list[tuple[str, int, int, int, int]], index: int
) -> int | None:
"""Line of HEAD that a pinned line became.
A replace block of equal size edits each line in place (a column added to
every row of a table), so its lines map one to one like an equal block.
Collapsed onto the first line, eight refs citing eight rows all landed on
the first row.
"""
for tag, i1, i2, j1, j2 in opcodes:
if i1 <= index < i2:
return j1 + (index - i1) if tag == "equal" else j1
in_place = tag == "equal" or (tag == "replace" and i2 - i1 == j2 - j1)
return j1 + (index - i1) if in_place else j1
return None
@@ -958,6 +972,14 @@ class ProfileReport:
# release, kept on the report so every output carries it.
version: VersionReport | None = None
version_warning: str | None = None
# --detect-new-files and --watch-hashes, kept on the report so --json
# carries them to check_freshness. None when not asked. A new file is
# {path, line, name, since_pin}: since_pin when the pinned revision of
# the same file does not carry the name. A new hash is {path, kind, value}.
new_files: list[dict] | None = None
new_files_elided: dict[str, int] = field(default_factory=dict)
new_hashes: list[dict] | None = None
detection_skipped: str | None = None
# Skips that say the profile cannot be checked as written: a repository
# that no longer answers, no pin to read, a death the forge contradicts.
@@ -3573,11 +3595,16 @@ def select_profiles(profiles: dict, args: argparse.Namespace) -> dict:
)
return {args.emulator: profiles[args.emulator]}
# --system matches as verify.py --list-systems lists and --system selects:
# by normalized id, so `saturn` reaches the profiles declaring sega-saturn.
wanted_system = _norm_system_id(args.system) if args.system else None
selected = {}
for name, profile in sorted(profiles.items()):
if profile.get("type") in ("alias", "test"):
continue
if args.system and args.system not in (profile.get("systems") or []):
if wanted_system and wanted_system not in {
_norm_system_id(str(s)) for s in profile.get("systems") or []
}:
continue
if args.type_ and profile.get("type") != args.type_:
continue
@@ -3599,7 +3626,8 @@ def select_profiles(profiles: dict, args: argparse.Namespace) -> dict:
)
hint = ""
if args.system:
systems = sorted({s for p in profiles.values() for s in p.get("systems") or []})
# Suggested from the spellings the named listing prints.
systems = sorted(spelling for spelling, _ in system_index(profiles).values())
hint = " " + suggest_names([args.system], systems, "verify.py --list-systems")
elif args.type_:
kinds = sorted({str(p.get("type")) for p in profiles.values() if p.get("type")})
@@ -3748,37 +3776,84 @@ def _cited_paths(report: ProfileReport) -> list[str]:
)
def _print_detection(
def _attach_detection(
args: argparse.Namespace, profile: dict, report: ProfileReport
) -> None:
"""Record undeclared names and hashes of the cited files on the report.
Kept on the report rather than printed, so --json carries them as the
text report does. A forge that refuses is named on the report: the ref
report already muted it, and one refusal must not end a pass over every
profile. A quota signal still stops everything.
"""
if not report.head or not (args.detect_new_files or args.watch_hashes):
return
repo = select_repo(profile)
if repo is None:
return
try:
_detect(args, profile, report, repo)
except upstream.RateLimitError:
raise
except upstream.UpstreamError as exc:
report.detection_skipped = str(exc)
def _detect(
args: argparse.Namespace, profile: dict, report: ProfileReport,
repo: upstream.Repo,
) -> None:
names, hashes = declared_names(profile), declared_hashes(profile)
new_files: list[dict] = []
new_hashes: list[dict] = []
for path in _cited_paths(report):
head_lines = upstream.fetch_file(
repo, report.head, path, args.cache_dir, args.offline
)
if head_lines is None:
continue
pin_lines = (
upstream.fetch_file(repo, report.pin, path, args.cache_dir, args.offline)
or []
)
if args.detect_new_files:
found, elided = detect_new_files(head_lines, names)
for number, name in found:
print(f" new file: {path}:{number} {name}")
if elided:
print(f" new file: {elided} further matches not shown")
if args.watch_hashes:
pin_lines = (
upstream.fetch_file(
repo, report.pin, path, args.cache_dir, args.offline
)
or []
at_pin = {
name.casefold()
for _, name in detect_new_files(pin_lines, names, cap=sys.maxsize)[0]
}
new_files.extend(
{"path": path, "line": number, "name": name,
"since_pin": name.casefold() not in at_pin}
for number, name in found
)
if elided:
report.new_files_elided[path] = elided
if args.watch_hashes:
added = [
line
for line in difflib.unified_diff(pin_lines, head_lines, n=0)
if line.startswith("+") and not line.startswith("+++")
]
for kind, value in watch_hashes(added, hashes):
print(f" new {kind}: {path} {value}")
new_hashes.extend(
{"path": path, "kind": kind, "value": value}
for kind, value in watch_hashes(added, hashes)
)
if args.detect_new_files:
report.new_files = new_files
if args.watch_hashes:
report.new_hashes = new_hashes
def _print_detection(report: ProfileReport) -> None:
if report.detection_skipped:
print(f" extras skipped: {report.detection_skipped}")
for item in report.new_files or ():
print(f" new file: {item['path']}:{item['line']} {item['name']}")
for path, elided in report.new_files_elided.items():
print(f" new file: {path}: {elided} further matches not shown")
for item in report.new_hashes or ():
print(f" new {item['kind']}: {item['path']} {item['value']}")
def _attach_version(
@@ -3812,13 +3887,8 @@ def _print_extras(
instead of ending a pass over every profile. A quota signal still
stops everything.
"""
wants = (
args.detect_new_files
or args.watch_hashes
or args.full_diff
or args.tree_diff
)
if not wants or not report.head:
_print_detection(report)
if not (args.full_diff or args.tree_diff) or not report.head:
return
repo = select_repo(profile)
if repo is None:
@@ -3835,8 +3905,6 @@ def _print_extras_from(
args: argparse.Namespace, profile: dict, report: ProfileReport,
repo: upstream.Repo,
) -> None:
if args.detect_new_files or args.watch_hashes:
_print_detection(args, profile, report, repo)
if args.full_diff:
for path in _cited_paths(report):
if args.ref and args.ref not in path:
@@ -3983,13 +4051,15 @@ def _refuse_flag_combinations(args: argparse.Namespace) -> None:
]
# The version check lives on the report, so the outputs built from the
# report (text, --json, --markdown) carry it. A URL list and a ranking
# do not.
# do not. Detection lives on the report too, and --json carries it for
# check_freshness. The markdown table has no column for it.
lists_only = args.fetch_plan or args.triage
off_json = lists_only or args.markdown
extras = [
flag for flag, on in (
("--check-version", args.check_version and lists_only),
("--detect-new-files", args.detect_new_files),
("--watch-hashes", args.watch_hashes),
("--detect-new-files", args.detect_new_files and off_json),
("--watch-hashes", args.watch_hashes and off_json),
("--full-diff", args.full_diff),
("--tree-diff", args.tree_diff),
) if on
@@ -4086,6 +4156,7 @@ def main() -> None:
profile["version_source"].get("no_genuine_copy") or "") or None
if args.check_version:
_attach_version(args, profile, report)
_attach_detection(args, profile, report)
reports.append(report)
if writes:
try:
+3 -1
View File
@@ -578,7 +578,9 @@ def main() -> None:
parser.add_argument(
"--dry-run", action="store_true", help="Preview without downloading"
)
parser.add_argument("--platform", type=platform_arg, help="Only refresh entries for this platform")
parser.add_argument(
"--platform", "-p", type=platform_arg, help="Only refresh entries for this platform"
)
parser.add_argument(
"--registry", default=DEFAULT_REGISTRY, help="Path to _data_dirs.yml"
)
+2 -2
View File
@@ -897,7 +897,7 @@ def main() -> int:
"the file a platform baseline would ship.",
)
parser.add_argument(
"--platform", type=platform_arg,
"--platform", "-p", type=platform_arg,
help="one platform instead of all",
)
parser.add_argument(
@@ -918,7 +918,7 @@ def main() -> int:
help="with all platforms, the archived ones too (RetroPie)",
)
parser.add_argument(
"--target", help="judge only the cores a hardware target builds",
"--target", "-t", help="judge only the cores a hardware target builds",
)
parser.add_argument(
"--strict",
+132
View File
@@ -3975,6 +3975,8 @@ class EveryToolRefusesAFlagItWouldDrop(unittest.TestCase):
["--offline", "--skip-packs", "--source", "truth"],
["--offline", "--skip-packs", "--all-variants"],
["--offline", "--all-variants", "--source", "platform"],
["--offline", "--all-variants", "--source", "full"],
["--offline", "--skip-packs", "--source", "full"],
],
}
@@ -6486,5 +6488,135 @@ class CommentsSayWhy(unittest.TestCase):
self.assertEqual(found, [])
class ExternalDownloadsFollowTheRunPolicy(unittest.TestCase):
"""download_external lived outside packresolve and fetched under
--offline, into the system /tmp, holding the whole body in memory."""
ENTRY = {"name": "ext.bin", "source_url": "https://example.invalid/ext.bin",
"sha1": hashlib.sha1(b"payload").hexdigest()}
def _response(self, body: bytes):
stream = io.BytesIO(body)
response = mock.MagicMock()
response.__enter__.return_value = stream
response.__exit__.return_value = False
return response
def test_offline_never_reaches_the_network(self):
import packresolve
with tempfile.TemporaryDirectory() as tmp, \
mock.patch.object(packresolve.urllib.request, "urlopen") as urlopen, \
contextlib.redirect_stdout(io.StringIO()):
ok = packresolve.download_external(self.ENTRY, f"{tmp}/x", offline=True)
self.assertFalse(ok)
urlopen.assert_not_called()
def test_the_run_switch_is_inherited(self):
import packresolve
saved = packresolve._OFFLINE
packresolve.set_offline(True)
try:
with tempfile.TemporaryDirectory() as tmp, \
mock.patch.object(packresolve.urllib.request, "urlopen") as urlopen, \
contextlib.redirect_stdout(io.StringIO()):
self.assertFalse(packresolve.download_external(self.ENTRY, f"{tmp}/x"))
urlopen.assert_not_called()
finally:
packresolve.set_offline(saved)
def test_a_verified_body_lands_and_a_wrong_one_is_removed(self):
import packresolve
with tempfile.TemporaryDirectory() as tmp:
target = Path(tmp) / "x"
with mock.patch.object(packresolve.urllib.request, "urlopen",
return_value=self._response(b"payload")):
self.assertTrue(packresolve.download_external(self.ENTRY, str(target), offline=False))
self.assertEqual(target.read_bytes(), b"payload")
target.unlink()
with mock.patch.object(packresolve.urllib.request, "urlopen",
return_value=self._response(b"other")), \
contextlib.redirect_stdout(io.StringIO()):
self.assertFalse(packresolve.download_external(self.ENTRY, str(target), offline=False))
self.assertFalse(target.exists())
def test_scratch_files_stay_in_the_repository(self):
source = (ROOT / "scripts" / "generate_pack.py").read_text(encoding="utf-8")
self.assertNotIn("def download_external", source)
self.assertNotIn("urlopen", source)
for call in re.findall(r"NamedTemporaryFile\(([^)]*)\)", source):
self.assertIn("dir=", call)
class ATargetTypoIsAnErrorInEveryTool(unittest.TestCase):
"""cross_reference.py ended a mistyped --target in a traceback, and the
hint sent cross_reference and slots users to a flag they do not have."""
def test_each_tool_names_the_closest_target(self):
for script in ("cross_reference.py", "slots.py", "verify.py"):
with self.subTest(script=script):
proc = subprocess.run(
[sys.executable, f"scripts/{script}", "--platform", "retroarch",
"--target", "swich"],
cwd=ROOT, capture_output=True, text=True, timeout=300,
)
output = proc.stdout + proc.stderr
self.assertNotEqual(proc.returncode, 0)
self.assertNotIn("Traceback", output)
self.assertIn("Did you mean: switch?", output)
self.assertIn("verify.py --platform retroarch --list-targets", output)
class AnUnknownSystemIsRefusedBeforeAnyWork(unittest.TestCase):
"""generate_pack checked --system inside the build: a typo took the
output lock and refreshed every data directory, then ended on a WARNING."""
def test_refused_before_the_lock_and_the_refresh(self):
with tempfile.TemporaryDirectory(dir=ROOT / "tmp") as tmp:
out = Path(tmp) / "never"
with mock.patch.dict(os.environ, {"RETROBIOS_OFFLINE": ""}):
proc = subprocess.run(
[sys.executable, "scripts/generate_pack.py", "--platform", "batocera",
"--system", "playstaton-nothing", "--output-dir", str(out)],
cwd=ROOT, capture_output=True, text=True, timeout=120,
)
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
self.assertIn("error: --system playstaton-nothing matches no system", proc.stderr)
self.assertFalse(out.exists())
def test_a_known_system_passes_the_early_check(self):
import argparse
import generate_pack
parser = argparse.ArgumentParser()
generate_pack._refuse_unknown_systems(
["sony-playstation"], ["retroarch"], str(ROOT / "platforms"), parser
)
class NoArchiveRepeatsAMemberName(unittest.TestCase):
"""s8k_cpu.zip held three member names twice (MAME reuses them across
two ROM_SYSTEM_BIOS of one device). Every pack build printed raw zipfile
warnings and Info-ZIP stops on an interactive replace? at extraction."""
def test_member_names_are_unique(self):
archives = sorted((ROOT / "bios").rglob("*.zip"))
if not archives:
self.skipTest("no archive under bios/")
repeated = []
for path in archives:
try:
with zipfile.ZipFile(path) as zf:
names = zf.namelist()
except (zipfile.BadZipFile, OSError):
continue
if len(names) != len(set(names)):
repeated.append(str(path.relative_to(ROOT)))
self.assertEqual(repeated, [])
if __name__ == "__main__":
unittest.main()
+23
View File
@@ -391,6 +391,29 @@ class ProfileSummaryTests(unittest.TestCase):
self.assertNotIn("a", by_subject)
self.assertNotIn("b", by_subject)
def test_new_upstream_names_and_hashes_are_stale(self):
"""A file name the cited code gained since the pin, or a hash nobody
declares, was printed only by the text pass that --profiles never ran."""
report = [
{"name": "a", "pin": "1" * 40, "head": "2" * 40, "needs_review": 0,
"new_files": [
{"path": "x.c", "line": 4, "name": "old.rom", "since_pin": False},
{"path": "x.c", "line": 9, "name": "new.rom", "since_pin": True},
], "new_hashes": []},
{"name": "b", "pin": "1" * 40, "head": "2" * 40, "needs_review": 0,
"new_files": [], "new_hashes": [{"path": "x.c", "kind": "md5", "value": "0" * 32}]},
{"name": "c", "pin": "1" * 40, "head": "2" * 40, "needs_review": 0,
"new_files": [
{"path": "x.c", "line": 4, "name": "old.rom", "since_pin": False},
], "new_hashes": []},
]
by_subject = {f.subject: f for f in cf.summarize_profile_sync(report)}
self.assertEqual(by_subject["a"].status, cf.STALE)
self.assertIn("new.rom", by_subject["a"].detail)
self.assertNotIn("old.rom", by_subject["a"].detail)
self.assertEqual(by_subject["b"].status, cf.STALE)
self.assertNotIn("c", by_subject)
def test_profile_sync_accepts_the_command_it_is_given(self):
"""--json with the detection flags was refused, so --profiles read an
empty stdout and reported "unreadable report" on every run."""
+37
View File
@@ -2744,6 +2744,31 @@ class TestE2E(unittest.TestCase):
self.assertIn("bios/scph5501.bin", names)
self.assertIn("Custom", os.path.basename(zip_path))
# A profile displayed under the platform's own name: BizHawk is
# both. The two layouts must not write one file.
emu_dir = os.path.join(tmpdir, "emulators-md5")
os.makedirs(emu_dir)
with open(os.path.join(emu_dir, "testplat.yml"), "w") as f:
yaml.dump({
"emulator": "TestPlat", "type": "standalone",
"systems": ["sony-playstation"],
"files": [{"name": "scph5501.bin", "md5": h["md5"]}],
}, f)
emu_zip = generate_md5_pack(
hashes=[("md5", h["md5"])],
db=db,
bios_dir=bios_dir,
output_dir=out_dir,
zip_contents=zip_contents,
emulator_name="testplat",
emulators_dir=emu_dir,
)
self.assertIsNotNone(emu_zip)
self.assertNotEqual(os.path.basename(emu_zip), os.path.basename(zip_path))
self.assertTrue(os.path.exists(zip_path))
with zipfile.ZipFile(emu_zip) as zf:
self.assertIn("scph5501.bin", zf.namelist())
def test_143_from_md5_not_in_repo(self):
"""--from-md5 reports files in DB but missing from repo."""
import contextlib
@@ -6016,6 +6041,18 @@ struct BurnDriver BurnDrvneogeo = {
"refuse"),
(["--platform", "retroarch", "--all-variants"], ["--required-only"],
"refuse"),
# The default value typed out is still the flag: compared with its
# default, `--source full` and `--group-by system` went through.
(["--platform", "retroarch", "--all-variants"], ["--source", "full"],
"refuse"),
(["--manifest-targets"], ["--source", "full"], "refuse"),
(["--emulator", "handy"], ["--source", "full"], "refuse"),
(["--platform", "retroarch", "--from-md5", "d8f1"],
["--source", "full"], "refuse"),
(["--platform", "retroarch", "--offline"], ["--group-by", "system"],
"refuse"),
(["--platform", "retroarch", "--verify-packs"], ["--group-by", "system"],
"refuse"),
(["--emulator", "duckstation"], ["--all-variants"], "refuse"),
(["--system", "sony-playstation"], ["--all-variants"], "refuse"),
(["--emulator", "duckstation", "--verify-packs"], ["--all-variants"],
+36 -2
View File
@@ -37,6 +37,22 @@ PAYLOAD_B = b"B" * (256 * 1024)
# network, so it is cleared here.
_ONLINE = mock.patch.dict(os.environ, {"RETROBIOS_OFFLINE": ""})
# The names these tests fetch, registered as .gitignore registers an asset:
# a bare name no registered path carries is never asked of the release.
TEST_REGISTERED = [
"bios/Test/asset.bin",
"bios/Test/big.bin",
"bios/Sony/PlayStation 3/PS3UPDAT.PUP",
]
def _test_registry(case: unittest.TestCase) -> None:
patcher = mock.patch.object(
largefiles, "load_registered_paths", return_value=TEST_REGISTERED
)
patcher.start()
case.addCleanup(patcher.stop)
def setUpModule():
_ONLINE.start()
@@ -71,6 +87,7 @@ class _SlowResponse(io.BytesIO):
class LargeFileCacheTest(unittest.TestCase):
def setUp(self):
_test_registry(self)
self.dir = tempfile.mkdtemp()
self._urlopen = largefiles.urllib.request.urlopen
@@ -237,6 +254,9 @@ class LargeFileCacheTest(unittest.TestCase):
class TheOfflineSwitchHoldsEverywhere(unittest.TestCase):
"""RETROBIOS_OFFLINE stops a download whatever the caller asked."""
def setUp(self):
_test_registry(self)
def test_no_request_is_made(self):
with tempfile.TemporaryDirectory() as tmp, mock.patch.dict(
os.environ, {"RETROBIOS_OFFLINE": "1"}
@@ -600,7 +620,7 @@ class DownloadFailuresAndRevisions(unittest.TestCase):
largefiles.urllib.request.urlopen = lambda *_a, **_k: _StalledResponse()
result = largefiles.fetch_large_file(
"big.bin", self.dir, expected_md5=hashlib.md5(PAYLOAD_A).hexdigest(),
registered=[],
registered=TEST_REGISTERED,
)
self.assertIsNone(result)
self.assertEqual(os.listdir(self.dir), [])
@@ -613,7 +633,7 @@ class DownloadFailuresAndRevisions(unittest.TestCase):
)
result = largefiles.fetch_large_file(
"PS3UPDAT.PUP", self.dir, expected_md5=hashlib.md5(PAYLOAD_B).hexdigest(),
registered=[],
registered=TEST_REGISTERED,
)
self.assertIsNone(result)
self.assertTrue(cached.exists())
@@ -629,6 +649,20 @@ class DownloadFailuresAndRevisions(unittest.TestCase):
)
class AnUnregisteredNameIsNoAsset(unittest.TestCase):
"""aes.zip and five others are in no .gitignore line: each online build
asked the release for them and printed an HTTP 404 per name."""
def test_no_candidate_and_no_request(self):
self.assertEqual(largefiles.asset_candidates("aes.zip", TEST_REGISTERED), [])
with tempfile.TemporaryDirectory() as tmp, \
mock.patch.object(largefiles.urllib.request, "urlopen") as urlopen:
self.assertIsNone(largefiles.fetch_large_file(
"aes.zip", dest_dir=tmp, offline=False, registered=TEST_REGISTERED,
))
urlopen.assert_not_called()
class GameDataHasOneDefinition(unittest.TestCase):
"""The release notes and the composition table sorted game data apart.
+89 -4
View File
@@ -378,6 +378,15 @@ class TestAnchorBlock(unittest.TestCase):
self.assertEqual(result.status, "CHANGED")
self.assertEqual(result.start, 3)
def test_rows_edited_in_place_keep_their_own_line(self):
rows = [f'{{"ipl{i}", 0x{i:02x}}}' for i in range(8)]
pin = ["a", "b", *rows, "z"]
head = ["a", "b", *[row + ", false" for row in rows], "z"]
for line in (3, 5, 10):
result = anchor_block(pin, head, line, line)
self.assertEqual(result.status, "CHANGED")
self.assertEqual(result.start, line)
def test_gone_when_content_vanished(self):
result = anchor_block(["a", "unique marker", "b"], ["x"] * 200, 2, 2)
self.assertEqual(result.status, "GONE")
@@ -2014,13 +2023,42 @@ class TestExtrasSurviveARefusingForge(unittest.TestCase):
upstream.fetch_file = refuse
try:
report = _sample_report()
profile_sync._attach_detection(self._args(), self._profile(), report)
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
profile_sync._print_extras(self._args(), self._profile(), _sample_report())
profile_sync._print_extras(self._args(), self._profile(), report)
finally:
upstream.fetch_file = original
self.assertEqual(report.detection_skipped, "https://x/y: HTTP 403")
self.assertIn("extras skipped: https://x/y: HTTP 403", buffer.getvalue())
def test_detection_lands_on_the_report(self):
"""A name the pinned file already carried is listed, but only a name
the pin did not carry is new since the pin."""
files = {
"pinsha0000": ['load("old.rom");'],
"headsha000": ['load("old.rom");', 'load("added.rom"); /* 0xdeadbeef */'],
}
original = upstream.fetch_file
upstream.fetch_file = lambda repo, rev, path, *a, **k: files[rev]
args = self._args()
args.watch_hashes = True
report = _sample_report()
try:
profile_sync._attach_detection(args, self._profile(), report)
finally:
upstream.fetch_file = original
by_name = {(i["path"], i["name"]): i["since_pin"] for i in report.new_files}
self.assertEqual(
by_name,
{("a.c", "old.rom"): False, ("a.c", "added.rom"): True,
("b.c", "old.rom"): False, ("b.c", "added.rom"): True},
)
self.assertIn(
{"path": "a.c", "kind": "crc32", "value": "deadbeef"}, report.new_hashes
)
def test_a_quota_signal_still_stops_the_pass(self):
original = upstream.fetch_file
@@ -2030,7 +2068,9 @@ class TestExtrasSurviveARefusingForge(unittest.TestCase):
upstream.fetch_file = quota
try:
with self.assertRaises(upstream.RateLimitError):
profile_sync._print_extras(self._args(), self._profile(), _sample_report())
profile_sync._attach_detection(
self._args(), self._profile(), _sample_report()
)
finally:
upstream.fetch_file = original
@@ -4103,11 +4143,35 @@ class TestOutputModesApplyTheirFlags(unittest.TestCase):
self.assertIn("clean", out)
self.assertIn("prose under an older pin", out)
def test_extras_are_refused_with_json(self):
code, _, err = self._main("--json", "--detect-new-files")
def test_diffs_are_refused_with_json(self):
code, _, err = self._main("--json", "--full-diff")
self.assertEqual(code, 1)
self.assertIn("--full-diff", err)
def test_markdown_refuses_detection(self):
code, _, err = self._main("--markdown", "--detect-new-files")
self.assertEqual(code, 1)
self.assertIn("--detect-new-files", err)
def test_json_carries_detection(self):
"""check_freshness reads --json: a new upstream file or hash printed
only in the text report stayed a human reading."""
from unittest import mock
def detect(args, profile, report):
report.new_files = [
{"path": "a.c", "line": 3, "name": "new.rom", "since_pin": True}
]
report.new_hashes = [{"path": "a.c", "kind": "md5", "value": "0" * 32}]
with mock.patch.object(profile_sync, "_attach_detection", side_effect=detect):
code, out, err = self._main("--json", "--detect-new-files", "--watch-hashes")
self.assertEqual(code, 0, err)
rows = json.loads(out)
self.assertEqual(rows[0]["new_files"][0]["name"], "new.rom")
self.assertTrue(rows[0]["new_files"][0]["since_pin"])
self.assertEqual(rows[0]["new_hashes"][0]["kind"], "md5")
def test_json_carries_the_version_check(self):
"""check_freshness asks for --json --check-version: refused, its
profile row read an empty report as unreadable on every run."""
@@ -4335,6 +4399,27 @@ class TestSelectionNamesWhatMatchedNothing(unittest.TestCase):
selected = profile_sync.select_profiles(self.PROFILES, self._args(system="atari-lynx"))
self.assertEqual(sorted(selected), ["handy", "mednafen_lynx"])
def test_system_matches_as_the_listing_does(self):
"""verify.py --list-systems printed `saturn`, which selected 1 profile
of 11 because 10 declare sega-saturn."""
profiles = {
"a": {"type": "libretro", "systems": ["sega-saturn"]},
"b": {"type": "libretro", "systems": ["saturn"]},
}
for spelling in ("saturn", "sega-saturn"):
selected = profile_sync.select_profiles(profiles, self._args(system=spelling))
self.assertEqual(sorted(selected), ["a", "b"])
def test_the_hint_counts_what_the_listing_prints(self):
profiles = {
"a": {"type": "libretro", "systems": ["sega-saturn"]},
"b": {"type": "libretro", "systems": ["saturn"]},
"c": {"type": "libretro", "systems": ["atari-lynx"]},
}
with self.assertRaises(SystemExit) as caught:
profile_sync.select_profiles(profiles, self._args(system="nothing"))
self.assertIn("Known: atari-lynx, sega-saturn.", str(caught.exception.code))
class TestTreeUrlIntoADirectory(unittest.TestCase):
+56
View File
@@ -145,6 +145,33 @@ class TheVerifySummaryCountsDiscrepancies(unittest.TestCase):
self.assertIn(part, message)
class OneSpellingPerOption(unittest.TestCase):
"""verify.py -p batocera worked and slots.py -p batocera was refused."""
SHORT = {"--platform": "-p", "--target": "-t"}
# migrate.py's --target is the directory files move to, not a hardware target.
OTHER_MEANING = {("migrate.py", "--target")}
def test_every_script_takes_the_short_forms(self):
import ast # noqa: PLC0415
missing = []
for path in sorted((REPO_ROOT / "scripts").rglob("*.py")):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if not (isinstance(node, ast.Call)
and getattr(node.func, "attr", None) == "add_argument"):
continue
flags = [a.value for a in node.args
if isinstance(a, ast.Constant) and isinstance(a.value, str)]
for long, short in self.SHORT.items():
if (path.name, long) in self.OTHER_MEANING:
continue
if long in flags and short not in flags:
missing.append(f"{path.relative_to(REPO_ROOT)}:{node.lineno} {long}")
self.assertEqual(missing, [])
class ThePipelineCountsItsRealSteps(unittest.TestCase):
"""Labels read N/8 for a run of 24 steps, and the summary used keys
the log never printed."""
@@ -167,12 +194,41 @@ class ThePipelineCountsItsRealSteps(unittest.TestCase):
# A step's own label, or a second command of a listed step.
self.assertIn(label.split(" ", 1)[0], numbers)
def test_no_title_counts_eight_steps(self):
"""Three print() headers of step 5 still read "5/8" beside the
computed "5 consistency check [20/24]"."""
import re # noqa: PLC0415
source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
self.assertEqual(re.findall(r'"[^"\n]*---[^"\n]*\b\d+[a-z]?/\d+\b[^"\n]*"', source), [])
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_a_skipped_schema_check_says_how_to_run_it(self):
"""Step 0 said only "jsonschema not installed", then the run read
COMPLETE and the schema failed in PR validation."""
argv = ["pipeline.py", "--offline", "--skip-packs", "--skip-docs"]
out = io.StringIO()
real_find_spec = pipeline.importlib.util.find_spec
with mock.patch.object(sys, "argv", argv), mock.patch.object(
pipeline, "run", return_value=(True, "")
), mock.patch.object(
pipeline.importlib.util, "find_spec",
side_effect=lambda name, *a: None if name == "jsonschema"
else real_find_spec(name, *a),
), contextlib.redirect_stdout(out), contextlib.suppress(SystemExit):
pipeline.main()
line = next(
row for row in out.getvalue().splitlines()
if "0 validate schemas" in row and "SKIPPED" in row
)
self.assertIn('pip install "jsonschema[format-nongpl]"', line)
self.assertIn("PR validation runs this check", line)
def test_an_offline_buildbot_check_has_a_summary_row(self):
argv = ["pipeline.py", "--offline", "--check-buildbot", "--skip-packs",
"--skip-docs"]
+24
View File
@@ -115,6 +115,30 @@ class VerifyPackChoice(unittest.TestCase):
for name in names[1:]:
self.assertIn(f"[{name}]: hashes checked during the build", printed)
def test_the_run_ends_on_a_summary(self):
"""A failing pack was named only above its own error lines, and the
log ended on the last platform's OK line."""
import contextlib
import io
pack = f"{self.stem}_BIOS_Pack.zip"
self._touch(pack)
out, err = io.StringIO(), io.StringIO()
failed = (False, 0, 0, ["x: missing"], 1, 0, 1, 1, 0)
with mock.patch.object(self.gp, "verify_pack_against_platform",
return_value=failed), \
contextlib.redirect_stdout(out), contextlib.redirect_stderr(err), \
self.assertRaises(SystemExit):
self.gp._run_verify_packs(self._args())
self.assertEqual(
err.getvalue().strip().splitlines()[-1],
f"ERROR: packs failed verification: retroarch ({pack})",
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
self._run(self._args())
self.assertEqual(out.getvalue().strip().splitlines()[-1], "Verified 1 pack: all OK")
def test_include_archived_reaches_the_archived_pack(self):
stem = self.gp._platform_pack_stem(["retropie"], "retropie", "platforms")
self._touch(f"{stem}_BIOS_Pack.zip")
+5 -2
View File
@@ -530,8 +530,11 @@ reported as `RENAMED` so `--rebase-refs` cleans the profile.
`--check-version` compares `core_version` with the latest upstream tag and
release. `--detect-new-files` lists filename literals at HEAD the profile
does not declare. `--watch-hashes` lists hash literals added upstream that
match no entry. `--tree-diff` shows added, removed and renamed files in the
directories the refs point at.
match no entry. Both land in the `--json` report (`new_files`, with
`since_pin` true for a name the pinned file did not carry, and
`new_hashes`), which `check_freshness.py --profiles` reads: a name gained
since the pin or an undeclared hash makes the profile stale. `--tree-diff`
shows added, removed and renamed files in the directories the refs point at.
Writes are explicit and mechanical only. `--backfill-commits` fills a
missing `source_commit`, `--rebase-refs` recales `SHIFTED` and `RENAMED`