From 77929f8c827a8005b7bc402a00f712a5765c040c Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Sat, 5 Sep 2026 08:41:30 +0200 Subject: [PATCH] feat: report slots where a profile contradicts the pack --- scripts/slots.py | 270 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_slots.py | 221 ++++++++++++++++++++++++++++++++++++ 2 files changed, 491 insertions(+) create mode 100644 scripts/slots.py create mode 100644 tests/test_slots.py diff --git a/scripts/slots.py b/scripts/slots.py new file mode 100644 index 00000000..4b134126 --- /dev/null +++ b/scripts/slots.py @@ -0,0 +1,270 @@ +"""Who decides what goes at a destination. + +A destination is a slot. Two layers can claim it: the platform YAML, scraped +from what the frontend declares and therefore able to carry an upstream error, +and the emulator profile, read from the emulator's own source. Until this +module existed nothing compared the two: the builder resolves the platform +entry first and drops the profile entry on a bare filename match, so a profile +that names the right file for a slot could never win and never even be heard. + +The index is keyed by destination AND by name, the name carrying the union of +what every same-named entry claims. Dolphin declares three IPL.bin separated +only by their path, so a name-only key merges them; a path-only key answers +nothing when a candidate is known by name alone. Ambiguity resolves to the +union and is never discarded, the same rule region.py already applies. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from common import ( + resolution_is_hash_exact, + resolve_local_file, +) + +# A profile entry can prove a slot without declaring a hash: Dolphin names no +# checksum for the GameCube boot ROM because its source names none, and the +# proof is then the path the repository stores the file at. +PROVEN_STATUSES = frozenset({"path_exact"}) + + +@dataclass +class Claim: + """One layer's answer to what belongs at a destination.""" + + origin: str + destination: str + name: str + emulator: str = "" + entry: dict = field(default_factory=dict) + local_path: str | None = None + status: str = "" + + @property + def is_proven(self) -> bool: + """Whether the claim rests on evidence rather than on a bare name.""" + return resolution_is_hash_exact(self.status) or self.status in PROVEN_STATUSES + + +@dataclass +class Conflict: + """Two proven claims on one destination that resolve to different files.""" + + destination: str + platform_claim: Claim + profile_claims: list[Claim] + + @property + def emulators(self) -> list[str]: + return sorted({c.emulator for c in self.profile_claims if c.emulator}) + + +def _normalize(destination: str) -> str: + """Comparable form of a destination, since layers write it differently.""" + return destination.strip().strip("/").replace("\\", "/").casefold() + + +def build_claim_index(claims: list[Claim]) -> dict[str, list[Claim]]: + """Index claims by destination and by name, the name carrying the union. + + A lookup by full destination answers for one slot. A lookup by bare name + answers with every slot that shares it, which is what a candidate known + only by its filename needs in order not to be dropped. + """ + index: dict[str, list[Claim]] = {} + for claim in claims: + keys = {_normalize(claim.destination)} + if claim.name: + keys.add(_normalize(claim.name)) + base = claim.destination.rsplit("/", 1)[-1] + if base: + keys.add(_normalize(base)) + for key in keys - {""}: + index.setdefault(key, []).append(claim) + return index + + +def platform_claims(config: dict, db: dict, base_dest: str = "") -> list[Claim]: + """What the platform YAML says belongs at each of its destinations.""" + claims: list[Claim] = [] + for system in (config.get("systems") or {}).values(): + for entry in system.get("files") or []: + if not isinstance(entry, dict): + continue + dest = entry.get("destination") or entry.get("name") or "" + if not dest: + continue + full = f"{base_dest}/{dest}" if base_dest else dest + local, status = resolve_local_file(entry, db, dest_hint=dest) + claims.append( + Claim( + origin="platform", + destination=full, + name=entry.get("name", ""), + entry=entry, + local_path=local, + status=status, + ) + ) + return claims + + +def profile_claims( + profiles: dict, db: dict, base_dest: str = "" +) -> list[Claim]: + """What each emulator profile says belongs at each destination it names.""" + claims: list[Claim] = [] + for emu_name, profile in sorted(profiles.items()): + if profile.get("type") in ("launcher", "alias"): + continue + for entry in profile.get("files") or []: + if not isinstance(entry, dict): + continue + dest = entry.get("path") or entry.get("name") or "" + if not dest: + continue + full = f"{base_dest}/{dest}" if base_dest else dest + local, status = resolve_local_file(entry, db, dest_hint=dest) + claims.append( + Claim( + origin="profile", + destination=full, + name=entry.get("name", ""), + emulator=emu_name, + entry=entry, + local_path=local, + status=status, + ) + ) + return claims + + +def find_conflicts( + config: dict, profiles: dict, db: dict, base_dest: str = "" +) -> list[Conflict]: + """Destinations where a proven profile claim contradicts what ships. + + Only proven claims are compared. A claim resolved by filename alone + asserts nothing about content and cannot contradict anything. + """ + by_dest: dict[str, Claim] = {} + for claim in platform_claims(config, db, base_dest): + by_dest.setdefault(_normalize(claim.destination), claim) + + # Grouped before judging: a profile may declare several revisions that are + # all acceptable at one destination, and the platform choosing one of them + # is agreement, not contradiction. Only a destination where no profile + # claim at all matches what ships is a disagreement. + by_slot: dict[str, list[Claim]] = {} + for claim in profile_claims(profiles, db, base_dest): + key = _normalize(claim.destination) + platform = by_dest.get(key) + if platform is None or not platform.is_proven or not claim.is_proven: + continue + by_slot.setdefault(key, []).append(claim) + + disputed = { + key: claims + for key, claims in by_slot.items() + if all(c.local_path != by_dest[key].local_path for c in claims) + } + + return [ + Conflict( + destination=by_dest[key].destination, + platform_claim=by_dest[key], + profile_claims=claims, + ) + for key, claims in sorted(disputed.items()) + ] + + +def format_conflict(conflict: Conflict) -> str: + """One line per conflict, naming both answers and who gave them.""" + emus = ", ".join(conflict.emulators) or "profile" + return ( + f"{conflict.destination}: pack ships {conflict.platform_claim.local_path} " + f"({conflict.platform_claim.status}), {emus} says " + f"{conflict.profile_claims[0].local_path} " + f"({conflict.profile_claims[0].status})" + ) + + +def scan_platform( + platform: str, profiles: dict, db: dict, platforms_dir: str = "platforms" +) -> list[Conflict]: + """Conflicts on one platform, using the cores that platform actually runs.""" + from common import load_platform_config, resolve_platform_cores + + config = load_platform_config(platform, platforms_dir) + keys = resolve_platform_cores(config, profiles) + relevant = {k: profiles[k] for k in keys if k in profiles} + return find_conflicts(config, relevant, db, config.get("base_destination", "")) + + +def main() -> int: + import argparse + import json + + from common import list_registered_platforms, load_emulator_profiles + + parser = argparse.ArgumentParser( + description="Report destinations where an emulator profile contradicts " + "the file a platform baseline would ship.", + ) + parser.add_argument("--platform", help="one platform instead of all") + parser.add_argument("--db", default="database.json") + parser.add_argument("--platforms-dir", default="platforms") + parser.add_argument("--emulators-dir", default="emulators") + parser.add_argument("--json", action="store_true", help="JSON output") + args = parser.parse_args() + + with open(args.db, encoding="utf-8") as handle: + db = json.load(handle) + profiles = load_emulator_profiles(args.emulators_dir) + names = ( + [args.platform] + if args.platform + else list_registered_platforms(args.platforms_dir) + ) + + found: dict[str, list[Conflict]] = {} + for name in names: + conflicts = scan_platform(name, profiles, db, args.platforms_dir) + if conflicts: + found[name] = conflicts + + if args.json: + print( + json.dumps( + { + platform: [ + { + "destination": c.destination, + "ships": c.platform_claim.local_path, + "ships_evidence": c.platform_claim.status, + "emulators": c.emulators, + "expected": c.profile_claims[0].local_path, + "expected_evidence": c.profile_claims[0].status, + } + for c in conflicts + ] + for platform, conflicts in found.items() + }, + indent=2, + ) + ) + else: + for platform, conflicts in found.items(): + print(f"{platform}: {len(conflicts)} contradicted") + for conflict in conflicts: + print(f" {format_conflict(conflict)}") + total = sum(len(c) for c in found.values()) + print(f"\n{total} destinations where a profile contradicts what ships.") + + return 1 if found else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_slots.py b/tests/test_slots.py new file mode 100644 index 00000000..ec7ea9fe --- /dev/null +++ b/tests/test_slots.py @@ -0,0 +1,221 @@ +"""Slot arbitration: who decides what goes at a destination.""" + +from __future__ import annotations + +import os +import shutil +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +import slots # noqa: E402 + + +def _db(entries: dict[str, dict]) -> dict: + """Minimal database shaped like the generated one, keyed by sha1. + + The resolver reads the disk before it answers, so the fixture writes the + files it indexes and the caller runs from the directory holding them. + """ + files, by_md5, by_name, by_path_suffix = {}, {}, {}, {} + for sha1, info in entries.items(): + path = info["path"] + name = path.rsplit("/", 1)[-1] + files[sha1] = { + "path": path, + "md5": info.get("md5", ""), + "name": name, + "crc32": info.get("crc32", ""), + "size": info.get("size", 0), + "sha1": sha1, + } + if info.get("md5"): + by_md5[info["md5"]] = sha1 + by_name.setdefault(name, []).append(sha1) + parts = path.split("/") + for depth in range(1, len(parts)): + by_path_suffix.setdefault("/".join(parts[-depth:]), []).append(sha1) + return { + "files": files, + "indexes": { + "by_md5": by_md5, + "by_name": by_name, + "by_crc32": {}, + "by_path_suffix": by_path_suffix, + }, + } + + +_ROOT = Path(tempfile.mkdtemp(prefix="slots-")) +_ENTRIES = { + "a" * 40: {"path": "bios/Console/GC/USA/IPL.bin", "md5": "m" * 32, "size": 3}, + "b" * 40: {"path": "bios/Console/GC/JAP/IPL.bin", "md5": "n" * 32, "size": 3}, +} +for _info in _ENTRIES.values(): + _target = _ROOT / _info["path"] + _target.parent.mkdir(parents=True, exist_ok=True) + _target.write_bytes(b"rom") + +REGIONS_DB = _db(_ENTRIES) + + +def setUpModule(): + global _PREVIOUS_CWD + _PREVIOUS_CWD = os.getcwd() + os.chdir(_ROOT) + + +def tearDownModule(): + os.chdir(_PREVIOUS_CWD) + shutil.rmtree(_ROOT, ignore_errors=True) + + +class TestClaimIndex(unittest.TestCase): + """Keyed by destination and by name, the name carrying the union.""" + + def _claims(self): + return [ + slots.Claim("profile", "GC/USA/IPL.bin", "IPL.bin", emulator="dolphin"), + slots.Claim("profile", "GC/JAP/IPL.bin", "IPL.bin", emulator="dolphin"), + ] + + def test_destination_key_separates_same_named_entries(self): + index = slots.build_claim_index(self._claims()) + self.assertEqual(len(index["gc/usa/ipl.bin"]), 1) + self.assertEqual(len(index["gc/jap/ipl.bin"]), 1) + + def test_bare_name_answers_with_the_union(self): + index = slots.build_claim_index(self._claims()) + self.assertEqual(len(index["ipl.bin"]), 2) + + def test_lookup_ignores_case_and_leading_separator(self): + index = slots.build_claim_index( + [slots.Claim("platform", "/BIOS/Disk.rom", "Disk.rom")] + ) + self.assertIn("bios/disk.rom", index) + + +class TestConflicts(unittest.TestCase): + """A proven profile claim contradicting what the baseline would ship.""" + + def _config(self, md5: str) -> dict: + return { + "systems": { + "console": { + "files": [ + { + "name": "IPL.bin", + "destination": "GC/JAP/IPL.bin", + "md5": md5, + } + ] + } + } + } + + def _profile(self, *paths: str) -> dict: + return { + "dolphin": { + "files": [{"name": "IPL.bin", "path": p} for p in paths] + } + } + + def test_wrong_region_under_a_region_slot_is_reported(self): + conflicts = slots.find_conflicts( + self._config("m" * 32), self._profile("GC/JAP/IPL.bin"), REGIONS_DB + ) + self.assertEqual(len(conflicts), 1) + self.assertEqual(conflicts[0].destination, "GC/JAP/IPL.bin") + self.assertTrue(conflicts[0].platform_claim.local_path.endswith("USA/IPL.bin")) + self.assertTrue( + conflicts[0].profile_claims[0].local_path.endswith("JAP/IPL.bin") + ) + self.assertEqual(conflicts[0].emulators, ["dolphin"]) + + def test_agreement_is_not_a_conflict(self): + conflicts = slots.find_conflicts( + self._config("n" * 32), self._profile("GC/JAP/IPL.bin"), REGIONS_DB + ) + self.assertEqual(conflicts, []) + + def test_one_matching_revision_among_several_settles_the_slot(self): + # A profile listing every acceptable revision agrees with the platform + # as soon as one of them is what ships. + profile = { + "dosbox": { + "files": [ + {"name": "IPL.bin", "path": "GC/JAP/IPL.bin", "md5": "n" * 32}, + {"name": "IPL.bin", "path": "GC/JAP/IPL.bin", "md5": "m" * 32}, + ] + } + } + conflicts = slots.find_conflicts(self._config("m" * 32), profile, REGIONS_DB) + self.assertEqual(conflicts, []) + + def test_a_name_only_resolution_asserts_nothing(self): + # No hash and no path that the repository carries: the entry is + # answered by filename alone and cannot contradict anything. + config = { + "systems": { + "console": { + "files": [{"name": "IPL.bin", "destination": "slot/IPL.bin"}] + } + } + } + profile = {"dolphin": {"files": [{"name": "IPL.bin", "path": "slot/IPL.bin"}]}} + self.assertEqual(slots.find_conflicts(config, profile, REGIONS_DB), []) + + def test_launchers_and_aliases_never_claim(self): + profile = { + "launcher": { + "type": "launcher", + "files": [{"name": "IPL.bin", "path": "GC/JAP/IPL.bin"}], + } + } + self.assertEqual( + slots.find_conflicts(self._config("m" * 32), profile, REGIONS_DB), [] + ) + + def test_base_destination_prefixes_both_sides(self): + conflicts = slots.find_conflicts( + self._config("m" * 32), + self._profile("GC/JAP/IPL.bin"), + REGIONS_DB, + base_dest="bios", + ) + self.assertEqual(conflicts[0].destination, "bios/GC/JAP/IPL.bin") + + def test_report_line_names_both_answers(self): + conflicts = slots.find_conflicts( + self._config("m" * 32), self._profile("GC/JAP/IPL.bin"), REGIONS_DB + ) + line = slots.format_conflict(conflicts[0]) + self.assertIn("USA/IPL.bin", line) + self.assertIn("JAP/IPL.bin", line) + self.assertIn("dolphin", line) + + +class TestProvenEvidence(unittest.TestCase): + """What counts as proof that a claim is about content, not about a name.""" + + def test_hash_evidence_is_proof(self): + self.assertTrue(slots.Claim("profile", "d", "n", status="sha1_exact").is_proven) + self.assertTrue(slots.Claim("profile", "d", "n", status="md5_exact").is_proven) + + def test_path_evidence_is_proof(self): + # Dolphin declares no checksum for the GameCube boot ROM because its + # source declares none; the path is then the whole assertion. + self.assertTrue(slots.Claim("profile", "d", "n", status="path_exact").is_proven) + + def test_a_name_is_not_proof(self): + self.assertFalse(slots.Claim("profile", "d", "n", status="name_exact").is_proven) + self.assertFalse( + slots.Claim("profile", "d", "n", status="hash_mismatch").is_proven + ) + + +if __name__ == "__main__": + unittest.main()