mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-23 20:29:06 -05:00
Abdessamad Derraz
5b083c45b5
The script exporters (Batocera, ROCKNIX, KNULLI, AmberELEC, EmuELEC) read the checker of the pinned original: when checkBios hands an item's md5 to a function that tests it with isinstance(..., list), as md5Matches does since batocera.linux 6b08b873, an entry whose emulator code accepts several dumps is written with all of them, and an addition that needed several becomes writable. Items with altmd5 keep a string, since checkBios wraps them in a list of its own. No cached original reads lists today, so every export is unchanged. The report records a list as a list, counts the lists written, and records a list the truth extends without contradicting it.
442 lines
19 KiB
Python
442 lines
19 KiB
Python
"""An md5 written as a list of strings in batocera-systems.
|
|
|
|
batocera.linux 6b08b873 lets an item's md5 be one string or a list, and
|
|
checkBios accepts a file whose md5 is any listed value (`md5Matches`, and
|
|
`checkInsideZip` for a zippedFile). The scrapers keep the values
|
|
comma-joined with the list form recorded, the export writes a list back as
|
|
a list, and verify accepts any listed md5.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
from common import resolve_local_file # noqa: E402
|
|
from exporter import discover_exporters # noqa: E402
|
|
from export_native import landed_changes # noqa: E402
|
|
from exporter.baseline import NativeFile, build_native_model # noqa: E402
|
|
from exporter.batocera_exporter import md5_list_support # noqa: E402
|
|
from scraper import rocknix_scraper # noqa: E402
|
|
from scraper.base_scraper import md5_field, requirement_entry # noqa: E402
|
|
from scraper.batocera_scraper import ( # noqa: E402
|
|
SYSTEM_SLUG_MAP,
|
|
extract_systems_dict,
|
|
platform_systems,
|
|
requirements_from_systems,
|
|
)
|
|
from siteexports import describe_change # noqa: E402
|
|
from verify import Status, verify_entry_md5 # noqa: E402
|
|
|
|
A = "dc2b9bf8da62ec93e868cfd29f0d067d"
|
|
B = "924e392ed05558ffdb115408c263dccf"
|
|
C = "8dd7d5296a650fac7319bce665a6a53c"
|
|
|
|
# The psx entry as 6b08b873 writes it, one string item beside one list item.
|
|
SCRIPT = (
|
|
"#!/usr/bin/env python\n"
|
|
"systems = {\n"
|
|
' "psx": { "name": "PSX", "biosFiles": [ '
|
|
'{ "md5": "' + C + '", "file": "bios/scph5500.bin" },\n'
|
|
' '
|
|
'{ "md5": [ "' + A + '", "' + B + '" ], "file": "bios/scph1001.bin" } ] },\n'
|
|
"}\n"
|
|
"def checkBios(systems, prefix, filterROMs):\n"
|
|
" return {}\n"
|
|
)
|
|
|
|
|
|
def _scraped(script: str = SCRIPT) -> dict:
|
|
systems = extract_systems_dict(script)
|
|
requirements = requirements_from_systems(systems, SYSTEM_SLUG_MAP)
|
|
return {"cores": [], "systems": platform_systems(requirements, systems, SYSTEM_SLUG_MAP)}
|
|
|
|
|
|
def _files(scraped: dict) -> dict[str, dict]:
|
|
return {fe["name"]: fe for system in scraped["systems"].values() for fe in system["files"]}
|
|
|
|
|
|
class Md5Field(unittest.TestCase):
|
|
def test_string_and_list(self):
|
|
self.assertEqual(md5_field(A), ([A], False))
|
|
self.assertEqual(md5_field(""), ([], False))
|
|
self.assertEqual(md5_field(None), ([], False))
|
|
self.assertEqual(md5_field([A, B]), ([A, B], True))
|
|
self.assertEqual(md5_field([A]), ([A], True))
|
|
self.assertEqual(md5_field([]), ([], True))
|
|
|
|
def test_anything_else_is_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
md5_field(32)
|
|
with self.assertRaises(ValueError):
|
|
md5_field([A, 1])
|
|
|
|
|
|
class BatoceraScraper(unittest.TestCase):
|
|
def test_a_string_stays_one_value(self):
|
|
entry = _files(_scraped())["scph5500.bin"]
|
|
self.assertEqual(entry["md5"], C)
|
|
self.assertNotIn("md5_list", entry)
|
|
|
|
def test_a_list_is_joined_and_marked(self):
|
|
entry = _files(_scraped())["scph1001.bin"]
|
|
self.assertEqual(entry["md5"], f"{A},{B}")
|
|
self.assertIs(entry["md5_list"], True)
|
|
|
|
def test_a_one_item_list_keeps_its_form(self):
|
|
script = SCRIPT.replace(f'[ "{A}", "{B}" ]', f'[ "{A}" ]')
|
|
entry = _files(_scraped(script))["scph1001.bin"]
|
|
self.assertEqual(entry["md5"], A)
|
|
self.assertIs(entry["md5_list"], True)
|
|
|
|
def test_an_empty_list_checks_by_existence(self):
|
|
script = SCRIPT.replace(f'[ "{A}", "{B}" ]', "[]")
|
|
entry = _files(_scraped(script))["scph1001.bin"]
|
|
self.assertNotIn("md5", entry)
|
|
self.assertIs(entry["md5_list"], True)
|
|
|
|
def test_the_schema_accepts_the_entry(self):
|
|
import json
|
|
|
|
try:
|
|
import jsonschema
|
|
except ImportError:
|
|
self.skipTest("jsonschema is a tooling dependency")
|
|
|
|
schema = json.loads((REPO_ROOT / "schemas" / "platform.schema.json").read_text())
|
|
entry = _files(_scraped())["scph1001.bin"]
|
|
jsonschema.validate(entry, schema["$defs"]["bios_file"])
|
|
|
|
|
|
class RocknixScraper(unittest.TestCase):
|
|
def _requirements(self, script: str):
|
|
scraper = rocknix_scraper.Scraper(url="https://example.invalid/rocknix-systems")
|
|
scraper._fetch_raw = lambda: script # type: ignore[method-assign]
|
|
return {req.name: requirement_entry(req) for req in scraper.fetch_requirements()}
|
|
|
|
def test_string_and_list(self):
|
|
entries = self._requirements(SCRIPT)
|
|
self.assertEqual(entries["scph5500.bin"]["md5"], C)
|
|
self.assertNotIn("md5_list", entries["scph5500.bin"])
|
|
self.assertEqual(entries["scph1001.bin"]["md5"], f"{A},{B}")
|
|
self.assertIs(entries["scph1001.bin"]["md5_list"], True)
|
|
|
|
|
|
class BatoceraExportRoundTrip(unittest.TestCase):
|
|
def _render(self, truth_files: list[dict], script: str = SCRIPT) -> str:
|
|
scraped = _scraped(script)
|
|
systems, report = build_native_model(
|
|
{"systems": {"sony-playstation": {"files": truth_files}}}, scraped
|
|
)
|
|
exporter = discover_exporters()["batocera"]()
|
|
return exporter.render(systems, report, {"batocera-systems": script}, scraped)[
|
|
"batocera-systems"]
|
|
|
|
def test_nothing_to_correct_leaves_the_script_unchanged(self):
|
|
self.assertEqual(self._render([]), SCRIPT)
|
|
|
|
def test_a_one_item_list_is_written_back_as_a_list(self):
|
|
script = SCRIPT.replace(f'[ "{A}", "{B}" ]', f'[ "{A}" ]')
|
|
self.assertEqual(self._render([], script), script)
|
|
|
|
def test_a_corrected_list_stays_a_list(self):
|
|
produced = self._render([{
|
|
"name": "scph1001.bin", "path": "scph1001.bin", "md5": [A, B, "f" * 32],
|
|
"_cores": ["pcsx_rearmed"],
|
|
}])
|
|
self.assertEqual(
|
|
produced,
|
|
SCRIPT.replace(f'[ "{A}", "{B}" ]', f'[ "{A}", "{B}", "{"f" * 32}" ]'),
|
|
)
|
|
|
|
def test_a_string_item_stays_a_string(self):
|
|
produced = self._render([{
|
|
"name": "scph5500.bin", "path": "scph5500.bin", "md5": "e" * 32,
|
|
"_cores": ["pcsx_rearmed"],
|
|
}])
|
|
self.assertEqual(produced, SCRIPT.replace(C, "e" * 32))
|
|
|
|
|
|
class VerifyAcceptsAnyListedMd5(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
scratch = REPO_ROOT / "tmp"
|
|
self.tmp = tempfile.TemporaryDirectory(dir=scratch if scratch.is_dir() else None)
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.root = Path(self.tmp.name)
|
|
|
|
def _file(self, payload: bytes) -> str:
|
|
path = self.root / "scph1001.bin"
|
|
path.write_bytes(payload)
|
|
return str(path)
|
|
|
|
def test_each_listed_md5_matches(self):
|
|
first, second, other = b"first", b"second", b"other"
|
|
entry = _files(_scraped(SCRIPT.replace(A, hashlib.md5(first).hexdigest())
|
|
.replace(B, hashlib.md5(second).hexdigest())))["scph1001.bin"]
|
|
for payload in (first, second):
|
|
self.assertEqual(verify_entry_md5(entry, self._file(payload))["status"], Status.OK)
|
|
self.assertEqual(verify_entry_md5(entry, self._file(other))["status"], Status.UNTESTED)
|
|
|
|
def test_the_resolver_finds_the_second_listed_dump(self):
|
|
"""The pack builder and the manifest resolve through this function."""
|
|
payload = b"second"
|
|
md5 = hashlib.md5(payload).hexdigest()
|
|
sha1 = hashlib.sha1(payload).hexdigest()
|
|
path = self._file(payload)
|
|
db = {
|
|
"files": {sha1: {"path": path, "md5": md5, "sha1": sha1, "name": "scph1001.bin"}},
|
|
"indexes": {"by_md5": {md5: sha1}, "by_name": {"scph1001.bin": [sha1]}},
|
|
}
|
|
entry = {"name": "scph1001.bin", "md5": f"{A},{md5}", "md5_list": True}
|
|
self.assertEqual(resolve_local_file(entry, db), (path, "md5_exact"))
|
|
|
|
def test_inside_a_zip(self):
|
|
member = b"member"
|
|
entry = {"name": "pack.zip", "zipped_file": "rom.bin",
|
|
"md5": f"{A},{hashlib.md5(member).hexdigest()}", "md5_list": True}
|
|
archive = self.root / "pack.zip"
|
|
with zipfile.ZipFile(archive, "w") as zf:
|
|
zf.writestr("rom.bin", member)
|
|
self.assertEqual(verify_entry_md5(entry, str(archive))["status"], Status.OK)
|
|
|
|
|
|
# checkBios as batocera.linux 6b08b873 writes it: a plain file's md5 goes
|
|
# through md5Matches, a member's through checkInsideZip, both reading lists.
|
|
LIST_CHECKER = """def md5Matches(computed, expected):
|
|
if isinstance(expected, list):
|
|
expectedList = expected
|
|
else:
|
|
expectedList = [expected]
|
|
return computed in expectedList
|
|
|
|
def checkInsideZip(container, fileName, md5sums):
|
|
if not isinstance(md5sums, list):
|
|
md5sums = [md5sums]
|
|
import zipfile
|
|
from hashlib import md5
|
|
with zipfile.ZipFile(container) as archive:
|
|
for fname in archive.namelist():
|
|
if fname.casefold() == fileName.casefold():
|
|
if md5sums == [""] or md5sums == []:
|
|
return True
|
|
return md5(archive.read(fname)).hexdigest() in md5sums
|
|
return False
|
|
|
|
def checkBios(systems, prefix, filterROMs):
|
|
from hashlib import md5
|
|
from os.path import isfile
|
|
missingBios = {}
|
|
for system in systems.keys():
|
|
for file in systems[system]["biosFiles"]:
|
|
filepath = prefix + "/" + file["file"]
|
|
if not isfile(filepath):
|
|
missingBios.setdefault(system, {})[file["file"]] = "MISSING"
|
|
elif not "zippedFile" in file:
|
|
with open(filepath, "rb") as handle:
|
|
digest = md5(handle.read()).hexdigest()
|
|
if not md5Matches(digest, file["md5"]) and file["md5"] != "" and file["md5"] != []:
|
|
missingBios.setdefault(system, {})[file["file"]] = "UNTESTED"
|
|
elif 'altmd5' in file:
|
|
if not checkInsideZip(filepath, file["zippedFile"], [file["md5"], file["altmd5"]]):
|
|
missingBios.setdefault(system, {})[file["file"]] = "UNTESTED"
|
|
elif not checkInsideZip(filepath, file["zippedFile"], file["md5"]):
|
|
missingBios.setdefault(system, {})[file["file"]] = "UNTESTED"
|
|
return missingBios
|
|
"""
|
|
|
|
# batocera-43.1's checker: one md5, compared with `!=`.
|
|
STRING_CHECKER = """def checkInsideZip(container, fileName, md5sum):
|
|
return md5sum == ""
|
|
|
|
def checkBios(systems, prefix, filterROMs):
|
|
missingBios = {}
|
|
for system in systems.keys():
|
|
for file in systems[system]["biosFiles"]:
|
|
if "zippedFile" in file:
|
|
checkInsideZip(prefix, file["zippedFile"], file["md5"])
|
|
elif "" != file["md5"]:
|
|
missingBios[system] = file["file"]
|
|
return missingBios
|
|
"""
|
|
|
|
DATA = SCRIPT.split("def checkBios")[0]
|
|
LISTING = DATA + LIST_CHECKER
|
|
STRINGS = DATA + STRING_CHECKER
|
|
D = "e" * 32
|
|
F = "f" * 32
|
|
SECOND = hashlib.md5(b"second").hexdigest()
|
|
|
|
|
|
class ListSupportIsReadFromTheChecker(unittest.TestCase):
|
|
def test_the_list_checker_reads_lists_for_both_kinds(self):
|
|
self.assertEqual(md5_list_support(LISTING), {"file", "zippedFile"})
|
|
|
|
def test_the_string_checker_reads_none(self):
|
|
self.assertEqual(md5_list_support(STRINGS), frozenset())
|
|
self.assertEqual(md5_list_support(SCRIPT), frozenset())
|
|
|
|
def test_a_list_test_on_another_parameter_is_not_support(self):
|
|
checker = LIST_CHECKER.replace(
|
|
"isinstance(expected, list)", "isinstance(computed, list)"
|
|
)
|
|
self.assertEqual(md5_list_support(DATA + checker), {"zippedFile"})
|
|
|
|
def test_a_script_that_does_not_parse_reads_none(self):
|
|
self.assertEqual(md5_list_support(LISTING + "\ndef (\n"), frozenset())
|
|
|
|
def test_the_cached_originals(self):
|
|
"""The pins cached today (batocera-43.1 and the forks) read one md5."""
|
|
root = REPO_ROOT / ".cache" / "upstream-native"
|
|
scripts = [
|
|
root / name / filename
|
|
for name, filename in (
|
|
("batocera", "batocera-systems"), ("rocknix", "rocknix-systems"),
|
|
("amberelec", "batocera-systems"), ("emuelec", "batocera-systems"),
|
|
("knulli", "knulli-systems"),
|
|
)
|
|
if (root / name / filename).is_file()
|
|
]
|
|
if not scripts:
|
|
self.skipTest(".cache/upstream-native/ is not filled")
|
|
for path in scripts:
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
expected = frozenset() if "md5Matches" not in text else {"file"}
|
|
with self.subTest(path.parent.name):
|
|
self.assertLessEqual(expected, md5_list_support(text))
|
|
|
|
|
|
class SeveralAcceptedMd5BecomeAList(unittest.TestCase):
|
|
"""Every md5 the truth accepts, once the pinned checker reads a list."""
|
|
|
|
CORRECTED = {
|
|
"name": "scph5500.bin", "path": "scph5500.bin", "md5": [C, D],
|
|
"_cores": ["pcsx_rearmed"],
|
|
}
|
|
ADDED = {
|
|
"name": "scph7001.bin", "path": "scph7001.bin", "md5": [D, F],
|
|
"_cores": ["pcsx_rearmed"],
|
|
}
|
|
|
|
@staticmethod
|
|
def _export(truth_files: list[dict], script: str):
|
|
scraped = _scraped(script)
|
|
scraped["cores"] = ["pcsx_rearmed"]
|
|
systems, report = build_native_model(
|
|
{"systems": {"sony-playstation": {"files": truth_files}}}, scraped
|
|
)
|
|
exporter = discover_exporters()["batocera"]()
|
|
produced = exporter.render(
|
|
systems, report, {"batocera-systems": script}, scraped
|
|
)["batocera-systems"]
|
|
files = {fe.name: fe for system in systems.values() for fe in system.files}
|
|
return produced, landed_changes(exporter, systems), exporter, files
|
|
|
|
def test_a_string_item_with_several_accepted_becomes_a_list(self):
|
|
produced, changes, _, _ = self._export([self.CORRECTED], LISTING)
|
|
self.assertEqual(
|
|
produced, LISTING.replace(f'"md5": "{C}"', f'"md5": [ "{C}", "{D}" ]')
|
|
)
|
|
record = next(r for r in changes.records if r["kind"] == "hash_corrected")
|
|
self.assertEqual((record["old"], record["new"]), (C, [C, D]))
|
|
self.assertEqual(changes.lists, 1)
|
|
|
|
def test_an_addition_with_several_accepted_is_written_as_a_list(self):
|
|
produced, changes, exporter, files = self._export([self.ADDED], LISTING)
|
|
self.assertTrue(exporter.lands(files["scph7001.bin"]))
|
|
self.assertIn(
|
|
f'{{ "md5": [ "{D}", "{F}" ], "file": "bios/scph7001.bin", '
|
|
'"emulator": "libretro", "core": "pcsx_rearmed" }',
|
|
produced,
|
|
)
|
|
record = next(r for r in changes.records if r["kind"] == "added")
|
|
self.assertEqual(record["new"]["md5"], [D, F])
|
|
self.assertEqual(changes.lists, 1)
|
|
|
|
def test_an_empty_md5_is_filled_with_the_list(self):
|
|
"""Left empty while one value was all the item could hold."""
|
|
produced, changes, _, _ = self._export([self.CORRECTED], LISTING.replace(C, ""))
|
|
self.assertEqual(
|
|
produced, LISTING.replace(f'"md5": "{C}"', f'"md5": [ "{C}", "{D}" ]')
|
|
)
|
|
record = next(r for r in changes.records if r["kind"] == "hash_filled")
|
|
self.assertEqual(record["new"], [C, D])
|
|
strings = STRINGS.replace(C, "")
|
|
self.assertEqual(self._export([self.CORRECTED], strings)[0], strings)
|
|
|
|
def test_one_accepted_md5_stays_a_string(self):
|
|
produced, changes, _, _ = self._export([dict(self.CORRECTED, md5=D)], LISTING)
|
|
self.assertEqual(produced, LISTING.replace(C, D))
|
|
self.assertEqual(changes.lists, 0)
|
|
|
|
def test_without_the_checker_the_export_is_unchanged(self):
|
|
"""The string checker keeps the first md5 and refuses the addition,
|
|
as every export did before the checker read lists."""
|
|
produced, changes, exporter, files = self._export(
|
|
[self.CORRECTED, self.ADDED], STRINGS
|
|
)
|
|
self.assertEqual(produced, STRINGS)
|
|
self.assertFalse(exporter.lands(files["scph7001.bin"]))
|
|
self.assertEqual(changes.lists, 0)
|
|
|
|
def test_the_written_list_passes_the_checker(self):
|
|
"""Run the produced script: either listed md5 passes, another fails."""
|
|
produced, _, _, _ = self._export([dict(self.CORRECTED, md5=[C, SECOND])], LISTING)
|
|
namespace: dict = {"__name__": "batocera_systems"}
|
|
exec(compile(produced, "batocera-systems", "exec"), namespace) # noqa: S102
|
|
item = namespace["systems"]["psx"]["biosFiles"][0]
|
|
self.assertEqual(item["md5"], [C, SECOND])
|
|
scratch = REPO_ROOT / "tmp"
|
|
with tempfile.TemporaryDirectory(dir=scratch if scratch.is_dir() else None) as tmp:
|
|
(Path(tmp) / "bios").mkdir()
|
|
systems = {"psx": {"name": "PSX", "biosFiles": [item]}}
|
|
for payload, missing in (
|
|
(b"second", {}),
|
|
(b"other", {"psx": {"bios/scph5500.bin": "UNTESTED"}}),
|
|
):
|
|
(Path(tmp) / "bios" / "scph5500.bin").write_bytes(payload)
|
|
with self.subTest(payload):
|
|
self.assertEqual(namespace["checkBios"](systems, tmp, False), missing)
|
|
|
|
def test_an_item_with_altmd5_stays_a_string(self):
|
|
"""checkBios nests [md5, altmd5]: a list there matches no digest."""
|
|
exporter = discover_exporters()["batocera"]()
|
|
exporter._md5_lists = md5_list_support(LISTING)
|
|
platform = {"name": "pack.zip", "md5": A, "alt_md5": B, "zipped_file": "rom.bin"}
|
|
fe = NativeFile("pack.zip", "pack.zip", "psx", platform=platform,
|
|
truth={"md5": [A, B, D]})
|
|
self.assertFalse(exporter.md5_list(fe))
|
|
self.assertTrue(exporter.md5_list(
|
|
NativeFile("rom.zip", "rom.zip", "psx", truth={"md5": [A, B]},
|
|
platform={"name": "rom.zip", "md5": A, "zipped_file": "rom.bin"})
|
|
))
|
|
|
|
def test_the_formats_that_share_the_script_read_the_same_checker(self):
|
|
for name in ("rocknix", "amberelec", "emuelec", "knulli"):
|
|
fe = NativeFile("boot.bin", "boot.bin", "dc", truth={"md5": [A, B]})
|
|
with self.subTest(name):
|
|
exporter = discover_exporters()[name]()
|
|
self.assertFalse(exporter.md5_list(fe))
|
|
exporter._md5_lists = md5_list_support(LISTING)
|
|
self.assertTrue(exporter.md5_list(fe))
|
|
self.assertEqual(exporter.written_list(fe, "md5"), [A, B])
|
|
|
|
|
|
class SitePageShowsAList(unittest.TestCase):
|
|
def test_a_list_reads_as_any_of_its_values(self):
|
|
line = describe_change({
|
|
"kind": "hash_corrected", "system": "psx", "name": "scph5500.bin",
|
|
"field": "md5", "old": C, "new": [C, D],
|
|
})
|
|
self.assertEqual(line, f"`psx/scph5500.bin` md5 `{C}` becomes `{C}` or `{D}`")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|