refactor: single-source the native mode policy

verify.py and generate_pack.py must reach the same verdict on the same
file, and CLAUDE.md calls any divergence critical, but each spelled out
mode == 'existence' in its own words. Both now ask nativemode whether
the frontend reads the file's bytes, which is the one fact the rest
follows from, and a test holds the two answers together.

The BaseScraper contract the wiki asks contributors to implement had no
caller anywhere, so nothing proved compare_with_config, has_changes or
test_connection still worked.
This commit is contained in:
Abdessamad Derraz committed 2026-08-11 00:54:46 +02:00
1 parent 057b0c18f2
commit ab6a3bb26d
4 files changed
+380 -6

No files matched your search

+60
View File
@@ -0,0 +1,60 @@
#!/usr/bin/env python3
"""The one place that decides what a platform's native check actually does.
verify.py reports coverage and generate_pack.py decides what goes in the ZIP.
They have to reach the same verdict on the same file or the pack and the
report describe different collections. The rule they share is small:
does the frontend read the file's bytes, or only look for the name?
Everything downstream follows from that answer. RetroArch, Lakka and RetroPie
call path_is_valid() and never open the file, so a declared hash contradicted
by the local dump is a documentation error, the file still loads, and the
builder ships it while the report flags the divergence. Batocera, Recalbox,
RetroBat, RetroDECK, RomM, ROCKNIX, MiSTer and BizHawk compare a digest, so
the same file would be rejected at runtime and shipping it would be shipping
a known failure.
Keeping the predicate here rather than in each caller is what makes
"verify.py and generate_pack.py must agree" a property of the code instead of
something a reviewer has to notice.
"""
from __future__ import annotations
DEFAULT_MODE = "existence"
#: Every verification mode a platform YAML may declare. Mirrors the enum in
#: schemas/platform.schema.json; tests/test_audit_regressions.py pins them
#: together so a new mode cannot be added to one without the other.
MODES = ("existence", "md5", "sha1")
#: The digest each content-checking mode compares. Modes absent from this map
#: do not read the file at all.
MODE_DIGEST = {"md5": "md5", "sha1": "sha1"}
def normalize(mode: str | None) -> str:
"""Return a declared mode, falling back to the schema default."""
return mode if mode in MODES else DEFAULT_MODE
def reads_file_contents(mode: str | None) -> bool:
"""Whether the frontend opens the file rather than just finding it."""
return normalize(mode) in MODE_DIGEST
def digest_algorithm(mode: str | None) -> str | None:
"""The hash this mode compares, or None when it compares nothing."""
return MODE_DIGEST.get(normalize(mode))
def hash_mismatch_excludes_file(mode: str | None) -> bool:
"""Whether a local dump contradicting a declared hash must be omitted.
True for the digest modes: the frontend would reject the file, so packing
it ships a known failure. False for existence: the frontend never looks,
and dropping the file over an upstream metadata error would remove
something that works.
"""
return reads_file_contents(mode)
+80 -6
View File
@@ -52,6 +52,7 @@ from common import (
) )
yaml = require_yaml() yaml = require_yaml()
from nativemode import reads_file_contents
from validation import ( from validation import (
_build_validation_index, _build_validation_index,
_parse_validation, _parse_validation,
@@ -261,7 +262,7 @@ def compute_severity(
if hle_fallback and status == Status.MISSING: if hle_fallback and status == Status.MISSING:
return Severity.INFO return Severity.INFO
if mode == "existence": if not reads_file_contents(mode):
if status == Status.MISSING: if status == Status.MISSING:
return Severity.WARNING if required else Severity.INFO return Severity.WARNING if required else Severity.INFO
return Severity.OK return Severity.OK
@@ -735,8 +736,13 @@ def verify_platform(
target_cores: set[str] | None = None, target_cores: set[str] | None = None,
data_dir_registry: dict | None = None, data_dir_registry: dict | None = None,
supplemental_names: set[str] | None = None, supplemental_names: set[str] | None = None,
regions: list[str] | None = None,
) -> dict: ) -> dict:
"""Verify all BIOS files for a platform, including cross-reference gaps.""" """Verify all BIOS files for a platform, including cross-reference gaps.
A region priority list narrows the report to the files a pack built with the
same list would carry, using the same selection function as the builder.
"""
mode = config.get("verification_mode", "existence") mode = config.get("verification_mode", "existence")
platform = config.get("platform", "unknown") platform = config.get("platform", "unknown")
@@ -776,8 +782,28 @@ def verify_platform(
file_required: dict[str, bool] = {} file_required: dict[str, bool] = {}
file_severity: dict[str, str] = {} file_severity: dict[str, str] = {}
region_drops: set[str] = set()
if regions:
import region as region_mod
region_groups: dict[str, list[tuple[str, str]]] = {}
for sys_id, system in verify_systems.items():
members = region_groups.setdefault(sys_id, [])
for fe in system.get("files", []):
dest = fe.get("destination", fe.get("name", ""))
if dest:
members.append((dest, fe.get("name", "")))
region_drops = region_mod.resolve_region_drops(
region_groups, region_mod.build_region_index(profiles), regions
)
for sys_id, system in verify_systems.items(): for sys_id, system in verify_systems.items():
for file_entry in system.get("files", []): for file_entry in system.get("files", []):
if region_drops and (
file_entry.get("destination", file_entry.get("name", ""))
in region_drops
):
continue
local_path, resolve_status = resolve_local_file( local_path, resolve_status = resolve_local_file(
file_entry, file_entry,
db, db,
@@ -1163,8 +1189,13 @@ def verify_emulator(
emulators_dir: str, emulators_dir: str,
db: dict, db: dict,
standalone: bool = False, standalone: bool = False,
regions: list[str] | None = None,
) -> dict: ) -> dict:
"""Verify files for specific emulator profiles.""" """Verify files for specific emulator profiles.
A region priority list narrows the report the same way a pack built with the
same list would be narrowed. One group per profile, as in generate_pack.
"""
load_emulator_profiles(emulators_dir) load_emulator_profiles(emulators_dir)
zip_contents = build_zip_contents_index(db) zip_contents = build_zip_contents_index(db)
@@ -1220,8 +1251,31 @@ def verify_emulator(
dest_to_name: dict[str, str] = {} dest_to_name: dict[str, str] = {}
data_dir_notices: list[str] = [] data_dir_notices: list[str] = []
region_drops: set[str] = set()
if regions:
import region as region_mod
region_index = region_mod.build_region_index(dict(selected))
region_groups: dict[str, list[tuple[str, str]]] = {}
for emu_name, profile in selected:
members = region_groups.setdefault(emu_name, [])
for fe in filter_files_by_mode(profile.get("files", []), standalone):
nm = fe.get("name", "")
key = fe.get("path") or nm
if key:
members.append((key, nm))
region_drops = region_mod.resolve_region_drops(
region_groups, region_index, regions
)
for emu_name, profile in selected: for emu_name, profile in selected:
files = filter_files_by_mode(profile.get("files", []), standalone) files = filter_files_by_mode(profile.get("files", []), standalone)
if region_drops:
files = [
fe
for fe in files
if (fe.get("path") or fe.get("name", "")) not in region_drops
]
# Check data directories (only notice if not cached) # Check data directories (only notice if not cached)
for dd in profile.get("data_directories", []): for dd in profile.get("data_directories", []):
@@ -1415,6 +1469,7 @@ def verify_system(
emulators_dir: str, emulators_dir: str,
db: dict, db: dict,
standalone: bool = False, standalone: bool = False,
regions: list[str] | None = None,
) -> dict: ) -> dict:
"""Verify files for all emulators supporting given system IDs.""" """Verify files for all emulators supporting given system IDs."""
profiles = load_emulator_profiles(emulators_dir) profiles = load_emulator_profiles(emulators_dir)
@@ -1449,7 +1504,7 @@ def verify_system(
) )
sys.exit(1) sys.exit(1)
return verify_emulator(matching, emulators_dir, db, standalone) return verify_emulator(matching, emulators_dir, db, standalone, regions=regions)
def print_emulator_result(result: dict, verbose: bool = False) -> None: def print_emulator_result(result: dict, verbose: bool = False) -> None:
@@ -1554,6 +1609,9 @@ def main():
) )
parser.add_argument("--include-archived", action="store_true") parser.add_argument("--include-archived", action="store_true")
parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)") parser.add_argument("--target", "-t", help="Hardware target (e.g., switch, rpi4)")
parser.add_argument(
"--region", help="Region priority list, best first (e.g. us,eu,jp)"
)
parser.add_argument( parser.add_argument(
"--list-targets", "--list-targets",
action="store_true", action="store_true",
@@ -1571,6 +1629,15 @@ def main():
parser.add_argument("--json", action="store_true", help="JSON output") parser.add_argument("--json", action="store_true", help="JSON output")
args = parser.parse_args() args = parser.parse_args()
requested_regions: list[str] = []
if getattr(args, "region", None):
import region as region_mod
try:
requested_regions = region_mod.parse_requested(args.region)
except ValueError as exc:
parser.error(str(exc))
if args.list_emulators: if args.list_emulators:
list_emulator_profiles(args.emulators_dir) list_emulator_profiles(args.emulators_dir)
return return
@@ -1615,7 +1682,10 @@ def main():
# Emulator mode # Emulator mode
if args.emulator: if args.emulator:
names = [n.strip() for n in args.emulator.split(",") if n.strip()] names = [n.strip() for n in args.emulator.split(",") if n.strip()]
result = verify_emulator(names, args.emulators_dir, db, args.standalone) result = verify_emulator(
names, args.emulators_dir, db, args.standalone,
regions=requested_regions,
)
if args.json: if args.json:
result["details"] = [ result["details"] = [
d for d in result["details"] if d["status"] != Status.OK d for d in result["details"] if d["status"] != Status.OK
@@ -1628,7 +1698,10 @@ def main():
# System mode # System mode
if args.system: if args.system:
system_ids = [s.strip() for s in args.system.split(",") if s.strip()] system_ids = [s.strip() for s in args.system.split(",") if s.strip()]
result = verify_system(system_ids, args.emulators_dir, db, args.standalone) result = verify_system(
system_ids, args.emulators_dir, db, args.standalone,
regions=requested_regions,
)
if args.json: if args.json:
result["details"] = [ result["details"] = [
d for d in result["details"] if d["status"] != Status.OK d for d in result["details"] if d["status"] != Status.OK
@@ -1687,6 +1760,7 @@ def main():
target_cores=tc, target_cores=tc,
data_dir_registry=data_registry, data_dir_registry=data_registry,
supplemental_names=suppl_names, supplemental_names=suppl_names,
regions=requested_regions,
) )
names = [ names = [
load_platform_config(p, args.platforms_dir).get("platform", p) load_platform_config(p, args.platforms_dir).get("platform", p)
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""The native-mode policy verify.py and generate_pack.py must share.
CLAUDE.md states the invariant as "verify.py and generate_pack.py must
produce identical results; any divergence is a critical bug". It used to rest
on both files spelling out `mode == "existence"` in their own words. Routing
both through scripts/nativemode.py makes the rule single-sourced, and these
tests make the agreement checkable instead of reviewable.
"""
from __future__ import annotations
import json
import sys
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import nativemode # noqa: E402
class ModeVocabulary(unittest.TestCase):
def test_modes_match_the_platform_schema(self):
"""A mode added to one side and not the other is the bug this catches."""
schema = json.loads(
(REPO_ROOT / "schemas" / "platform.schema.json").read_text()
)
enum = schema["properties"]["verification_mode"]["enum"]
self.assertEqual(sorted(nativemode.MODES), sorted(enum))
def test_schema_default_matches_the_module_default(self):
schema = json.loads(
(REPO_ROOT / "schemas" / "platform.schema.json").read_text()
)
self.assertEqual(
schema["properties"]["verification_mode"]["default"],
nativemode.DEFAULT_MODE,
)
def test_unknown_or_missing_mode_falls_back_to_the_default(self):
for value in (None, "", "sha256", "MD5"):
self.assertEqual(nativemode.normalize(value), nativemode.DEFAULT_MODE)
def test_every_declared_platform_mode_is_known(self):
import yaml
for path in sorted((REPO_ROOT / "platforms").glob("*.yml")):
if path.name.startswith("_"):
continue
data = yaml.safe_load(path.read_text()) or {}
mode = data.get("verification_mode")
if mode is None:
continue
self.assertIn(mode, nativemode.MODES, f"{path.name} declares {mode!r}")
class ContentReadingPolicy(unittest.TestCase):
def test_existence_reads_nothing(self):
self.assertFalse(nativemode.reads_file_contents("existence"))
self.assertIsNone(nativemode.digest_algorithm("existence"))
def test_digest_modes_read_their_digest(self):
self.assertEqual(nativemode.digest_algorithm("md5"), "md5")
self.assertEqual(nativemode.digest_algorithm("sha1"), "sha1")
self.assertTrue(nativemode.reads_file_contents("md5"))
self.assertTrue(nativemode.reads_file_contents("sha1"))
def test_exclusion_follows_content_reading_for_every_mode(self):
"""The builder omits exactly what the frontend would reject."""
for mode in nativemode.MODES:
self.assertEqual(
nativemode.hash_mismatch_excludes_file(mode),
nativemode.reads_file_contents(mode),
f"mode {mode} disagrees with itself",
)
class BothConsumersAgree(unittest.TestCase):
"""verify.py and generate_pack.py must answer the same for every mode."""
def test_builder_and_verifier_share_one_predicate(self):
from generate_pack import _intentional_hash_exclusion
from verify import compute_severity # noqa: F401 (import proves wiring)
for mode in nativemode.MODES:
# No entries is never an exclusion, whatever the mode.
self.assertFalse(_intentional_hash_exclusion([], {}, verification_mode=mode))
# The builder must refuse to exclude under existence even when asked.
self.assertFalse(
_intentional_hash_exclusion(
[{"name": "x.bin"}], {}, verification_mode="existence"
)
)
def test_severity_uses_the_shared_predicate(self):
from verify import Severity, Status, compute_severity
# Existence mode: a missing required file is a warning, not critical.
self.assertEqual(
compute_severity(Status.MISSING, True, "existence"),
Severity.WARNING,
)
# Digest mode: the same absence is critical.
self.assertEqual(
compute_severity(Status.MISSING, True, "md5"),
Severity.CRITICAL,
)
# An unknown mode must fall back to existence, not to the strictest
# reading: a typo in a platform YAML should not invent CRITICALs.
self.assertEqual(
compute_severity(Status.MISSING, True, "sha256"),
Severity.WARNING,
)
if __name__ == "__main__":
unittest.main()
+120
View File
@@ -0,0 +1,120 @@
#!/usr/bin/env python3
"""The BaseScraper contract the wiki asks contributors to implement.
wiki/adding-a-scraper.md and wiki/adding-a-platform.md document
compare_with_config, ChangeSet.has_changes and test_connection as the
interface a new scraper plugs into. No scraper in the repository calls them,
so nothing proved they still worked; a documented contract that is never
exercised is a promise to contributors backed by nothing.
"""
from __future__ import annotations
import sys
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "scripts"))
from scraper.base_scraper import ( # noqa: E402
BaseScraper,
BiosRequirement,
ChangeSet,
)
class _StubScraper(BaseScraper):
"""Minimal scraper: returns what it was handed, or raises."""
def __init__(self, requirements=None, error: Exception | None = None):
super().__init__(url="https://example.test/bios.xml")
self._requirements = requirements or []
self._error = error
def fetch_requirements(self) -> list[BiosRequirement]:
if self._error is not None:
raise self._error
return self._requirements
def validate_format(self, raw_data: str) -> bool:
return True
def _config(*files: dict) -> dict:
return {"systems": {"test-sys": {"files": list(files)}}}
class ChangeSetSummary(unittest.TestCase):
def test_empty_changeset_has_no_changes(self):
changes = ChangeSet()
self.assertFalse(changes.has_changes)
self.assertEqual(changes.summary(), "no changes")
def test_summary_counts_each_kind(self):
req = BiosRequirement(name="a.bin", system="test-sys")
changes = ChangeSet(added=[req], removed=[req], modified=[(req, req)])
self.assertTrue(changes.has_changes)
self.assertEqual(changes.summary(), "+1 added, -1 removed, ~1 modified")
class CompareWithConfig(unittest.TestCase):
def test_new_upstream_file_is_reported_as_added(self):
scraper = _StubScraper([BiosRequirement(name="new.bin", system="test-sys")])
changes = scraper.compare_with_config(_config())
self.assertEqual([r.name for r in changes.added], ["new.bin"])
self.assertTrue(changes.has_changes)
def test_file_dropped_upstream_is_reported_as_removed(self):
scraper = _StubScraper([])
changes = scraper.compare_with_config(_config({"name": "old.bin"}))
self.assertEqual([r.name for r in changes.removed], ["old.bin"])
def test_changed_sha1_is_reported_as_modified(self):
scraper = _StubScraper(
[BiosRequirement(name="a.bin", system="test-sys", sha1="b" * 40)]
)
changes = scraper.compare_with_config(_config({"name": "a.bin", "sha1": "a" * 40}))
self.assertEqual(len(changes.modified), 1)
before, after = changes.modified[0]
self.assertEqual((before.sha1, after.sha1), ("a" * 40, "b" * 40))
def test_changed_md5_is_reported_when_no_sha1_is_declared(self):
scraper = _StubScraper(
[BiosRequirement(name="a.bin", system="test-sys", md5="b" * 32)]
)
changes = scraper.compare_with_config(_config({"name": "a.bin", "md5": "a" * 32}))
self.assertEqual(len(changes.modified), 1)
def test_identical_hashes_are_not_a_change(self):
scraper = _StubScraper(
[BiosRequirement(name="a.bin", system="test-sys", sha1="a" * 40)]
)
changes = scraper.compare_with_config(_config({"name": "a.bin", "sha1": "a" * 40}))
self.assertFalse(changes.has_changes)
def test_a_file_in_another_system_is_not_the_same_file(self):
scraper = _StubScraper([BiosRequirement(name="a.bin", system="other-sys")])
changes = scraper.compare_with_config(_config({"name": "a.bin"}))
self.assertEqual([r.name for r in changes.added], ["a.bin"])
self.assertEqual([r.name for r in changes.removed], ["a.bin"])
class TestConnection(unittest.TestCase):
def test_reachable_source_reports_true(self):
self.assertTrue(_StubScraper([]).test_connection())
def test_network_failure_reports_false(self):
self.assertFalse(_StubScraper(error=OSError("unreachable")).test_connection())
def test_bad_payload_reports_false(self):
self.assertFalse(_StubScraper(error=ValueError("bad format")).test_connection())
def test_an_unexpected_error_is_not_swallowed(self):
"""Only reachability failures are absorbed; a bug must surface."""
with self.assertRaises(KeyError):
_StubScraper(error=KeyError("bug")).test_connection()
if __name__ == "__main__":
unittest.main()