feat: test packs the way a user installs them

pack_probe.py lays a platform's BIOS folder out from its pack, or with install.py --dest, and starts each test game through core_probe with that folder, mapping every file the core opened to the pack entry that put it there.
This commit is contained in:
Abdessamad Derraz committed 2026-10-27 15:37:40 +01:00
1 parent c43872cdb6
commit f0320a2bd9
5 files changed
+1110 -2

No files matched your search

+129
View File
@@ -0,0 +1,129 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://abdess.github.io/retrobios/api/v1/schemas/pack-probe.schema.json",
"title": "RetroBIOS pack probe",
"description": "One line of measurements/pack-probes.jsonl: one run of scripts/pack_probe.py, a game started by a libretro core with its system directory taken from a platform's BIOS folder as a user installs it, the pack extracted or install.py run on a scratch root. The core_probe records the line names are in measurements/<emulator>.jsonl. Lines are appended, never rewritten.",
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "id", "probed_at", "platform", "platform_version", "target", "install",
"source", "system", "platform_system", "emulator", "core", "content", "records", "rounds",
"staged", "verdict", "verdict_evidence", "files_opened", "findings", "error"],
"$defs": {
"id": { "type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}-[0-9a-f]{12}$" },
"sha1": { "type": "string", "pattern": "^[0-9a-f]{40}$" },
"sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" },
"pack": {
"type": "object",
"additionalProperties": false,
"required": ["file", "sha256", "size", "members"],
"properties": {
"file": { "type": "string" },
"sha256": { "$ref": "#/$defs/sha256" },
"size": { "type": "integer", "minimum": 0 },
"members": { "type": ["integer", "null"], "minimum": 0 }
}
}
},
"properties": {
"schema_version": { "const": 1 },
"id": { "$ref": "#/$defs/id" },
"probed_at": { "type": "string", "format": "date-time" },
"platform": { "type": "string", "description": "Key of platforms/_registry.yml." },
"platform_version": { "type": "string" },
"target": { "type": ["string", "null"], "description": "Hardware target the cores were chosen for." },
"install": {
"enum": ["pack", "installer"],
"description": "pack: the platform's pack extracted at the root of the BIOS folder. installer: install.py --dest on the folder, from that pack and the published manifest, offline."
},
"source": {
"type": "object",
"additionalProperties": false,
"required": ["pack", "manifest", "repo_commit"],
"properties": {
"pack": { "$ref": "#/$defs/pack" },
"manifest": {
"type": ["object", "null"],
"additionalProperties": false,
"required": ["path", "sha256", "generated", "pack_files"],
"properties": {
"path": { "type": "string" },
"sha256": { "$ref": "#/$defs/sha256" },
"generated": { "type": ["string", "null"] },
"pack_files": { "type": ["integer", "null"] }
}
},
"save_dir_files": {
"type": "integer", "minimum": 0,
"description": "Files the installer put in the save directory, which the core's run does not see."
},
"installer_summary": { "type": "string" },
"repo_commit": { "type": ["string", "null"], "description": "Commit of the checkout the pack was built from." }
}
},
"system": { "type": "string" },
"platform_system": { "type": ["string", "null"], "description": "The platform's own id of the system." },
"emulator": { "type": "string", "description": "Profile key, the file under measurements/ the records are in." },
"core": {
"type": "object",
"additionalProperties": false,
"required": ["file", "sha256", "artifact", "last_modified"],
"properties": {
"file": { "type": "string" },
"sha256": { "$ref": "#/$defs/sha256" },
"artifact": { "type": "string" },
"last_modified": { "type": ["string", "null"] }
}
},
"content": {
"type": "object",
"additionalProperties": false,
"required": ["label", "catalog", "region"],
"properties": {
"label": { "type": "string" },
"catalog": { "type": "string" },
"region": { "type": ["string", "null"] }
}
},
"records": {
"type": "array",
"items": { "$ref": "#/$defs/id" },
"description": "core_probe records of this probe, first run first. A second one ran with the closed staging and supersedes the first."
},
"rounds": { "type": "integer", "minimum": 1, "description": "Runs until the core asked for nothing more the layout holds." },
"staged": { "type": "integer", "minimum": 0, "description": "Files of the layout staged for the last run." },
"verdict": { "enum": ["boots", "bios_error", "crash", "hang", "unknown"] },
"verdict_evidence": { "type": ["object", "null"] },
"files_opened": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["path", "entry", "sha1", "created"],
"properties": {
"path": { "type": "string", "description": "Path under the system directory." },
"entry": { "type": ["string", "null"], "description": "The pack member, or the manifest entry, that put the file there." },
"sha1": { "anyOf": [{ "$ref": "#/$defs/sha1" }, { "type": "null" }] },
"created": { "type": "boolean", "description": "Written by the core, not taken from the layout." }
}
}
},
"findings": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["kind", "path", "declared_by", "paths"],
"properties": {
"kind": {
"enum": ["missing_declared", "missing_undeclared", "elsewhere", "not_staged", "unlisted"],
"description": "missing_declared: the core looked for a file its profile declares, the layout lacks it. missing_undeclared: the same for a file no entry of the profile names. elsewhere: the layout holds a file of that name under another path. not_staged: the layout holds it but the staging never closed. unlisted: the core opened a file of the layout that no pack or manifest entry names."
},
"path": { "type": "string" },
"declared_by": { "type": "array", "items": { "type": "string" } },
"paths": { "type": "array", "items": { "type": "string" } }
}
}
},
"error": { "type": ["string", "null"], "description": "Why no record was written, when core_probe failed." }
}
}
+4 -2
View File
@@ -84,13 +84,15 @@ def records_path(emulator: str) -> str:
# The campaign's own files beside the records: append-only like them, one
# line per finding or per pass, each line hashed to its id. No profile takes
# these names (tests/test_campaign.py checks it).
# line per finding, per pass or per pack probe, each line hashed to its id.
# No profile takes these names (tests/test_campaign.py checks it).
FINDINGS = "findings"
CAMPAIGN_RUNS = "campaign-runs"
PACK_PROBES = "pack-probes"
CAMPAIGN_FILES = {
FINDINGS: ("found_at", "campaign-finding.schema.json"),
CAMPAIGN_RUNS: ("started_at", "campaign-run.schema.json"),
PACK_PROBES: ("probed_at", "pack-probe.schema.json"),
}
+665
View File
@@ -0,0 +1,665 @@
#!/usr/bin/env python3
"""Test a platform's pack the way a user installs it.
For one platform, the probe lays the platform's BIOS folder out in a scratch
root, in one or both of the ways a user gets it:
pack the pack generate_pack.py builds for the platform, extracted
installer install.py --dest on the scratch root, from that pack and the
published manifest (install/<platform>.json), with no network
Then for each system a test game covers (tests/campaign/content.yml and
tests/campaign/pack_content.yml), it resolves the libretro cores the platform
runs for that system (platform config, target file, standalone and dual-mode
lists), and runs the game through core_probe.py with the system directory
taken from that layout.
core_probe copies what it stages, so the probe stages the files the core can
reach rather than the whole folder: the files the core's profile and the
platform's list declare, every file beside them in their directory, and for
a core that finds its BIOS by content, every root file of a size it accepts.
A file the core then looks for that the layout holds is staged and the run
repeated, until the core asks for nothing more the layout has.
Each run is a core_probe record in measurements/<emulator>.jsonl and one line
of measurements/pack-probes.jsonl naming the platform, the pack or manifest,
the verdict, every file the core opened mapped to the pack entry that put it
there, and the findings: a file the core looked for that the layout lacks,
or holds under another path.
Usage:
flock ~/.local/state/retrobios/probe.lock \\
python scripts/pack_probe.py run --platform retroarch [--system sony-playstation]
python scripts/pack_probe.py plan --platform batocera
"""
from __future__ import annotations
import argparse
import datetime as dt
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import time
import urllib.request
import zipfile
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import autorun
import campaign
import measurement_records as mr
from common import (
_norm_system_id,
load_emulator_profiles,
load_platform_config,
load_target_config,
resolve_platform_cores,
runs_standalone,
)
from validation import filter_files_by_mode
from ziptools import safe_extract_zip
ROOT = Path(__file__).resolve().parent.parent
MAIN = autorun.main_checkout()
PACK_CONTENT_FILE = ROOT / "tests" / "campaign" / "pack_content.yml"
CONTENT_DIR = MAIN / ".cache" / "test-content"
CORE_CACHE = MAIN / ".cache" / "pack-probe" / "cores"
WORK_DIR = ROOT / "tmp" / "pack_probe"
BUILDBOT = campaign.BUILDBOT
PACK_PROBES = mr.PACK_PROBES
INSTALLS = ("pack", "installer")
VERDICTS = ("boots", "bios_error", "crash", "hang", "unknown")
MAX_ROUNDS = 6
PROBE_TIMEOUT = 900
IDLE_WAIT = 1800
EXIT_BUSY = 3
# The hardware target each platform is probed for: the one whose libretro
# cores are the buildbot's x86_64 Linux builds, or the platform's only one.
TARGETS = {"retroarch": "linux-x86_64", "batocera": "x86_64", "retrobat": "windows"}
class ProbeFailure(RuntimeError):
"""The probe cannot go on: no pack, no installer layout, a busy host."""
# --------------------------------------------------------------------------
# Layouts: the platform's BIOS folder as a user ends up with it
# --------------------------------------------------------------------------
@dataclass
class Layout:
"""A BIOS folder: every file by its path under the folder, and the pack
or manifest entry that put it there (None for a file nothing declares)."""
install: str
root: Path
files: dict[str, str | None]
source: dict
sizes: dict[str, int] = field(default_factory=dict)
def size(self, rel: str) -> int:
if rel not in self.sizes:
self.sizes[rel] = (self.root / rel).stat().st_size
return self.sizes[rel]
def walk(root: Path) -> list[str]:
return sorted(p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file())
def file_sha(path: Path, algorithm: str = "sha1") -> str:
digest = hashlib.new(algorithm)
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1 << 20), b""):
digest.update(chunk)
return digest.hexdigest()
def pack_layout(pack: Path, root: Path, identity: dict) -> Layout:
"""The pack extracted as a user does, at the root of the BIOS folder.
Each file's entry is the archive member it came from."""
if root.exists():
shutil.rmtree(root)
safe_extract_zip(str(pack), str(root))
with zipfile.ZipFile(pack) as handle:
members = {info.filename for info in handle.infolist() if not info.is_dir()}
files = {rel: rel if rel in members else None for rel in walk(root)}
return Layout("pack", root, files, {"pack": identity, "manifest": None})
def pack_identity(pack: Path) -> dict:
with zipfile.ZipFile(pack) as handle:
members = sum(1 for info in handle.infolist() if not info.is_dir())
return {"file": pack.name, "sha256": file_sha(pack, "sha256"), "size": pack.stat().st_size,
"members": members}
def manifest_entries(manifest: dict) -> dict[str, str]:
"""Destination -> the manifest entry that names it, as the installer
places it under --dest: a file by its dest, a data directory member
under its directory's dest."""
out: dict[str, str] = {}
for item in manifest.get("files") or ():
out[item["dest"]] = f"files:{item['dest']}"
for item in manifest.get("case_variants") or ():
out.setdefault(item["dest"], f"case_variants:{item['dest']}")
for directory in manifest.get("data_directories") or ():
base = str(directory.get("dest") or "").strip("/")
for member in directory.get("files") or {}:
dest = f"{base}/{member}" if base else member
out.setdefault(dest, f"data_directories:{directory['ref']}:{member}")
return out
def installer_layout(platform: str, pack: Path, identity: dict, manifest_path: Path,
root: Path, narrow: Sequence[str] = ()) -> Layout:
"""install.py --dest on a scratch root, from the pack, offline. The
manifest is the published one, the file a user's installer reads.
``narrow`` passes install.py options such as --system, for the tests."""
system_dir, save_dir = root / "system", root / "saves"
if root.exists():
shutil.rmtree(root)
argv = [sys.executable, str(ROOT / "install.py"), "--platform", platform, "--dest", str(system_dir),
"--save-dir", str(save_dir), "--from-pack", str(pack), "--manifest", str(manifest_path),
"--offline", "--yes", "--quiet", *narrow]
proc = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=ROOT)
if proc.returncode != 0 or not system_dir.is_dir():
raise ProbeFailure(f"install.py exited {proc.returncode}: {(proc.stderr or proc.stdout)[-1500:]}")
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
entries = manifest_entries(manifest)
files = {rel: entries.get(rel) for rel in walk(system_dir)}
source = {
"pack": identity,
"manifest": {"path": manifest_path.relative_to(ROOT).as_posix() if manifest_path.is_relative_to(ROOT)
else manifest_path.name,
"sha256": file_sha(manifest_path, "sha256"),
"generated": manifest.get("generated"), "pack_files": manifest.get("pack_files")},
"save_dir_files": len(walk(save_dir)) if save_dir.is_dir() else 0,
"installer_summary": " ".join(proc.stdout.split())[-400:],
}
return Layout("installer", system_dir, files, source)
def build_pack(platform: str, out_dir: Path, narrow: Sequence[str] = ()) -> Path:
"""The platform's pack, built from this checkout as a release builds it.
``narrow`` passes generate_pack.py options such as --system, for the tests."""
out_dir.mkdir(parents=True, exist_ok=True)
for old in out_dir.glob("*_BIOS_Pack*.zip"):
old.unlink()
argv = [sys.executable, str(ROOT / "scripts" / "generate_pack.py"), "--platform", platform,
"--offline", "--output-dir", str(out_dir), *narrow]
proc = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=ROOT)
packs = sorted(p for p in out_dir.glob("*_BIOS_Pack*.zip") if ".part" not in p.name)
if proc.returncode != 0 or len(packs) != 1:
raise ProbeFailure(f"generate_pack.py exited {proc.returncode} with {len(packs)} packs: "
f"{proc.stderr[-1500:]}")
return packs[0]
# --------------------------------------------------------------------------
# Which cores a platform runs for a system
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class CoreChoice:
emulator: str
so_name: str
def platform_system(config: dict, system: str) -> str | None:
"""The platform's own id of a system, compared on the normalised id."""
wanted = _norm_system_id(system)
matches = sorted(s for s in config.get("systems") or {} if _norm_system_id(s) == wanted)
return matches[0] if matches else None
def libretro_build_runs(name: str, profile: dict, config: dict) -> bool:
"""The platform launches this profile's libretro build: it has one, and
the platform does not run the emulator standalone only."""
if "libretro" not in str(profile.get("type", "")) or profile.get("type") == "alias":
return False
standalone = {str(c) for c in config.get("standalone_cores") or ()}
if not runs_standalone(name, profile, standalone):
return True
dual = {str(c) for c in config.get("dual_mode_cores") or ()}
return bool(dual & ({name} | {str(c) for c in profile.get("cores") or ()}))
def so_names(name: str, profile: dict, buildbot: set[str], target: set[str] | None) -> list[str]:
"""The buildbot builds of a profile the platform runs. Every one the
target names, else the build that carries the profile's key, else the
first of its cores: that is on the buildbot."""
names = [str(c) for c in profile.get("cores") or () if str(c) in buildbot]
if target is not None:
named = [n for n in names if n in target]
if named:
return sorted(named)
if name in buildbot:
return [name]
return names[:1]
def choose_cores(config: dict, profiles: dict[str, dict], system: str, buildbot: set[str],
target: set[str] | None) -> list[CoreChoice]:
wanted = _norm_system_id(system)
out = []
for name in sorted(resolve_platform_cores(config, profiles, target)):
profile = profiles[name]
if wanted not in {_norm_system_id(s) for s in profile.get("systems") or ()}:
continue
if not libretro_build_runs(name, profile, config):
continue
out.extend(CoreChoice(name, so) for so in so_names(name, profile, buildbot, target))
return out
# --------------------------------------------------------------------------
# Staging: what the core can reach, grown to a fixed point
# --------------------------------------------------------------------------
def entry_paths(entry: dict) -> list[str]:
"""The paths a profile entry names under the system directory: its
path or name, and its aliases beside it."""
main = campaign.entry_dest(entry)
folder = main.rsplit("/", 1)[0] + "/" if "/" in main else ""
return [main] + [f"{folder}{alias}" for alias in entry.get("aliases") or ()]
def libretro_entries(profile: dict) -> list[dict]:
return filter_files_by_mode(profile.get("files") or [], standalone=False)
def declared_sizes(profile: dict) -> set[int]:
sizes: set[int] = set()
for entry in libretro_entries(profile):
value = entry.get("size")
sizes.update(v for v in (value if isinstance(value, list) else [value]) if isinstance(v, int))
return sizes
def with_folders(files: Iterable[str], staged: set[str]) -> set[str]:
"""Every file of every folder a staged file sits in, below the root: a
core that lists its folder sees what the platform's folder holds."""
folders = {rel.rsplit("/", 1)[0] + "/" for rel in staged if "/" in rel}
return set(staged) | {rel for rel in files if any(rel.startswith(f) for f in folders)}
def initial_stage(layout: Layout, profile: dict, platform_dests: Iterable[str]) -> set[str]:
files = set(layout.files)
staged = {p for entry in libretro_entries(profile) for p in entry_paths(entry) if p in files}
staged |= {d for d in platform_dests if d in files}
if profile.get("found_by") == "content":
sizes = declared_sizes(profile)
staged |= {rel for rel in files if "/" not in rel and layout.size(rel) in sizes}
return with_folders(files, staged)
def grow_stage(files: Iterable[str], staged: set[str], probed: Iterable[str]) -> set[str]:
"""The files the layout holds that the core looked for, as a file or as
a folder, and their folders: empty once the staging is closed."""
files = set(files)
wanted = set()
for path in probed:
if path in files:
wanted.add(path)
else:
wanted |= {rel for rel in files if rel.startswith(path.rstrip("/") + "/")}
return with_folders(files, staged | wanted) - staged
# --------------------------------------------------------------------------
# Reading a run
# --------------------------------------------------------------------------
def run_verdict(record: dict | None) -> tuple[str, dict | None]:
"""The verdict of a run record: variants[0] from record version 4,
unknown before the oracle or without a record."""
if not record or not record.get("variants"):
return "unknown", None
variant = record["variants"][0]
verdict = variant.get("verdict")
if verdict not in VERDICTS:
return "unknown", None
return verdict, variant.get("verdict_evidence")
def first_repetition(record: dict) -> dict:
return record["variants"][0]["repetitions"][0]
def declaring(profile: dict, path: str) -> list[str]:
return sorted({entry["name"] for entry in libretro_entries(profile) if path in entry_paths(entry)})
def map_run(layout: Layout, staged: set[str], rep: dict, profile: dict) -> tuple[list[dict], list[dict]]:
"""The files the core opened, each with its pack entry, and the findings."""
opened = []
for item in rep.get("files_opened") or ():
path = item["path"]
opened.append({"path": path, "entry": layout.files.get(path) if path in staged else None,
"sha1": item.get("sha1_after"), "created": path not in staged})
by_name: dict[str, list[str]] = {}
for rel in layout.files:
by_name.setdefault(rel.rsplit("/", 1)[-1].casefold(), []).append(rel)
findings = []
for path in rep.get("files_probed_absent") or ():
names = declaring(profile, path)
if path in layout.files:
findings.append({"kind": "not_staged", "path": path, "declared_by": names, "paths": [path]})
continue
elsewhere = sorted(p for p in by_name.get(path.rsplit("/", 1)[-1].casefold(), ()) if p != path)
if elsewhere:
kind = "elsewhere"
else:
kind = "missing_declared" if names else "missing_undeclared"
findings.append({"kind": kind, "path": path, "declared_by": names, "paths": elsewhere})
for item in opened:
if item["path"] in staged and item["entry"] is None:
findings.append({"kind": "unlisted", "path": item["path"], "declared_by": [], "paths": []})
return opened, findings
# --------------------------------------------------------------------------
# Cores and content
# --------------------------------------------------------------------------
def buildbot_listing() -> set[str]:
with urllib.request.urlopen(BUILDBOT, timeout=120) as response:
page = response.read().decode("utf-8", "replace")
return set(re.findall(r"([A-Za-z0-9_]+)_libretro\.so\.zip", page))
def fetch_core(name: str, refresh: bool = False) -> dict:
"""The buildbot nightly of a core, kept with where and when it came from."""
folder = CORE_CACHE / name
so, meta_path = folder / f"{name}_libretro.so", folder / "build.json"
if so.is_file() and meta_path.is_file() and not refresh:
meta = json.loads(meta_path.read_text(encoding="utf-8"))
if meta.get("sha256") == file_sha(so, "sha256"):
return {**meta, "path": str(so)}
url = f"{BUILDBOT}{name}_libretro.so.zip"
folder.mkdir(parents=True, exist_ok=True)
archive = folder / f"{name}.zip.part"
with urllib.request.urlopen(url, timeout=600) as response, archive.open("wb") as out:
shutil.copyfileobj(response, out)
modified = response.headers.get("Last-Modified")
campaign.safe_extract_so(archive, f"{name}_libretro.so", so)
archive.unlink()
meta = {"name": name, "url": url, "last_modified": modified, "sha256": file_sha(so, "sha256"),
"fetched_at": now_iso()}
meta_path.write_text(json.dumps(meta, indent=1) + "\n", encoding="utf-8")
return {**meta, "path": str(so)}
def load_contents(files: Sequence[Path]) -> dict[str, list[campaign.Content]]:
out: dict[str, list[campaign.Content]] = {}
for path in files:
if path.is_file():
for system, items in campaign.load_content(path).items():
out.setdefault(system, []).extend(items)
return out
def content_folder(content: campaign.Content, fetch: bool) -> Path:
"""The game's folder, fetched through romfetch when absent and asked to."""
folder = CONTENT_DIR / content.romfetch_system / content.name
if not (folder / "identity.json").is_file() and fetch:
romfetch = shutil.which("romfetch")
if not romfetch:
raise campaign.Unavailable("romfetch not found")
proc = subprocess.run([romfetch, "get", content.romfetch_system, content.name, "--json", "-q",
"--content-dir", str(CONTENT_DIR)], capture_output=True, text=True,
check=False, env=campaign.child_env(romfetch=True))
if proc.returncode != 0:
raise campaign.Unavailable(f"romfetch exit {proc.returncode} for {content.name!r}")
campaign.verify_identity(folder)
return folder
# --------------------------------------------------------------------------
# One probe: a core, a game, a layout
# --------------------------------------------------------------------------
def now_iso() -> str:
return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
def wait_idle(limit: float = IDLE_WAIT, poll: float = 30.0) -> None:
"""Wait for the load core_probe records under, a quarter of the CPUs.
Extracting a pack loads the host itself for a minute or two."""
threshold = (os.cpu_count() or 1) / 4
waited = 0.0
while os.getloadavg()[0] > threshold:
if waited >= limit:
raise ProbeFailure(f"busy host: load {os.getloadavg()[0]:.2f} above {threshold} "
f"for {int(waited)} s")
time.sleep(poll)
waited += poll
def core_probe(argv: list[str], staged: set[str], layout: Layout, work: Path) -> tuple[int, dict | None, str]:
"""core_probe.py run, the system files passed in an argument file."""
if "--record" in argv:
wait_idle()
work.mkdir(parents=True, exist_ok=True)
argfile = work / "system-files.args"
argfile.write_text("".join(f"--system-file\n{rel}={layout.root / rel}\n" for rel in sorted(staged)),
encoding="utf-8")
command = [sys.executable, str(ROOT / "scripts" / "core_probe.py"), *argv, f"@{argfile}"]
proc = subprocess.run(command, capture_output=True, text=True, check=False, cwd=ROOT,
timeout=PROBE_TIMEOUT * 2 + 120)
record = None
if proc.returncode == 0:
try:
record = json.loads(proc.stdout)
except ValueError:
record = None
return proc.returncode, record, proc.stderr.strip()[-1500:]
def probe_argv(core: dict, choice: CoreChoice, content: campaign.Content, load: Path,
subject: str, record: bool, supersedes: str | None = None) -> list[str]:
argv = ["run", "--core", core["path"], "--content", str(load), "--content-label", content.name,
"--frames", str(content.frames), "--frame-hash-every", str(content.frame_hash_every),
"--require-sandbox", "--timeout", str(PROBE_TIMEOUT),
"--work-dir", str(ROOT / "tmp" / "core_probe"), "--core-artifact", core["url"]]
if record:
argv += ["--record", str(ROOT), "--emulator", choice.emulator, "--subject", subject]
if supersedes:
argv += ["--supersedes", supersedes]
return argv
def subject_for(platform: str, install: str, system: str, so: str, content: str) -> str:
return mr.slugify(f"pack {platform} {install} {system} {so} {content}", 90)
def probe_one(layout: Layout, platform: str, config: dict, profile: dict, choice: CoreChoice,
core: dict, system: str, content: campaign.Content, load: Path, work: Path) -> dict:
"""Run until the staging is closed. The first run is recorded, and when
the staging grew, the closed one is recorded too and supersedes it."""
native = platform_system(config, system)
dests = [f.get("destination") or f.get("name")
for f in ((config.get("systems") or {}).get(native) or {}).get("files") or ()]
staged = initial_stage(layout, profile, dests)
subject = subject_for(platform, layout.install, system, choice.so_name, content.name)
code, record, error = core_probe(probe_argv(core, choice, content, load, subject, True),
staged, layout, work)
if code == EXIT_BUSY:
raise ProbeFailure(f"core_probe refused on a busy host: {error}")
rounds, first_id, records = 1, None, []
if record:
first_id = record["id"]
records.append(first_id)
added = grow_stage(layout.files, staged, first_repetition(record).get("files_probed_absent") or ())
while added and rounds < MAX_ROUNDS:
staged |= added
rounds += 1
code, record, error = core_probe(probe_argv(core, choice, content, load, subject, False),
staged, layout, work)
if not record:
break
added = grow_stage(layout.files, staged, first_repetition(record).get("files_probed_absent") or ())
if rounds > 1 and record:
code, record, error = core_probe(probe_argv(core, choice, content, load, subject, True, first_id),
staged, layout, work)
if code == EXIT_BUSY:
raise ProbeFailure(f"core_probe refused on a busy host: {error}")
if record:
records.append(record["id"])
verdict, evidence = run_verdict(record)
opened, findings = map_run(layout, staged, first_repetition(record), profile) if record else ([], [])
return {
"schema_version": 1,
"probed_at": now_iso(),
"platform": platform,
"platform_version": str(config.get("version") or ""),
"target": TARGETS.get(platform),
"install": layout.install,
"source": {**layout.source, "repo_commit": git_head()},
"system": system,
"platform_system": native,
"emulator": choice.emulator,
"core": {"file": f"{choice.so_name}_libretro.so", "sha256": core["sha256"], "artifact": core["url"],
"last_modified": core.get("last_modified")},
"content": {"label": content.name, "catalog": content.catalog, "region": content.region},
"records": records,
"rounds": rounds,
"staged": len(staged),
"verdict": verdict,
"verdict_evidence": evidence,
"files_opened": opened,
"findings": findings,
"error": None if record else (error or f"core_probe exited {code}"),
}
def git_head() -> str | None:
proc = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=False,
cwd=ROOT)
return proc.stdout.strip() or None
def append_line(line: dict, root: Path = ROOT) -> dict:
sealed = mr.seal_line(line, mr.CAMPAIGN_FILES[PACK_PROBES][0])
path = root / mr.records_path(PACK_PROBES)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as handle:
handle.write(mr.line_text(sealed) + "\n")
return sealed
# --------------------------------------------------------------------------
# Plans and the command line
# --------------------------------------------------------------------------
@dataclass
class Unit:
system: str
content: campaign.Content
cores: list[CoreChoice]
def plan(platform: str, systems: set[str] | None, cores: set[str] | None,
buildbot: set[str]) -> tuple[dict, dict, list[Unit]]:
config = load_platform_config(platform, str(ROOT / "platforms"))
profiles = load_emulator_profiles(str(ROOT / "emulators"))
target = load_target_config(platform, TARGETS[platform], str(ROOT / "platforms"))
units = []
for system, contents in sorted(load_contents([campaign.CONTENT_FILE, PACK_CONTENT_FILE]).items()):
if systems and system not in systems:
continue
if not platform_system(config, system):
continue
choices = [c for c in choose_cores(config, profiles, system, buildbot, target)
if not cores or c.so_name in cores or c.emulator in cores]
units.extend(Unit(system, content, choices) for content in contents)
return config, profiles, units
def run(args: argparse.Namespace) -> int:
buildbot = buildbot_listing()
config, profiles, units = plan(args.platform, args.systems, args.cores, buildbot)
if args.mode == "plan":
for unit in units:
print(f"{unit.system}: {unit.content.name}: "
f"{', '.join(f'{c.so_name} ({c.emulator})' for c in unit.cores) or 'no core'}")
return 0
if not units:
print(f"ERROR: no test game for any system of {args.platform}", file=sys.stderr)
return 2
work = Path(args.work_dir) / args.platform
pack = Path(args.pack) if args.pack else build_pack(args.platform, work / "dist")
identity = pack_identity(pack)
status = 0
for install in args.install:
if install == "pack":
layout = pack_layout(pack, work / "pack", identity)
else:
layout = installer_layout(args.platform, pack, identity,
ROOT / "install" / f"{args.platform}.json", work / "installer")
try:
for unit in units:
try:
load = campaign.load_file(content_folder(unit.content, args.fetch), unit.content.load)
except campaign.Unavailable as exc:
print(f"SKIP {unit.system}: {exc}", file=sys.stderr)
continue
for choice in unit.cores:
core = fetch_core(choice.so_name, args.refresh_cores)
line = probe_one(layout, args.platform, config, profiles[choice.emulator], choice,
core, unit.system, unit.content, load, work / "runs")
sealed = append_line(line)
print(f"{args.platform} {install} {unit.system} {choice.so_name}: {line['verdict']}, "
f"{len(line['files_opened'])} opened, {len(line['findings'])} findings, "
f"line {sealed['id']}")
if line["error"]:
status = 1
finally:
if not args.keep:
shutil.rmtree(layout.root if install == "pack" else layout.root.parent, ignore_errors=True)
return status
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("mode", choices=("run", "plan"))
parser.add_argument("--platform", required=True, choices=sorted(TARGETS))
parser.add_argument("--system", dest="systems", action="append", default=[],
help="only this system (repeat)")
parser.add_argument("--core", dest="cores", action="append", default=[],
help="only this core, by buildbot name or profile (repeat)")
parser.add_argument("--install", action="append", choices=INSTALLS, default=[],
help="pack, installer or both (default both)")
parser.add_argument("--pack", help="a pack already built for the platform, instead of building one")
parser.add_argument("--fetch", action="store_true", help="fetch a missing game through romfetch")
parser.add_argument("--refresh-cores", action="store_true", help="download the nightlies again")
parser.add_argument("--keep", action="store_true", help="keep the extracted layouts")
parser.add_argument("--work-dir", default=str(WORK_DIR))
return parser
def main(argv: Sequence[str] | None = None) -> int:
args = build_parser().parse_args(argv)
args.systems = set(args.systems) or None
args.cores = set(args.cores) or None
args.install = args.install or list(INSTALLS)
try:
return run(args)
except ProbeFailure as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return EXIT_BUSY if "busy host" in str(exc) else 1
if __name__ == "__main__":
sys.exit(main())
+32
View File
@@ -0,0 +1,32 @@
# Test content of the pack probe (scripts/pack_probe.py), beside the
# campaign's own list (content.yml), in the same format and read the same way.
# A game here starts a system whose cores need a BIOS from the pack.
systems:
sega-saturn:
romfetch: saturn
content:
- name: "Lunar - Silver Star Story Complete (Japan)"
catalog: redump
reason: >-
A Japanese disc. Every Saturn core needs a BIOS image to start a disc,
so the run shows whether the BIOS the pack put in the folder is one
the core reads, under the name it looks for.
region: japan
load: cue
frames: 3600
frame_hash_every: 10
pinned: true
sega-mega-cd:
romfetch: segacd
content:
- name: "After Burner III (USA)"
catalog: redump
reason: >-
A US disc. A Mega CD core runs the console's boot ROM before the disc,
and the BIOS it reads for a US disc is the Sega CD one.
region: north-america
load: cue
frames: 3600
frame_hash_every: 10
pinned: true
+280
View File
@@ -0,0 +1,280 @@
"""pack_probe.py tests a platform's pack the way a user installs it.
The rules (which cores a platform runs, what is staged, how a run maps back
to the pack, the verdict read from a record, the line written) are checked
on synthetic data. Two classes lay a real BIOS folder out from this
checkout, one per install path: the pack generate_pack.py builds, extracted,
and install.py --dest from that pack and the published manifest. They are
narrowed to one system to stay short, and skip when the collection is not
on disk.
"""
from __future__ import annotations
import contextlib
import io
import json
import shutil
import unittest
import zipfile
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
import sys # noqa: E402
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import measurement_records as mr # noqa: E402
import pack_probe as pp # noqa: E402
TMP = REPO_ROOT / "tmp" / "tests" / "pack_probe"
SATURN = {
"type": "libretro",
"cores": ["mednafen_saturn", "beetle-saturn"],
"systems": ["sega-saturn"],
"files": [
{"name": "sega_101.bin", "region": ["japan"]},
{"name": "mpr-17933.bin", "aliases": ["saturn_bios.bin"]},
{"name": "standalone.bin", "mode": "standalone"},
],
}
SCANNER = {
"type": "libretro",
"cores": ["swanstation"],
"systems": ["sony-playstation"],
"found_by": "content",
"files": [{"name": "scph5501.bin", "size": 524288},
{"name": "ps2.bin", "size": [524288, 4194304]}],
}
KRONOS = {
"type": "libretro",
"cores": ["kronos"],
"systems": ["sega-saturn"],
"files": [{"name": "saturn_bios.bin", "path": "kronos/saturn_bios.bin"}],
}
def layout(files: dict[str, int], entries: dict[str, str | None] | None = None) -> pp.Layout:
"""A layout whose sizes are given, nothing on disk."""
return pp.Layout("pack", Path("/nonexistent"), entries or {rel: rel for rel in files}, {},
sizes=dict(files))
class CoreChoice(unittest.TestCase):
def test_system_ids_compare_normalised(self) -> None:
config = {"systems": {"megacd": {}, "psx": {}}}
self.assertEqual(pp.platform_system(config, "sega-mega-cd"), "megacd")
self.assertIsNone(pp.platform_system(config, "sega-saturn"))
def test_standalone_only_is_not_run_as_a_core(self) -> None:
both = {"type": "standalone + libretro", "cores": ["duckstation", "duckstation-sa"]}
self.assertTrue(pp.libretro_build_runs("duckstation", both, {}))
self.assertFalse(pp.libretro_build_runs("duckstation", both, {"standalone_cores": ["duckstation"]}))
self.assertTrue(pp.libretro_build_runs("duckstation", both, {
"standalone_cores": ["duckstation"], "dual_mode_cores": ["duckstation-sa"]}))
self.assertFalse(pp.libretro_build_runs("x", {"type": "standalone"}, {}))
self.assertFalse(pp.libretro_build_runs("x", {"type": "alias"}, {}))
def test_buildbot_name(self) -> None:
buildbot = {"mednafen_psx", "mednafen_psx_hw", "kronos"}
beetle = {"cores": ["beetle_psx", "mednafen_psx", "mednafen_psx_hw"]}
# The target names both builds: both run.
self.assertEqual(pp.so_names("beetle_psx", beetle, buildbot, {"mednafen_psx", "mednafen_psx_hw"}),
["mednafen_psx", "mednafen_psx_hw"])
# A target in the platform's own names: the first buildbot name.
self.assertEqual(pp.so_names("beetle_psx", beetle, buildbot, {"beetle-psx"}), ["mednafen_psx"])
self.assertEqual(pp.so_names("kronos", {"cores": ["kronos"]}, buildbot, None), ["kronos"])
self.assertEqual(pp.so_names("gone", {"cores": ["gone"]}, buildbot, None), [])
def test_choose_cores(self) -> None:
profiles = {"beetle_saturn": SATURN, "kronos": KRONOS, "swanstation": SCANNER,
"ssf": {"type": "standalone", "cores": ["ssf"], "systems": ["sega-saturn"]}}
config = {"_key": "p", "cores": ["beetle-saturn", "kronos", "ssf", "swanstation"],
"systems": {"saturn": {}}}
buildbot = {"mednafen_saturn", "kronos", "swanstation"}
self.assertEqual(pp.choose_cores(config, profiles, "sega-saturn", buildbot, None),
[pp.CoreChoice("beetle_saturn", "mednafen_saturn"), pp.CoreChoice("kronos", "kronos")])
class Staging(unittest.TestCase):
FILES = {"sega_101.bin": 524288, "saturn_bios.bin": 524288, "scph5501.bin": 524288,
"big.bin": 4194304, "other.rom": 1000, "kronos/saturn_bios.bin": 524288,
"kronos/stvbios.zip": 10, "mame/hash/saturn.xml": 5, "standalone.bin": 1}
def test_declared_names_aliases_and_platform_list(self) -> None:
staged = pp.initial_stage(layout(self.FILES), SATURN, ["other.rom", "absent.bin"])
self.assertEqual(staged, {"sega_101.bin", "saturn_bios.bin", "other.rom"})
def test_a_folder_comes_whole(self) -> None:
staged = pp.initial_stage(layout(self.FILES), KRONOS, [])
self.assertEqual(staged, {"kronos/saturn_bios.bin", "kronos/stvbios.zip"})
def test_a_core_that_scans_gets_every_root_file_of_its_sizes(self) -> None:
staged = pp.initial_stage(layout(self.FILES), SCANNER, [])
self.assertEqual(staged, {"sega_101.bin", "saturn_bios.bin", "scph5501.bin", "big.bin"})
def test_grow_to_a_fixed_point(self) -> None:
staged = {"sega_101.bin"}
added = pp.grow_stage(self.FILES, staged, ["other.rom", "absent.bin", "mame/hash", "kronos"])
self.assertEqual(added, {"other.rom", "mame/hash/saturn.xml", "kronos/saturn_bios.bin",
"kronos/stvbios.zip"})
self.assertEqual(pp.grow_stage(self.FILES, staged | added, ["absent.bin"]), set())
class Mapping(unittest.TestCase):
def test_opened_files_map_to_their_entry_and_probes_to_findings(self) -> None:
lay = layout({"sega_101.bin": 1, "BIOS/Saturn_BIOS.bin": 1, "extra.bin": 1},
{"sega_101.bin": "sega_101.bin", "BIOS/Saturn_BIOS.bin": "BIOS/Saturn_BIOS.bin",
"extra.bin": None})
rep = {"files_opened": [{"path": "sega_101.bin", "sha1_after": "a" * 40},
{"path": "extra.bin", "sha1_after": "b" * 40},
{"path": "bkram.bin", "sha1_after": "c" * 40}],
"files_probed_absent": ["saturn_bios.bin", "mpr-17933.bin", "cfg.ini"]}
opened, findings = pp.map_run(lay, {"sega_101.bin", "extra.bin"}, rep, SATURN)
self.assertEqual(opened, [
{"path": "sega_101.bin", "entry": "sega_101.bin", "sha1": "a" * 40, "created": False},
{"path": "extra.bin", "entry": None, "sha1": "b" * 40, "created": False},
{"path": "bkram.bin", "entry": None, "sha1": "c" * 40, "created": True}])
self.assertEqual(findings, [
{"kind": "elsewhere", "path": "saturn_bios.bin", "declared_by": ["mpr-17933.bin"],
"paths": ["BIOS/Saturn_BIOS.bin"]},
{"kind": "missing_declared", "path": "mpr-17933.bin", "declared_by": ["mpr-17933.bin"], "paths": []},
{"kind": "missing_undeclared", "path": "cfg.ini", "declared_by": [], "paths": []},
{"kind": "unlisted", "path": "extra.bin", "declared_by": [], "paths": []}])
def test_manifest_entries(self) -> None:
manifest = {"files": [{"dest": "a.bin"}],
"case_variants": [{"dest": "A.BIN"}, {"dest": "a.bin"}],
"data_directories": [{"ref": "bluemsx", "dest": "", "files": {"Machines/x.ini": {}}},
{"ref": "dolphin", "dest": "dolphin-emu/Sys/", "files": {"y": {}}}]}
self.assertEqual(pp.manifest_entries(manifest), {
"a.bin": "files:a.bin", "A.BIN": "case_variants:A.BIN",
"Machines/x.ini": "data_directories:bluemsx:Machines/x.ini",
"dolphin-emu/Sys/y": "data_directories:dolphin:y"})
class Verdict(unittest.TestCase):
def test_a_record_before_the_oracle_is_unknown(self) -> None:
self.assertEqual(pp.run_verdict(None), ("unknown", None))
self.assertEqual(pp.run_verdict({"variants": [{"repetitions": [{}]}]}), ("unknown", None))
def test_the_first_variant_decides(self) -> None:
evidence = {"rule": "reference_frame", "detail": "frame 120"}
record = {"variants": [{"verdict": "boots", "verdict_evidence": evidence, "repetitions": [{}]}]}
self.assertEqual(pp.run_verdict(record), ("boots", evidence))
self.assertEqual(pp.run_verdict({"variants": [{"verdict": "maybe"}]}), ("unknown", None))
class Line(unittest.TestCase):
def setUp(self) -> None:
self.root = TMP / "line"
shutil.rmtree(self.root, ignore_errors=True)
self.root.mkdir(parents=True)
def line(self) -> dict:
return {
"schema_version": 1, "probed_at": "2026-10-27T12:00:00+00:00", "platform": "retroarch",
"platform_version": "v1.22.2", "target": "linux-x86_64", "install": "pack",
"source": {"pack": {"file": "RetroArch_BIOS_Pack.zip", "sha256": "0" * 64, "size": 1,
"members": 1}, "manifest": None, "repo_commit": "1" * 40},
"system": "sega-saturn", "platform_system": "sega-saturn", "emulator": "beetle_saturn",
"core": {"file": "mednafen_saturn_libretro.so", "sha256": "2" * 64,
"artifact": "https://buildbot.libretro.com/x.zip", "last_modified": None},
"content": {"label": "Game (Japan)", "catalog": "redump", "region": "japan"},
"records": ["2026-10-27-0123456789ab"], "rounds": 1, "staged": 3, "verdict": "unknown",
"verdict_evidence": None,
"files_opened": [{"path": "sega_101.bin", "entry": "sega_101.bin", "sha1": "3" * 40,
"created": False}],
"findings": [{"kind": "missing_declared", "path": "x.bin", "declared_by": ["x.bin"], "paths": []}],
"error": None,
}
def test_the_line_is_sealed_valid_and_read_back(self) -> None:
sealed = pp.append_line(self.line(), self.root)
self.assertEqual(mr.read_campaign(self.root, mr.PACK_PROBES), [sealed])
try:
from jsonschema import Draft202012Validator
except ImportError:
self.skipTest("jsonschema is a tooling dependency")
schema = json.loads((REPO_ROOT / "schemas" / "pack-probe.schema.json").read_text())
self.assertEqual(list(Draft202012Validator(schema).iter_errors(sealed)), [])
def test_the_cli_refuses_an_unknown_platform(self) -> None:
with self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()):
pp.build_parser().parse_args(["run", "--platform", "nowhere"])
class SitePages(unittest.TestCase):
"""The Measurements page and the platform pages show the latest probe."""
def test_latest_line_per_probe_with_its_link(self) -> None:
import sitemeasurements as sm
root = TMP / "site"
shutil.rmtree(root, ignore_errors=True)
root.mkdir(parents=True)
first = Line.line(None)
later = {**first, "probed_at": "2026-10-28T12:00:00+00:00", "verdict": "boots"}
other = {**first, "platform": "batocera", "install": "installer"}
for line in (first, later, other):
pp.append_line(line, root)
items = sm.pack_summary(root, {"repo": "https://example.org/r", "commit": "c" * 40})
self.assertEqual([(i["platform"], i["install"], i["verdict"]) for i in items],
[("batocera", "installer", "unknown"), ("retroarch", "pack", "boots")])
self.assertEqual(items[1]["record_url"],
f"https://example.org/r/blob/{'c' * 40}/measurements/pack-probes.jsonl#L2")
self.assertEqual(items[1]["core"], "mednafen_saturn")
explorer = "\n".join(sm.packs_section(items))
self.assertIn("| Platform | System |", explorer)
self.assertIn("| retroarch | sega-saturn | pack | <code>mednafen&#95;saturn</code> | Game (Japan) "
"| `boots` | 1 | [line](", explorer)
page = sm.pack_tested_section(sm.by_platform({"packs": items})["retroarch"])
self.assertEqual(page[0], "## Tested")
self.assertIn("../wiki/measurements.md#packs", page[2])
self.assertNotIn("| retroarch |", "\n".join(page))
self.assertEqual(sm.pack_tested_section([]), [])
def collection_on_disk() -> bool:
return (REPO_ROOT / "bios" / "Sony" / "PlayStation" / "scph5501.bin").is_file()
@unittest.skipUnless(collection_on_disk(), "the collection is not on disk")
class RealLayouts(unittest.TestCase):
"""Both install paths, from this checkout, narrowed to PlayStation."""
@classmethod
def setUpClass(cls) -> None:
cls.work = TMP / "real"
shutil.rmtree(cls.work, ignore_errors=True)
cls.pack = pp.build_pack("retroarch", cls.work / "dist", ["--system", "sony-playstation"])
cls.identity = pp.pack_identity(cls.pack)
@classmethod
def tearDownClass(cls) -> None:
shutil.rmtree(cls.work, ignore_errors=True)
def test_pack_path(self) -> None:
lay = pp.pack_layout(self.pack, self.work / "pack", self.identity)
with zipfile.ZipFile(self.pack) as handle:
members = {i.filename for i in handle.infolist() if not i.is_dir()}
self.assertEqual(set(lay.files), members)
self.assertEqual(lay.files["scph5501.bin"], "scph5501.bin")
self.assertEqual(lay.source["pack"]["members"], len(members))
swan = pp.load_emulator_profiles(str(REPO_ROOT / "emulators"))["swanstation"]
self.assertIn("scph5501.bin", pp.initial_stage(lay, swan, []))
def test_installer_path(self) -> None:
lay = pp.installer_layout("retroarch", self.pack, self.identity,
REPO_ROOT / "install" / "retroarch.json", self.work / "installer",
["--system", "sony-playstation"])
self.assertEqual(lay.root, self.work / "installer" / "system")
self.assertEqual(lay.files["scph5501.bin"], "files:scph5501.bin")
self.assertEqual(lay.source["manifest"]["path"], "install/retroarch.json")
self.assertTrue(all(entry for entry in lay.files.values()), lay.files)
if __name__ == "__main__":
unittest.main()