chore: merge audit batch, truth and export

This commit is contained in:
Abdessamad Derraz committed 2026-11-05 13:16:03 +01:00
commit 50be709240
8 files changed
+445 -83

No files matched your search

+25 -4
View File
@@ -84,6 +84,26 @@ def comparable(entry: dict | None) -> dict:
return {k: v for k, v in entry.items() if k != "md5"}
def contradicted_hashes(truth_entry: dict, platform_entry: dict | None) -> list[str]:
"""The hash fields a platform entry declares against the truth.
A field contradicts when both sides declare it without a value in
common, among the hashes the code checks and the platform values a
profile can be compared with. export_native corrects exactly these and
diff_truth reports them: the diff read three fields and let one shared
value settle all of them, so it stayed silent on an md5 or a sha256
the export rewrote.
"""
checked = checked_hashes(truth_entry) or {}
theirs = comparable(platform_entry)
return [
field_name for field_name in HASH_FIELDS
if (ours := set(_hash_values(checked, field_name)))
and (held := set(_hash_values(theirs, field_name)))
and not ours & held
]
def is_label(entry: dict) -> bool:
"""A name only content-scanning cores give: no core opens that file.
@@ -595,17 +615,18 @@ def build_native_model(
if matched is not None:
matched.truth = truth_entry
checked = checked_hashes(truth_entry) or {}
contradicted = contradicted_hashes(truth_entry, matched.platform)
for field_name in HASH_FIELDS:
ours = set(_hash_values(checked, field_name))
if field_name == "md5" and matched.composite_md5:
continue
theirs = set(_hash_values(comparable(matched.platform), field_name))
if ours and theirs and not ours & theirs:
if field_name in contradicted:
matched.corrections.append(field_name)
report.hashes_corrected.append(
f"{matched.native_system}/{matched.name} {field_name}"
)
elif ours and not theirs:
elif _hash_values(checked, field_name) and not _hash_values(
comparable(matched.platform), field_name
):
matched.filled.append(field_name)
t_req = truth_entry.get("required")
p_req = (matched.platform or {}).get("required")
+10 -4
View File
@@ -86,8 +86,14 @@ class Exporter(BaseExporter):
return super().states(fe, field_name)
def written(self, fe: NativeFile, field_name: str) -> str:
"""An entry is checked against one md5, the first accepted."""
return fe.hash(field_name)
"""An entry is one record: its md5, size and url describe one dump.
The first value the truth accepts replaced MiSTer's own accepted
md5 while the url still served MiSTer's dump, which the Downloader
then refused (fetch_file_worker.py:127), and the report said
nothing. record_hash keeps the platform's value when it is accepted.
"""
return fe.record_hash(field_name)
def _by_path(self, systems: dict[str, NativeSystem]) -> dict[str, object]:
indexed: dict[str, object] = {}
@@ -119,7 +125,7 @@ class Exporter(BaseExporter):
fe = indexed.get(path)
if fe is None:
continue
md5 = fe.hash("md5")
md5 = self.written(fe, "md5")
if not md5:
# The record is checked by its md5: where the truth takes the
# entry without one, a new size beside the old hash describes
@@ -185,7 +191,7 @@ class Exporter(BaseExporter):
indexed = self._by_path(systems)
for path, fe in indexed.items():
declared = files.get(path)
md5 = fe.hash("md5")
md5 = self.written(fe, "md5")
if (
declared is not None and md5 and declared.get("hash") != md5
and path not in getattr(self, "_unsized", [])
+62 -20
View File
@@ -156,6 +156,30 @@ class Exporter(BaseExporter):
@classmethod
def _enumeration(cls, help_text: str) -> tuple[int, int, list[str]] | None:
"""The first BIOS clause's list of names to extend, or None."""
clauses = cls._clauses(help_text)
return clauses[0][2] if clauses else None
@classmethod
def _clauses(
cls, help_text: str
) -> list[tuple[int, list[str], tuple[int, int, list[str]] | None]]:
"""Each BIOS clause that names files: its start, its names, and the
list to extend or None when it cannot be extended.
flycast's help has a Dreamcast clause and a Naomi/Atomiswave one.
Only the first was read, and naomi2.zip and segasp.zip were added to
the Dreamcast list.
"""
found = []
for clause in _BIOS_CLAUSE.finditer(help_text):
names = [m.group(0) for m in cls._names_in(clause.group(0))]
if names:
found.append((clause.start(), names, cls._extendable(clause)))
return found
@classmethod
def _extendable(cls, clause: re.Match[str]) -> tuple[int, int, list[str]] | None:
"""The list of names to extend: its span and its names, or None.
A list of alternatives ("a or b", "a/b") is not an enumeration: an
@@ -164,9 +188,6 @@ class Exporter(BaseExporter):
are a list of examples ("eg. a, b") and one with words between its
names, which a rewrite would lose.
"""
clause = _BIOS_CLAUSE.search(help_text)
if clause is None:
return None
text = clause.group(0)
names = cls._names_in(text)
if not names:
@@ -308,13 +329,12 @@ class Exporter(BaseExporter):
by_core = self._files_by_core(systems)
# Aliases count as names we know: RetroPie writes dc_flash.bin where
# flycast's profile files it as an alias of another primary.
known: set[str] = set()
systems_of: dict[str, set[str]] = {}
for system in systems.values():
for fe in system.files:
known.add(fe.name.lower())
known.update(
str(a).lower() for a in (fe.native("aliases", []) or [])
)
for name in [fe.name, *(fe.native("aliases", []) or [])]:
systems_of.setdefault(str(name).lower(), set()).add(fe.native_system)
known = set(systems_of)
self._skipped: dict[str, list[str]] = {}
self._named: set[int] = set()
produced: dict[str, str] = {}
@@ -362,30 +382,52 @@ class Exporter(BaseExporter):
and (slot := self._slot_of(fe)) is not None
}
proposed = self._one_per_slot(candidates, filled)
missing = self._in_search_order(proposed)
if not missing:
if not proposed:
continue
enumeration = self._enumeration(help_text)
if enumeration is None:
# No enumeration to extend, and a sentence we would have to
# write ourselves is a documentation change, not a correction.
skip("whose help text lists no files to add to", module_id)
# A file joins the list of its own system: the clause whose
# names the platform files under that system, and only one.
clauses = self._clauses(help_text)
homes: dict[int, list[NativeFile]] = {}
homeless = False
for fe in proposed:
owners = [
i for i, (_, names, _) in enumerate(clauses)
if fe.native_system in set().union(
*(systems_of.get(n.lower(), set()) for n in names))
]
if len(owners) == 1 and clauses[owners[0]][2] is not None:
homes.setdefault(owners[0], []).append(fe)
else:
homeless = True
if homeless:
# No enumeration of that system to extend, and a sentence we
# would have to write ourselves is a documentation change,
# not a correction.
skip("whose help text lists no files of that system to add to", module_id)
if not homes:
continue
if len(listed) + len(missing) > self._longest:
added = sum(len(self._in_search_order(fes)) for fes in homes.values())
if len(listed) + added > self._longest:
skip("whose help text would grow longer than any it has today", module_id)
continue
new_help = self._extended(help_text, enumeration, missing)
if len(listed) + len(missing) > 1:
new_help = _BIOS_NOUN.sub(r"\1 files", new_help, count=1)
new_help = help_text
for i in sorted(homes, reverse=True):
start, _, enumeration = clauses[i]
missing = self._in_search_order(homes[i])
new_help = self._extended(new_help, enumeration, missing)
if len(enumeration[2]) + len(missing) > 1:
new_help = new_help[:start] + _BIOS_NOUN.sub(
r"\1 files", new_help[start:enumeration[0]], count=1
) + new_help[enumeration[0]:]
self._named.update(id(fe) for fe in homes[i])
produced[relative] = originals[relative].replace(
f'rp_module_help="{help_text}"',
f'rp_module_help="{new_help}"',
1,
)
self._named.update(id(fe) for fe in proposed)
return produced
+54 -45
View File
@@ -18,7 +18,7 @@ from common import (
resolve_platform_cores,
standalone_set as platform_standalone_set,
)
from hashing import compute_hashes, parse_md5_list
from hashing import compute_hashes
from validation import (
filter_files_by_mode,
installed_with_emulator,
@@ -113,7 +113,6 @@ _FILLED_FIELDS = (
"note",
"aliases",
"contents",
"region",
)
_COPIED_FIELDS = (
"sha1",
@@ -290,6 +289,27 @@ def _fold_into(existing: dict, file_entry: dict, emu_name: str) -> None:
if file_entry.get(field) is not None and existing.get(field) is None:
existing[field] = file_entry[field]
_merge_priority(existing, file_entry.get("priority"))
_merge_region(existing, file_entry.get("region"))
def _merge_region(existing: dict, theirs: str | list | None) -> None:
"""The regions every reader selects the file for, the pack's own rule.
region.build_region_index takes the union of the tags and keeps a file
untagged as soon as one reader declares it without a region. The first
tag used to win: psxonpsp660.bin came out japan from pcsx_rearmed while
SwanStation tags it world and beetle_psx not at all.
"""
if existing.get("_region_void"):
return
if not theirs:
existing.pop("region", None)
existing["_region_void"] = True
return
ours = existing.get("region") or []
ours = ours if isinstance(ours, list) else [ours]
added = [r for r in (theirs if isinstance(theirs, list) else [theirs]) if r not in ours]
existing["region"] = [*ours, *added]
def _merge_hashes(
@@ -373,6 +393,8 @@ def _new_truth_entry(file_entry: dict, emu_name: str, db: dict | None) -> dict:
crc = entry.get("crc32")
if isinstance(crc, str):
entry["crc32"] = crc.removeprefix("0x").lower()
if not file_entry.get("region"):
entry["_region_void"] = True
entry["_cores"] = {emu_name}
entry["_required_by"] = {emu_name} if file_entry.get("required") else set()
entry["_named_by"] = {emu_name} if file_entry.get("_opens_by_name", True) else set()
@@ -486,14 +508,19 @@ def _sizes_clash(a: dict, b: dict) -> bool:
return False
def _identity_values(entry: dict, db: dict | None) -> set[str]:
"""Every hash an entry declares, with the siblings the database knows."""
def _identity(entry: dict, db: dict | None) -> dict:
"""The entry with the hashes and size the database knows for its hash."""
probe = {k: entry[k] for k in ("sha1", "md5", "crc32", "size") if entry.get(k)}
if db and probe and not any(isinstance(v, list) for v in probe.values()):
_enrich_hashes(probe, db)
return {**entry, **{k: v for k, v in probe.items() if not entry.get(k)}}
def _identity_values(entry: dict) -> set[str]:
"""Every hash an identity declares."""
values: set[str] = set()
for field in ("sha1", "md5", "sha256", "crc32"):
values |= _declared(entry, field) | _declared(probe, field)
values |= _declared(entry, field)
return values
@@ -503,10 +530,14 @@ def _declaration_verdict(entry: dict, declaration: dict, db: dict | None) -> str
A shared hash proves it. Sizes no value of which agrees refute it, as
name_match_size_ok refutes a candidate. Two hashes at one size are two
revisions of one slot, as slots.py reads them, so they refute nothing.
The size the database holds for a declared hash counts as declared:
Batocera's BK BASIC10.ROM names only an md5, of a 24576-byte file, and
clk's 16 KB Oric basic10.rom was filed under the BK beside it.
"""
if _identity_values(entry, db) & _identity_values(declaration, db):
ours, theirs = _identity(entry, db), _identity(declaration, db)
if _identity_values(ours) & _identity_values(theirs):
return "same"
if _sizes_clash(entry, declaration):
if _sizes_clash(ours, theirs):
return "other"
return "undecided"
@@ -814,6 +845,7 @@ def generate_platform_truth(
for sys_id, sys_data in systems.items():
for fe in sys_data.get("files", []):
fe.pop("_hash_by", None)
fe.pop("_region_void", None)
fe["_cores"] = sorted(fe.get("_cores", set()))
fe["_required_by"] = sorted(fe.get("_required_by", set()))
fe["_named_by"] = sorted(fe.get("_named_by", set()))
@@ -846,53 +878,30 @@ def generate_platform_truth(
# Platform truth diffing
def _hash_values(entry: dict, field: str) -> set[str]:
"""The values a diffed entry declares for a hash field.
A profile writes a list, a platform YAML one comma-separated string
(Recalbox's multi-hash): both are read the way the exporter reads them.
A number is a hash YAML coerced and no longer spells: it is ignored.
"""
raw = entry.get(field)
return set(parse_md5_list(raw)) if isinstance(raw, (str, list)) else set()
def _hash_set(entry: dict) -> set[str]:
return {value for h in ("sha1", "md5", "crc32") for value in _hash_values(entry, h)}
def _path_tail(value: object) -> str:
return str(value or "").replace("\\", "/").casefold()
def _hash_mismatch(t_entry: dict, s_entry: dict) -> dict | None:
"""The first hash both sides declare without a value in common.
"""The first hash field the export would correct, by the export's rule.
A value shared on any field settles it: the truth holds every revision a
core accepts, and a profile may give one revision's sha1 without its md5.
A hash the code does not check contradicts nothing (checked_hashes), and
a platform md5 that is a composite of a zip's members is no container
md5 (comparable), the rule pair_entries applies.
contradicted_hashes is the one decision: a hash the code does not check
contradicts nothing (checked_hashes), and a platform md5 that is a
composite of a zip's members is no container md5 (comparable).
"""
from exporter.baseline import checked_hashes, comparable
from exporter.baseline import contradicted_hashes
s_entry = comparable(s_entry)
if _hash_set(t_entry) & _hash_set(s_entry):
fields = contradicted_hashes(t_entry, s_entry)
if not fields:
return None
t_entry = checked_hashes(t_entry) or {}
for h in ("sha1", "md5", "crc32"):
t_hash = t_entry.get(h, "")
s_hash = s_entry.get(h, "")
t_values, s_values = _hash_values(t_entry, h), _hash_values(s_entry, h)
if t_values and s_values and not t_values & s_values:
return {
"name": s_entry["name"],
"hash_type": h,
f"truth_{h}": t_hash,
f"scraped_{h}": s_hash,
"truth_cores": list(t_entry.get("_cores", [])),
}
return None
h = fields[0]
return {
"name": s_entry["name"],
"hash_type": h,
f"truth_{h}": t_entry.get(h, ""),
f"scraped_{h}": s_entry.get(h, ""),
"truth_cores": list(t_entry.get("_cores", [])),
}
def _diff_system(truth_sys: dict, scraped_sys: dict) -> dict:
+74 -7
View File
@@ -34,6 +34,8 @@ from common import (
parse_md5_list,
)
from hashing import parse_hash_list
from nativemode import digest_algorithm
from validation import frontend_digest_matches, inner_rom_check
try:
import yaml
@@ -140,7 +142,8 @@ def load_platform_hashes(platforms_dir: str, db: dict | None = None) -> dict:
bios/<Manufacturer>/<Console>/ folders the collection files its system
under, read from the other files of that system it holds.
"""
known = {"sha1": set(), "md5": set(), "names": set(), "folders": {}, "sizes": {}}
known = {"sha1": set(), "md5": set(), "sha256": set(), "crc32": set(),
"names": set(), "folders": {}, "sizes": {}, "entries": {}}
systems_of: dict[str, set[str]] = {}
folders_of: dict[str, dict[str, int]] = {}
files = (db or {}).get("files", {})
@@ -160,15 +163,31 @@ def load_platform_hashes(platforms_dir: str, db: dict | None = None) -> dict:
print(f"WARNING: {name}: {exc}", file=sys.stderr)
continue
mode, rule = config.get("verification_mode"), config.get("hash_rule", "")
for system_id, system in config.get("systems", {}).items():
for file_entry in system.get("files", []):
sha1s = parse_hash_list(file_entry.get("sha1"))
md5s = parse_md5_list(file_entry.get("md5"))
sha256s = parse_hash_list(file_entry.get("sha256"))
known["sha1"].update(sha1s)
known["md5"].update(md5s)
known["sha256"].update(sha256s)
if not (sha1s or md5s or sha256s) and isinstance(file_entry.get("size"), int):
# The resolver's reading: a crc32 counts where no stronger
# hash is declared, bounded by the declared size.
known["crc32"].update(
(crc.zfill(8), file_entry["size"])
for crc in parse_hash_list(file_entry.get("crc32"))
)
if "name" in file_entry:
known["names"].add(file_entry["name"])
_note_sizes(known["sizes"], [file_entry["name"]], file_entry)
# A frontend that hashes a ZIP's member or its members
# together, or that takes any hash at the declared size,
# answers by its own rule, read per entry.
known["entries"].setdefault(file_entry["name"].lower(), []).append(
(file_entry, mode, rule, name)
)
for value in (*sha1s, *md5s):
systems_of.setdefault(value, set()).add(system_id)
_note_folder(folders_of, system_id, files, by_md5, sha1s, md5s)
@@ -354,6 +373,14 @@ def validate_file(
sha1_known = result.sha1 in platform_hashes.get("sha1", set())
md5_known = result.md5 in platform_hashes.get("md5", set())
other_known = (
hashes.get("sha256", "") in platform_hashes.get("sha256", set())
or (result.crc32, result.size) in platform_hashes.get("crc32", set())
)
accepted_by = (
set() if sha1_known or md5_known or other_known
else _platforms_accepting(filepath, result.filename, platform_hashes)
)
name_known = result.filename in platform_hashes.get("names", set())
profile_hashes = profile_hashes or {}
@@ -369,6 +396,12 @@ def validate_file(
result.add_check(True, "SHA1 matches known platform requirement")
elif md5_known:
result.add_check(True, "MD5 matches known platform requirement")
elif other_known:
result.add_check(True, "Hash matches known platform requirement")
elif accepted_by:
result.add_check(
True, "Accepted by the platform's own check: " + ", ".join(sorted(accepted_by))
)
elif by_profile:
names = sorted(by_profile)
shown = ", ".join(names[:5]) + (f" and {len(names) - 5} more" if len(names) > 5 else "")
@@ -415,10 +448,10 @@ def validate_file(
"File is not under bios/ yet. Its system's files are in "
+ " or ".join(f"`{f}`" for f in expected_folders)
)
elif _variant_home(db, result.filename):
elif _variant_home(db, result.filename, result.sha1):
result.add_warning(
"File is not under bios/ yet. The collection keeps this name in "
f"`{_variant_home(db, result.filename)}`"
f"`{_variant_home(db, result.filename, result.sha1)}`"
)
else:
result.add_warning(
@@ -439,7 +472,8 @@ def validate_file(
f"this file has {result.size:,}: truncated or another file"
)
if name_known and not sha1_known and not md5_known and not by_profile and size_fits:
if (name_known and not (sha1_known or md5_known or other_known or accepted_by)
and not by_profile and size_fits):
home = _variant_home(db, result.filename)
result.add_info(
"This may be a valid variant. If accepted, it will be placed in "
@@ -449,10 +483,43 @@ def validate_file(
return result
def _variant_home(db: dict | None, name: str) -> str:
"""The folder the collection keeps a name in, for its variants beside it."""
def _platforms_accepting(filepath: str, filename: str, platform_hashes: dict) -> set[str]:
"""Platforms whose declaration of this name accepts the file by their
own check, as verify and the pack builder read it.
Batocera pins a ROM inside the ZIP (zipped_file), Recalbox hashes a
ZIP's members together, RomM takes any hash at the declared size: the
container's sha1 or md5 is in none of those lists, and adam_fdc.zip,
which Batocera accepts, was called an unknown variant.
"""
accepted: set[str] = set()
for entry, mode, rule, platform in platform_hashes.get("entries", {}).get(
filename.lower(), []
):
if platform in accepted:
continue
if entry.get("zipped_file"):
ok = inner_rom_check(entry, filepath) == "ok"
else:
algorithm = digest_algorithm(mode)
ok = bool(algorithm or rule) and frontend_digest_matches(
entry, filepath, algorithm or "", rule
)
if ok:
accepted.add(platform)
return accepted
def _variant_home(db: dict | None, name: str, held: str = "") -> str:
"""The folder the collection keeps a name in, for its variants beside it.
The folder of an identical copy comes first: adam_fdc.zip, held in
bios/Coleco/Adam/, was sent to bios/Arcade/MAME/, the first folder
the name index gave.
"""
files = (db or {}).get("files", {})
for sha1 in (db or {}).get("indexes", {}).get("by_name", {}).get(name, []):
by_name = (db or {}).get("indexes", {}).get("by_name", {}).get(name, [])
for sha1 in ([held] if held in files else []) + list(by_name):
path = str(files.get(sha1, {}).get("path", ""))
folder = path.rsplit("/", 1)[0] + "/" if "/" in path else ""
if folder and "/.variants" not in "/" + folder:
+55
View File
@@ -737,6 +737,40 @@ class SearchOrder(unittest.TestCase):
self.assertNotIn("us_scd2_9306", produced)
self.assertNotIn("jp_mcd1_9112", produced)
def test_a_file_joins_the_list_of_its_own_system(self):
"""flycast's help names the Dreamcast BIOS, then the Naomi/Atomiswave
ones. naomi2.zip, an arcade archive, went into the Dreamcast list:
only the first BIOS clause was read."""
from exporter.baseline import NativeSystem # noqa: PLC0415
def entry(name, system):
return NativeFile(name=name, destination=f"dc/{name}", native_system=system,
truth={"name": name, "required": True, "md5": "0" * 32,
"_cores": ["flycast"], "_required_by": ["flycast"]})
dc, arcade = "Sega - Dreamcast", "Sega - Dreamcast-based Arcade"
systems = {
dc: NativeSystem(dc, files=[entry("dc_boot.bin", dc)]),
arcade: NativeSystem(arcade, files=[entry("naomi.zip", arcade),
entry("awbios.zip", arcade),
entry("naomi2.zip", arcade)]),
"other": NativeSystem("other", files=[entry("other.zip", "other")]),
}
script = ('rp_module_id="lr-flycast"\nrp_module_help="Copy the required Dreamcast '
'BIOS file dc_boot.bin to $biosdir/dc\\n\\nCopy the required '
'Naomi/Atomiswave BIOS files naomi.zip and awbios.zip to $biosdir/dc"\n')
longest = ('rp_module_id="xm7"\nrp_module_help="Copy bios files '
+ ", ".join(f"n{i}.rom" for i in range(12)) + ' and n12.rom to $biosdir"\n')
exporter = discover_exporters()["retropie"]()
produced = exporter.render(
systems, None,
{"scriptmodules/libretrocores/lr-flycast.sh": script,
"scriptmodules/emulators/xm7.sh": longest},
)["scriptmodules/libretrocores/lr-flycast.sh"]
self.assertIn("Dreamcast BIOS file dc_boot.bin to", produced)
self.assertIn("BIOS files naomi.zip, awbios.zip and naomi2.zip to", produced)
self.assertNotIn("other.zip", produced)
def test_cores_that_disagree_keep_the_best_rank(self):
"""Nothing is dropped on the rank here, so the best one wins.
@@ -1849,6 +1883,27 @@ class MisterDatabaseComesBackAsWritten(unittest.TestCase):
self.assertEqual(len(exporter.withheld()), 1)
self.assertEqual(exporter.validate(systems, produced), [])
def test_an_accepted_hash_is_not_swapped_for_another(self):
"""The url serves MiSTer's dump, and the truth accepts it among two.
The first accepted md5 went in its place with no change reported,
and the Downloader refused every install of the file."""
database = {"db_id": "bios_db", "files": {
"games/X/boot.rom": {"hash": "b" * 32, "size": 10, "url": "u"},
}}
original = json.dumps(database)
scraped = {"systems": {"x": {"files": [{
"name": "boot.rom", "destination": "X/boot.rom", "md5": "b" * 32, "size": 10,
}]}}}
truth = {"systems": {"x": {"files": [{
"name": "boot.rom", "path": "X/boot.rom", "md5": ["a" * 32, "b" * 32],
"size": 10, "_cores": ["x"],
}]}}}
exporter = discover_exporters()["misterfpga"]()
systems, report = build_native_model(truth, scraped)
produced = exporter.render(systems, report, {"bios_db.json": original}, scraped)
self.assertEqual(produced["bios_db.json"], original)
self.assertEqual(exporter.validate(systems, produced), [])
def test_an_empty_truth_hands_back_the_database(self):
archive = UPSTREAM_CACHE / "misterfpga" / "bios_db.json.zip"
if not archive.exists():
+92 -2
View File
@@ -177,10 +177,13 @@ class EveryRevisionACoreAcceptsStays(unittest.TestCase):
def test_a_revision_known_by_one_hash_is_no_mismatch(self):
"""1.04 carries md5 and sha1, 1.07 only its sha1: a platform entry
giving 1.07's sha1 and md5 is that revision, not a mismatch."""
giving 1.07's sha1 and md5 is that revision, not a mismatch. dosbox
checks the sha1 (validation), so its md5 describes one revision and
contradicts nothing, the rule the export applies too."""
from truth import _hash_mismatch
truth = {"name": "MT32_CONTROL.ROM", "md5": "c" * 32, "sha1": ["a" * 40, "b" * 40]}
truth = {"name": "MT32_CONTROL.ROM", "md5": "c" * 32, "sha1": ["a" * 40, "b" * 40],
"validation": ["sha1"]}
scraped = {"name": "MT32_CONTROL.ROM", "md5": "d" * 32, "sha1": "b" * 40}
self.assertIsNone(_hash_mismatch(truth, scraped))
self.assertEqual(
@@ -189,6 +192,40 @@ class EveryRevisionACoreAcceptsStays(unittest.TestCase):
class DiffAndExportShareOneMismatchRule(unittest.TestCase):
"""diff_truth let one shared value settle every field and read no
sha256, while the export corrected each field on its own: the diff
stayed silent on hashes the export rewrote."""
CASES = (
({"sha1": "a" * 40, "md5": "1" * 32}, {"sha1": "a" * 40, "md5": "2" * 32}),
({"sha256": "a" * 64}, {"sha256": "b" * 64}),
({"sha1": "a" * 40, "md5": "1" * 32}, {"sha1": "a" * 40, "md5": "1" * 32}),
)
def test_the_diff_names_what_the_export_corrects(self):
from exporter.baseline import build_native_model # noqa: PLC0415
from truth import diff_platform_truth # noqa: PLC0415
for t_hashes, p_hashes in self.CASES:
with self.subTest(truth=t_hashes, platform=p_hashes):
truth = {"systems": {"s": {"files": [
{"name": "f.bin", "required": True, "_cores": ["c"], **t_hashes}]}}}
scraped = {"systems": {"s": {"files": [
{"name": "f.bin", "destination": "f.bin", "required": True, **p_hashes}]}}}
diff = diff_platform_truth(truth, scraped)
reported = {
m["hash_type"]
for system in diff.get("divergences", {}).values()
for m in system.get("hash_mismatch", [])
}
_, report = build_native_model(truth, scraped)
corrected = {item.split()[-1] for item in report.hashes_corrected}
# One record per entry, naming the first field the export corrects.
self.assertEqual(bool(reported), bool(corrected))
self.assertLessEqual(reported, corrected)
def _filed(truth: dict, name: str) -> list[str]:
"""The systems an entry of that name was filed under."""
return sorted(
@@ -454,5 +491,58 @@ class TheModelIsAFunctionOfItsInputs(unittest.TestCase):
)
class DatabaseSizeRefutesASameNamedDeclaration(unittest.TestCase):
"""Batocera's BK system declares BASIC10.ROM by md5 alone. clk's Oric
basic10.rom, 16 KB with its own system:, was filed under the BK and
added to its list: the 24576 bytes the database holds for that md5
were never compared."""
def test_the_size_behind_a_declared_hash_counts(self):
bk_md5, oric_md5 = "3" * 32, "e" * 32
db = {"files": {"1" * 40: {"sha1": "1" * 40, "md5": bk_md5, "size": 24576}},
"indexes": {"by_md5": {bk_md5: "1" * 40}}}
profiles = {"clk": {"emulator": "clk", "type": "libretro", "systems": ["oric"],
"files": [{"name": "basic10.rom", "path": "Oric/basic10.rom",
"system": "oric", "size": 16384, "md5": oric_md5}]}}
config = {"cores": ["clk"], "systems": {"elektronika-bk": {"files": [
{"name": "BASIC10.ROM", "md5": bk_md5}]}}}
truth = generate_platform_truth("p", config, {}, profiles, db=db)
self.assertNotIn("basic10.rom", {
f["name"] for f in truth["systems"].get("elektronika-bk", {}).get("files", [])})
class RegionFollowsEveryReader(unittest.TestCase):
"""The first core to tag a file set its region in the truth: the ST-V
BIOS kept beetle_saturn's Americas and lost mednafen_stv's Korea, and
psxonpsp660.bin read japan although beetle_psx declares it untagged.
The pack's region index takes the union and voids it on an untagged
reader, and the truth now says the same."""
@staticmethod
def _profile(name: str, region: list[str] | None) -> dict:
entry = {"name": "bios.bin", "system": "sega-stv", "md5": "a" * 32}
if region is not None:
entry["region"] = region
return {"emulator": name, "type": "libretro", "systems": ["sega-stv"],
"files": [entry]}
def _region(self, profiles: dict) -> object:
config = {"cores": sorted(profiles), "systems": {"sega-stv": {"files": []}}}
truth = generate_platform_truth("p", config, {}, profiles)
(entry,) = truth["systems"]["sega-stv"]["files"]
self.assertNotIn("_region_void", entry)
return entry.get("region")
def test_tags_of_every_reader_are_kept(self):
region = self._region({"a": self._profile("a", ["north-america"]),
"b": self._profile("b", ["south-korea", "north-america"])})
self.assertEqual(region, ["north-america", "south-korea"])
def test_an_untagged_reader_leaves_no_region(self):
for order in (("a", "b"), ("b", "a")):
profiles = {"a": self._profile("a", ["japan"]), "b": self._profile("b", None)}
self.assertIsNone(self._region({k: profiles[k] for k in order}))
if __name__ == "__main__":
unittest.main()
+73 -1
View File
@@ -329,7 +329,8 @@ class Reporting(_Fixture):
class PlatformHashLoading(_Fixture):
def test_missing_directory_yields_empty_sets(self):
known = validate_pr.load_platform_hashes(str(self.tmp / "absent"))
self.assertEqual(known, {"sha1": set(), "md5": set(), "names": set(), "folders": {}, "sizes": {}})
self.assertEqual(known, {"sha1": set(), "md5": set(), "sha256": set(), "crc32": set(),
"names": set(), "folders": {}, "sizes": {}, "entries": {}})
def test_declared_hashes_and_names_are_collected(self):
platforms = self.tmp / "platforms"
@@ -352,6 +353,77 @@ class PlatformHashLoading(_Fixture):
self.assertIn("boot.bin", known["names"])
class ThePlatformsOwnCheckDecides(_Fixture):
"""Batocera pins a ROM inside adam_fdc.zip, Recalbox hashes a ZIP's
members, RetroDECK declares a sha256: the container's sha1 and md5 are
in none of those lists, and a file verify counts OK was reported as an
unknown variant to be filed under .variants/."""
def _platforms(self, registry: str, systems: str) -> dict:
platforms = self.tmp / "platforms"
platforms.mkdir()
(platforms / "_registry.yml").write_text(
f"platforms:\n p:\n config: p.yml\n status: active\n{registry}",
encoding="utf-8")
(platforms / "p.yml").write_text(f"platform: P\nsystems:\n s:\n files:\n{systems}",
encoding="utf-8")
return validate_pr.load_platform_hashes(str(platforms))
def _zip(self) -> str:
import zipfile # noqa: PLC0415
path = self.tmp / "contrib" / "set.zip"
path.parent.mkdir()
with zipfile.ZipFile(path, "w") as archive:
archive.writestr("rom.u10", b"ROM")
return "contrib/set.zip"
def _verdict(self, path: str, known: dict) -> str:
return validate_pr.validate_file(path, None, known).to_markdown()
def test_a_member_pinned_by_zipped_file(self):
import hashlib # noqa: PLC0415
md5 = hashlib.md5(b"ROM").hexdigest()
known = self._platforms(
" verification_mode: md5\n",
f" - {{name: set.zip, md5: {md5}, zipped_file: rom.u10}}\n")
markdown = self._verdict(self._zip(), known)
self.assertIn("Accepted by the platform's own check: p", markdown)
self.assertNotIn(".variants/", markdown)
def test_a_member_digest(self):
from hashing import md5_composite # noqa: PLC0415
path = self._zip()
known = self._platforms(
" verification_mode: md5\n zip_digest: composite\n",
f" - {{name: set.zip, md5: {md5_composite(path)}}}\n")
self.assertIn("Accepted by the platform's own check: p", self._verdict(path, known))
def test_a_sha256(self):
import hashlib # noqa: PLC0415
path = self._zip()
digest = hashlib.sha256((self.tmp / path).read_bytes()).hexdigest()
known = self._platforms(" verification_mode: md5\n",
f" - {{name: set.zip, sha256: {digest}}}\n")
self.assertIn("Hash matches known platform requirement", self._verdict(path, known))
def test_an_identical_copy_names_the_folder(self):
db = {"files": {"m" * 40: {"path": "bios/Arcade/MAME/set.zip"},
"c" * 40: {"path": "bios/Coleco/Adam/set.zip"}},
"indexes": {"by_name": {"set.zip": ["m" * 40, "c" * 40]}}}
self.assertEqual(validate_pr._variant_home(db, "set.zip", "c" * 40), "bios/Coleco/Adam/")
self.assertEqual(validate_pr._variant_home(db, "set.zip"), "bios/Arcade/MAME/")
def test_another_member_is_still_a_variant(self):
known = self._platforms(
" verification_mode: md5\n",
f" - {{name: set.zip, md5: {'0' * 32}, zipped_file: rom.u10}}\n")
self.assertIn("may be a variant", self._verdict(self._zip(), known))
class FlagsAreAppliedOrRefused(unittest.TestCase):
"""Files named beside --changed were dropped for the git diff and the run
said "No changed BIOS files detected" about files it never read; --json