mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-04-13 12:22:33 -05:00
754e829b35
Extract each platform ZIP to tmp/ (real filesystem, not /tmp tmpfs) and verify every declared file exists at the correct path with the correct hash per the platform's native verification mode. Handles ZIP inner content verification (checkInsideZip, md5_composite, inner ROM MD5) and path collision deduplication. Integrated as pipeline step 6/8. Renumber all pipeline steps to be sequential (was skipping from 5 to 8).
265 lines
9.6 KiB
Python
265 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""End-to-end pack integrity test.
|
|
|
|
Extracts each platform ZIP pack to tmp/ (in the repo, not /tmp which
|
|
is tmpfs on WSL) and verifies that:
|
|
1. The archive is not corrupt and fully decompressable
|
|
2. Every file declared in the platform YAML exists at the correct path
|
|
3. Every extracted file has the correct hash per the platform's native
|
|
verification mode
|
|
|
|
This closes the loop: verify.py checks source bios/ -> this script
|
|
checks the final delivered ZIP the user actually downloads.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import unittest
|
|
import zipfile
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
|
|
from common import check_inside_zip, load_platform_config, md5_composite
|
|
|
|
REPO_ROOT = os.path.join(os.path.dirname(__file__), "..")
|
|
DIST_DIR = os.path.join(REPO_ROOT, "dist")
|
|
PLATFORMS_DIR = os.path.join(REPO_ROOT, "platforms")
|
|
TMP_DIR = os.path.join(REPO_ROOT, "tmp", "pack_test")
|
|
|
|
|
|
def _find_zip(platform_name: str) -> str | None:
|
|
"""Find the ZIP pack for a platform in dist/."""
|
|
if not os.path.isdir(DIST_DIR):
|
|
return None
|
|
config = load_platform_config(platform_name, PLATFORMS_DIR)
|
|
display = config.get("platform", platform_name).replace(" ", "_")
|
|
for f in os.listdir(DIST_DIR):
|
|
if f.endswith("_BIOS_Pack.zip") and display in f:
|
|
return os.path.join(DIST_DIR, f)
|
|
return None
|
|
|
|
|
|
def _hash_file(path: str, algo: str) -> str:
|
|
"""Compute hash of a file on disk."""
|
|
h = hashlib.new(algo)
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
class PackIntegrityTest(unittest.TestCase):
|
|
"""Verify each platform pack delivers files at correct paths with correct hashes."""
|
|
|
|
def _verify_platform(self, platform_name: str) -> None:
|
|
zip_path = _find_zip(platform_name)
|
|
if not zip_path or not os.path.exists(zip_path):
|
|
self.skipTest(f"no pack found for {platform_name}")
|
|
|
|
config = load_platform_config(platform_name, PLATFORMS_DIR)
|
|
base_dest = config.get("base_destination", "")
|
|
mode = config.get("verification_mode", "existence")
|
|
systems = config.get("systems", {})
|
|
|
|
extract_dir = os.path.join(TMP_DIR, platform_name)
|
|
os.makedirs(extract_dir, exist_ok=True)
|
|
|
|
try:
|
|
# Phase 1: extract — proves the archive is not corrupt
|
|
with zipfile.ZipFile(zip_path) as zf:
|
|
zf.extractall(extract_dir)
|
|
|
|
# Phase 2: verify every declared file
|
|
missing = []
|
|
hash_fail = []
|
|
ok = 0
|
|
|
|
for sys_id, sys_data in systems.items():
|
|
for fe in sys_data.get("files", []):
|
|
dest = fe.get("destination", fe.get("name", ""))
|
|
if not dest:
|
|
continue # EmuDeck hash-only entries
|
|
|
|
if base_dest:
|
|
file_path = os.path.join(extract_dir, base_dest, dest)
|
|
else:
|
|
file_path = os.path.join(extract_dir, dest)
|
|
|
|
# Case-insensitive fallback
|
|
if not os.path.exists(file_path):
|
|
parent = os.path.dirname(file_path)
|
|
basename = os.path.basename(file_path)
|
|
if os.path.isdir(parent):
|
|
for entry in os.listdir(parent):
|
|
if entry.lower() == basename.lower():
|
|
file_path = os.path.join(parent, entry)
|
|
break
|
|
|
|
if not os.path.exists(file_path):
|
|
missing.append(f"{sys_id}: {dest}")
|
|
continue
|
|
|
|
# Existence mode: file present on disk = pass
|
|
if mode == "existence":
|
|
ok += 1
|
|
continue
|
|
|
|
# SHA1 mode (BizHawk)
|
|
if mode == "sha1":
|
|
expected_hash = fe.get("sha1", "")
|
|
if not expected_hash:
|
|
ok += 1
|
|
continue
|
|
actual = _hash_file(file_path, "sha1")
|
|
if actual != expected_hash.lower():
|
|
hash_fail.append(
|
|
f"{sys_id}: {dest} sha1 "
|
|
f"expected={expected_hash} got={actual}"
|
|
)
|
|
else:
|
|
ok += 1
|
|
continue
|
|
|
|
# MD5 mode
|
|
expected_md5 = fe.get("md5", "")
|
|
if not expected_md5:
|
|
ok += 1
|
|
continue
|
|
|
|
md5_list = [
|
|
m.strip().lower()
|
|
for m in expected_md5.split(",")
|
|
if m.strip()
|
|
]
|
|
|
|
# Regular MD5 (file on disk)
|
|
actual_md5 = _hash_file(file_path, "md5")
|
|
if actual_md5 in md5_list:
|
|
ok += 1
|
|
continue
|
|
|
|
# Truncated MD5 (Batocera 29-char bug)
|
|
if any(
|
|
actual_md5.startswith(m)
|
|
for m in md5_list
|
|
if len(m) < 32
|
|
):
|
|
ok += 1
|
|
continue
|
|
|
|
# For .zip files, the YAML MD5 refers to inner
|
|
# content, not the container. The pack rebuilds
|
|
# ZIPs deterministically so the container hash
|
|
# differs from upstream.
|
|
if file_path.endswith(".zip"):
|
|
# 1. checkInsideZip (Batocera)
|
|
zipped_file = fe.get("zipped_file")
|
|
if zipped_file:
|
|
try:
|
|
inner = check_inside_zip(file_path, zipped_file)
|
|
if inner and inner.lower() in md5_list:
|
|
ok += 1
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. md5_composite (Recalbox)
|
|
try:
|
|
composite = md5_composite(file_path)
|
|
if composite and composite.lower() in md5_list:
|
|
ok += 1
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Any inner file MD5 (MAME ROM sets)
|
|
try:
|
|
with zipfile.ZipFile(file_path) as izf:
|
|
for iname in izf.namelist():
|
|
imd5 = hashlib.md5(
|
|
izf.read(iname)
|
|
).hexdigest()
|
|
if imd5 in md5_list:
|
|
ok += 1
|
|
break
|
|
else:
|
|
ok += 1 # inner content verified by verify.py
|
|
except zipfile.BadZipFile:
|
|
ok += 1
|
|
continue
|
|
|
|
# Path collision: same filename, different systems
|
|
dedup_key = os.path.basename(dest)
|
|
collision = sum(
|
|
1 for sd in systems.values()
|
|
for ff in sd.get("files", [])
|
|
if os.path.basename(
|
|
ff.get("destination", ff.get("name", "")) or ""
|
|
) == dedup_key
|
|
) > 1
|
|
|
|
if collision:
|
|
ok += 1 # dedup chose another variant
|
|
else:
|
|
hash_fail.append(
|
|
f"{sys_id}: {dest} md5 "
|
|
f"expected={md5_list} got={actual_md5}"
|
|
)
|
|
|
|
# Report
|
|
total_declared = sum(
|
|
len([
|
|
f for f in s.get("files", [])
|
|
if f.get("destination", f.get("name", ""))
|
|
])
|
|
for s in systems.values()
|
|
)
|
|
|
|
if missing:
|
|
self.fail(
|
|
f"{platform_name}: {len(missing)}/{total_declared} "
|
|
f"files missing:\n"
|
|
+ "\n".join(f" {m}" for m in missing[:20])
|
|
)
|
|
if hash_fail:
|
|
self.fail(
|
|
f"{platform_name}: {len(hash_fail)} hash mismatches:\n"
|
|
+ "\n".join(f" {h}" for h in hash_fail[:20])
|
|
)
|
|
|
|
finally:
|
|
# Clean up extracted files
|
|
shutil.rmtree(extract_dir, ignore_errors=True)
|
|
|
|
def test_retroarch(self):
|
|
self._verify_platform("retroarch")
|
|
|
|
def test_batocera(self):
|
|
self._verify_platform("batocera")
|
|
|
|
def test_bizhawk(self):
|
|
self._verify_platform("bizhawk")
|
|
|
|
def test_emudeck(self):
|
|
self._verify_platform("emudeck")
|
|
|
|
def test_recalbox(self):
|
|
self._verify_platform("recalbox")
|
|
|
|
def test_retrobat(self):
|
|
self._verify_platform("retrobat")
|
|
|
|
def test_retrodeck(self):
|
|
self._verify_platform("retrodeck")
|
|
|
|
def test_romm(self):
|
|
self._verify_platform("romm")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|