feat: read batocera md5 written as a list

This commit is contained in:
Abdessamad Derraz committed 2026-10-23 07:39:57 +02:00
commit bb561f2f12
9 files changed
+298 -13

No files matched your search

+4
View File
@@ -510,6 +510,10 @@
"type": "string",
"description": "Display name the platform gives its own system"
},
"md5_list": {
"type": "boolean",
"description": "Batocera: the platform writes md5 as a list of strings, any of which matches. md5 holds the items comma-joined, in order"
},
"alt_md5": {
"type": "string",
"pattern": "^[a-fA-F0-9]{32}(,[a-fA-F0-9]{32})*$",
+40 -5
View File
@@ -23,7 +23,9 @@ _ENTRY_START = re.compile(r'^(\s{4})"([^"]+)":\s*\{')
_INDENT = " " * 4
# One BIOS item: a dict literal with no nested braces.
_ITEM = re.compile(r"\{[^{}\n]*\}")
_MD5_VALUE = re.compile(r'("md5"\s*:\s*")[^"]*(")')
# The md5 of one item: a string, or a list of strings since batocera.linux
# 6b08b873 (checkBios accepts any of them).
_MD5_VALUE = re.compile(r'("md5"\s*:\s*)("[^"]*"|\[[^\[\]]*\])')
_STRING = re.compile(r'"(?:[^"\\\n]|\\.)*"')
@@ -56,6 +58,17 @@ def _closing_line(text: str, start: int) -> int | None:
return None
def _md5_literal(value: str | list[str], spaced: bool = True) -> str:
"""An md5 as the script writes it: a quoted string, or a list of them
with Batocera's spacing inside the brackets (`[ "a", "b" ]`)."""
if isinstance(value, str):
return f'"{value}"'
if not value:
return "[]"
inner = ", ".join(f'"{v}"' for v in value)
return f"[ {inner} ]" if spaced else f"[{inner}]"
def _note_key(text: str) -> tuple[str, str]:
"""(file, zippedFile) of the item a line holds, ("", "") for none."""
path = re.search(r'"file"\s*:\s*"([^"]+)"', text)
@@ -68,7 +81,8 @@ class Exporter(BaseExporter):
# checkBios compares one md5 for a plain file. altmd5 is read only for
# a zippedFile: retrodream's boot.bin written with the first of four
# md5 refused the three other dumps the core accepts.
# md5 refused the three other dumps the core accepts. An item the
# platform writes with a list of md5 keeps its list (md5_list).
SINGLE_HASH_FIELD = "md5"
# checkBios reads emulator names from es_systems, and Batocera's
@@ -164,8 +178,25 @@ class Exporter(BaseExporter):
return False
return fe.platform is not None or self._addition_keys(fe) is not None
@staticmethod
def md5_list(fe: NativeFile) -> bool:
"""Whether the platform writes this item's md5 as a list. Its
checker then accepts any listed value, so every one is written."""
return fe.platform is not None and bool(fe.platform.get("md5_list"))
def md5_value(self, fe: NativeFile) -> str | list[str]:
"""The md5 an item is written with, in the form the platform used."""
if self.md5_list(fe):
return fe.hashes("md5")
return self.single_value(fe, "md5")
def written(self, fe: NativeFile, field_name: str) -> str:
if field_name == "md5" and self.md5_list(fe):
return ",".join(fe.hashes("md5"))
return super().written(fe, field_name)
def states(self, fe: NativeFile, field_name: str) -> bool:
if field_name == "md5" and "md5" in fe.filled:
if field_name == "md5" and "md5" in fe.filled and not self.md5_list(fe):
return bool(self.single_value(fe, "md5"))
return super().states(fe, field_name)
@@ -222,7 +253,7 @@ class Exporter(BaseExporter):
def _bios_item(self, fe: NativeFile, runner: list[str] | None, originals: dict) -> str:
# The platform states an unhashed file as an empty md5 rather
# than leaving it out, and so do we.
item = [f'"md5": "{self.single_value(fe, "md5")}"']
item = [f'"md5": {_md5_literal(self.md5_value(fe))}']
alt = fe.native("alt_md5", "")
if alt:
item.append(f'"altmd5": "{alt}"')
@@ -379,7 +410,11 @@ class Exporter(BaseExporter):
if {k for k in {*ours, *theirs} if ours.get(k) != theirs.get(k)} - {"md5"}:
return None
patched = _MD5_VALUE.sub(
lambda m: m.group(1) + str(ours["md5"]) + m.group(2), match.group(0), count=1
lambda m: m.group(1) + _md5_literal(
ours["md5"], spaced=not m.group(2).startswith("[\"")
),
match.group(0),
count=1,
)
edits.append((match.start(), match.end(), patched))
if last is None:
+19
View File
@@ -74,6 +74,25 @@ class ChangeSet:
MAX_RESPONSE_SIZE = 50 * 1024 * 1024 # 50 MB
def md5_field(raw: object) -> tuple[list[str], bool]:
"""The md5 values a batocera-systems item declares, and whether it
writes them as a list.
Since batocera.linux 6b08b873 an item's md5 is one string or a list of
strings, and checkBios accepts a file whose md5 is any of them. The
platform YAML joins the values with commas, as it does for Recalbox, and
records the list form so the export writes a list back as a list: a
one-item list and an empty one are not the same text as a string.
"""
if raw is None:
return [], False
if isinstance(raw, str):
return ([raw] if raw else []), False
if isinstance(raw, list) and all(isinstance(item, str) for item in raw):
return [item for item in raw if item], True
raise ValueError(f"md5 is neither a string nor a list of strings: {raw!r}")
def requirement_entry(req: BiosRequirement) -> dict:
"""Serialize a requirement to a platform YAML file entry.
+9 -2
View File
@@ -24,6 +24,7 @@ from .base_scraper import (
BaseScraper,
BiosRequirement,
github_headers,
md5_field,
requirement_entry,
)
from .batocera_launchers import extract_launchers
@@ -420,10 +421,16 @@ def requirements_from_systems(
for bios in sys_data.get("biosFiles", []):
declared_path = bios.get("file", "")
file_path = declared_path
md5 = _resolve_truncated_md5(bios.get("md5", ""), md5_index)
values, as_list = md5_field(bios.get("md5"))
md5 = ",".join(
_resolve_truncated_md5(value, md5_index) for value in values
)
zipped_file = bios.get("zippedFile", "")
if strip and file_path.startswith(strip):
file_path = file_path[len(strip):]
native: dict[str, object] = {"native_name": sys_data.get("name", "")}
if as_list:
native["md5_list"] = True
requirements.append(
BiosRequirement(
name=file_path.split("/")[-1],
@@ -434,7 +441,7 @@ def requirements_from_systems(
required=True,
zipped_file=zipped_file or None,
native_id=sys_key,
native={"native_name": sys_data.get("name", "")},
native=native,
)
)
return requirements
+15 -3
View File
@@ -22,6 +22,7 @@ try:
BaseScraper,
BiosRequirement,
fetch_github_latest_version,
md5_field,
requirement_entry,
)
from .rocknix_launchers import extract_launchers
@@ -31,6 +32,7 @@ except ImportError:
BaseScraper,
BiosRequirement,
fetch_github_latest_version,
md5_field,
requirement_entry,
)
from rocknix_launchers import extract_launchers
@@ -133,6 +135,10 @@ _SYSTEM_HEADER = re.compile(
r'\s*"biosFiles"\s*:\s*\[',
)
_FIELD = re.compile(r'"(\w+)"\s*:\s*"([^"]*)"')
# A field written as a list of strings (`"md5": [ "a", "b" ]`), the form
# batocera.linux 6b08b873 gives the md5 of a file several dumps satisfy.
_LIST_FIELD = re.compile(r'"(\w+)"\s*:\s*\[([^\[\]]*)\]')
_LIST_ITEM = re.compile(r'"([^"]*)"')
def _slice_bios_block(source: str, start: int) -> tuple[str, int]:
@@ -162,7 +168,9 @@ def parse_systems(source: str) -> dict[str, dict]:
body, _ = _slice_bios_block(source, header.end() - 1)
files = []
for raw_entry in re.findall(r"\{[^{}]*\}", body):
fields = dict(_FIELD.findall(raw_entry))
fields: dict[str, str | list[str]] = dict(_FIELD.findall(raw_entry))
for key, items in _LIST_FIELD.findall(raw_entry):
fields[key] = _LIST_ITEM.findall(items)
if fields.get("file"):
files.append(fields)
systems[header.group("key")] = {
@@ -236,15 +244,19 @@ class Scraper(BaseScraper):
# Declared paths are relative to the storage root: strip the
# leading bios/ so destinations match base_destination
destination = path[len("bios/"):] if path.startswith("bios/") else path
values, as_list = md5_field(bios.get("md5"))
native: dict[str, object] = {"native_name": entry["name"]}
if as_list:
native["md5_list"] = True
req = BiosRequirement(
name=destination.rsplit("/", 1)[-1],
system=system_slug,
md5=bios.get("md5", "") or None,
md5=",".join(values) or None,
alt_md5=bios.get("altmd5", "") or None,
destination=destination,
required=True,
native_id=key,
native={"native_name": entry["name"]},
native=native,
)
if bios.get("zippedFile"):
req.zipped_file = bios["zippedFile"]
+208
View File
@@ -0,0 +1,208 @@
"""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 exporter.baseline import build_native_model # 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 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)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -529,7 +529,7 @@ class ShortMd5Handling(unittest.TestCase):
from scraper import batocera_scraper
scraper = inspect.getsource(batocera_scraper)
self.assertIn("md5 = _resolve_truncated_md5(", scraper)
self.assertIn("_resolve_truncated_md5(value, md5_index) for value in values", scraper)
resolver = inspect.getsource(common.resolve_local_file)
self.assertIn("if len(md5_candidate) < 32:", resolver)
self.assertIn("_md5_name_ok(path)", resolver)
+1 -1
View File
@@ -86,7 +86,7 @@ def validate_format(self, raw_data: str) -> bool:
| `alt_md5` | `str \| None` | A second accepted MD5 (Batocera and ROCKNIX `altmd5`) |
| `native_path` | `str \| None` | The file reference as the platform writes it, kept when it differs from `name`. libretro writes `iplromco.dat` bare although the file lives under `keropi/`, and `ep128emu/roms/cpc464.rom` with its directory |
| `mirror_of` | `str \| None` | The native system whose entry this one repeats under another system. The platform's own file does not carry the copy, so the export skips it |
| `native` | `dict` | Fields the platform declares that the model has no place for, written back unchanged by the export: `native_name`, `core`, `note`, `description`, `hash_match_mandatory`, `mandatory_declared`, `alt_paths`, `component`, `required_label` |
| `native` | `dict` | Fields the platform declares that the model has no place for, written back unchanged by the export: `native_name`, `md5_list` (a Batocera md5 written as a list), `core`, `note`, `description`, `hash_match_mandatory`, `mandatory_declared`, `alt_paths`, `component`, `required_label` |
`requirement_entry()` in `base_scraper.py` is the single serializer from a
requirement to a platform YAML entry, and every scraper goes through it.
+1 -1
View File
@@ -54,7 +54,7 @@ Every module in `tests/`, by the area it guards.
| Installer and downloads | loopback server | `test_check_release`, `test_download`, `test_install`, `test_install_cores`, `test_install_experience`, `test_install_selection` |
| Handheld port | stubbed PortMaster dialogs, fake installer | `test_port` |
| Truth and native export | synthetic + cached originals | `test_export_counts`, `test_export_native_layer`, `test_export_report`, `test_exporters`, `test_truth_data_dir_readers`, `test_truth_diff`, `test_truth_layout` |
| Scrapers, upstream and provenance | inline sources, synthetic | `test_check_freshness`, `test_dedup`, `test_elec_scrapers`, `test_fbneo_parser`, `test_find_pinned_revision`, `test_hash_merge`, `test_launchers`, `test_launchers_emudeck`, `test_launchers_recalbox`, `test_launchers_retrobat`, `test_launchers_retrodeck`, `test_launchers_retropie`, `test_launchers_rocknix`, `test_mame_devices`, `test_mame_parser`, `test_mame_ref_audit`, `test_profile_sync`, `test_provenance`, `test_refresh_data_dirs`, `test_refresh_failures`, `test_refresh_stale`, `test_retrodeck_scraper_errors`, `test_romset_recipes`, `test_scraper_contract`, `test_upstream`, `test_upstream_index` |
| Scrapers, upstream and provenance | inline sources, synthetic | `test_batocera_md5_list`, `test_check_freshness`, `test_dedup`, `test_elec_scrapers`, `test_fbneo_parser`, `test_find_pinned_revision`, `test_hash_merge`, `test_launchers`, `test_launchers_emudeck`, `test_launchers_recalbox`, `test_launchers_retrobat`, `test_launchers_retrodeck`, `test_launchers_retropie`, `test_launchers_rocknix`, `test_mame_devices`, `test_mame_parser`, `test_mame_ref_audit`, `test_profile_sync`, `test_provenance`, `test_refresh_data_dirs`, `test_refresh_failures`, `test_refresh_stale`, `test_retrodeck_scraper_errors`, `test_romset_recipes`, `test_scraper_contract`, `test_upstream`, `test_upstream_index` |
| Site, docs and repository contracts | rendered pages, sources | `test_contributing_text`, `test_coverage_cache`, `test_data_dir_refs`, `test_dataclass_defaults`, `test_doc_pages`, `test_doc_surfaces`, `test_optional_dependencies`, `test_prose_style`, `test_schema_semantic_gate`, `test_site_availability`, `test_site_content_data`, `test_site_copy`, `test_site_cores`, `test_site_emulator_page`, `test_site_exports`, `test_site_funnel`, `test_site_images`, `test_site_picker`, `test_site_platform_hashes`, `test_site_r2_components`, `test_site_r2_css`, `test_site_r2_emulators`, `test_site_r2_entry`, `test_site_r2_nav`, `test_site_r2_p0`, `test_site_r2_reference`, `test_site_r2_systems`, `test_site_r2_wiki`, `test_site_render`, `test_site_system_links`, `test_site_validation`, `test_site_weight`, `test_site_write`, `test_validate_pr`, `test_validate_pr_flow`, `test_validate_pr_hashes`, `test_workflow_paths` |
## Test architecture