mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-10 21:43:23 -05:00
342 lines
13 KiB
Python
342 lines
13 KiB
Python
"""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 TestArbitration(unittest.TestCase):
|
|
"""A pack answers to whoever asked for it, and serves both when it can."""
|
|
|
|
def _conflict(self):
|
|
platform = slots.Claim(
|
|
"platform", "GC/JAP/IPL.bin", "IPL.bin",
|
|
local_path="bios/Console/GC/USA/IPL.bin", status="md5_exact",
|
|
)
|
|
profile = slots.Claim(
|
|
"profile", "GC/JAP/IPL.bin", "IPL.bin", emulator="dolphin",
|
|
local_path="bios/Console/GC/JAP/IPL.bin", status="path_exact",
|
|
)
|
|
return slots.Conflict("GC/JAP/IPL.bin", platform, [profile])
|
|
|
|
def test_existence_mode_serves_both(self):
|
|
decision = slots.arbitrate(self._conflict(), "existence")
|
|
self.assertEqual(decision.winner.origin, "profile")
|
|
self.assertTrue(decision.serves_both)
|
|
|
|
def test_md5_mode_keeps_the_frontend_green(self):
|
|
decision = slots.arbitrate(self._conflict(), "md5")
|
|
self.assertEqual(decision.winner.origin, "platform")
|
|
self.assertFalse(decision.serves_both)
|
|
|
|
def test_sha1_mode_is_content_checking_too(self):
|
|
decision = slots.arbitrate(self._conflict(), "sha1")
|
|
self.assertEqual(decision.winner.origin, "platform")
|
|
|
|
def test_an_emulator_pack_answers_to_the_emulator(self):
|
|
decision = slots.arbitrate(self._conflict(), "md5", addressee="emulator")
|
|
self.assertEqual(decision.winner.origin, "profile")
|
|
self.assertFalse(decision.serves_both)
|
|
|
|
def test_the_reported_line_gives_the_ground_for_the_decision(self):
|
|
kept = slots.format_decision(slots.arbitrate(self._conflict(), "md5"))
|
|
self.assertIn("verifies content", kept)
|
|
served = slots.format_decision(slots.arbitrate(self._conflict(), "existence"))
|
|
self.assertIn("both are satisfied", served)
|
|
|
|
|
|
class TestBuilderAndVerifierAgree(unittest.TestCase):
|
|
"""The two tools must settle a contested destination the same way."""
|
|
|
|
def test_both_read_the_same_decision_function(self):
|
|
# The rule lives in one place. A second copy would let the pack and the
|
|
# report describe different files, which the project forbids outright.
|
|
builder = Path(__file__).resolve().parents[1] / "scripts" / "generate_pack.py"
|
|
verifier = Path(__file__).resolve().parents[1] / "scripts" / "verify.py"
|
|
for source in (builder, verifier):
|
|
text = source.read_text(encoding="utf-8")
|
|
self.assertIn("slots.find_conflicts(", text, source.name)
|
|
self.assertIn("slots.arbitrate(", text, source.name)
|
|
self.assertIn("decision.serves_both", text, source.name)
|
|
|
|
def test_neither_reimplements_the_mode_test(self):
|
|
# A local "if mode == md5" beside the override would drift from the
|
|
# arbitration rule the moment either side is edited.
|
|
for name in ("generate_pack.py", "verify.py"):
|
|
source = Path(__file__).resolve().parents[1] / "scripts" / name
|
|
body = source.read_text(encoding="utf-8")
|
|
marker = body.find("decision.serves_both")
|
|
window = body[max(0, marker - 600):marker]
|
|
self.assertNotIn('verification_mode") == "md5"', window, name)
|
|
|
|
|
|
class TestSelfContradictingDestinations(unittest.TestCase):
|
|
"""One path holds one file, whatever the upstream list says."""
|
|
|
|
def _config(self, *entries: dict) -> dict:
|
|
return {"systems": {"console": {"files": list(entries)}}}
|
|
|
|
def test_two_declarations_landing_on_two_files_are_reported(self):
|
|
config = self._config(
|
|
{"name": "IPL.bin", "destination": "disk.rom", "md5": "m" * 32},
|
|
{"name": "IPL.bin", "destination": "disk.rom", "md5": "n" * 32},
|
|
)
|
|
collisions = slots.find_collisions(config, REGIONS_DB)
|
|
self.assertEqual(len(collisions), 1)
|
|
self.assertEqual(len(collisions[0].resolved), 2)
|
|
|
|
def test_one_archive_named_by_its_inner_roms_is_not_a_collision(self):
|
|
# The documented zipped_file pattern: several md5s, one archive.
|
|
config = self._config(
|
|
{"name": "IPL.bin", "destination": "a.zip", "md5": "m" * 32,
|
|
"zipped_file": "one.bin"},
|
|
{"name": "IPL.bin", "destination": "a.zip", "md5": "m" * 32,
|
|
"zipped_file": "two.bin"},
|
|
)
|
|
self.assertEqual(slots.find_collisions(config, REGIONS_DB), [])
|
|
|
|
def test_a_primary_and_its_pinned_variant_are_one_family(self):
|
|
self.assertTrue(
|
|
slots._same_file_family(
|
|
["bios/M/C/rom.zip", "bios/M/C/.variants/rom.zip.abcd1234"]
|
|
)
|
|
)
|
|
|
|
def test_two_machines_are_not_one_family(self):
|
|
self.assertFalse(
|
|
slots._same_file_family(
|
|
["bios/Microsoft/MSX/DISK.ROM", "bios/Tandy/CoCo/disk.rom"]
|
|
)
|
|
)
|
|
|
|
def test_a_destination_declared_once_is_never_a_collision(self):
|
|
config = self._config(
|
|
{"name": "IPL.bin", "destination": "disk.rom", "md5": "m" * 32}
|
|
)
|
|
self.assertEqual(slots.find_collisions(config, REGIONS_DB), [])
|
|
|
|
def test_the_line_names_every_file_claiming_the_path(self):
|
|
config = self._config(
|
|
{"name": "IPL.bin", "destination": "disk.rom", "md5": "m" * 32},
|
|
{"name": "IPL.bin", "destination": "disk.rom", "md5": "n" * 32},
|
|
)
|
|
line = slots.format_collision(slots.find_collisions(config, REGIONS_DB)[0])
|
|
self.assertIn("USA/IPL.bin", line)
|
|
self.assertIn("JAP/IPL.bin", 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()
|