From cfc7a13be29e1a6c050da6fddf5379ceeac4a6be Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:17:49 +0100 Subject: [PATCH] feat: add the standalone emulator probe standalone_probe.py runs the release binary a platform installs (EmuDeck's asset rule for DuckStation and PCSX2), unpacked without running the AppImage runtime, in core_probe's allowlist sandbox with a private Xvfb, strace and core_probe's oracle and record functions. Records go to measurements/.jsonl. --- scripts/standalone_probe.py | 695 +++++++++++++++++++++++++++++++++ tests/test_standalone_probe.py | 177 +++++++++ 2 files changed, 872 insertions(+) create mode 100644 scripts/standalone_probe.py create mode 100644 tests/test_standalone_probe.py diff --git a/scripts/standalone_probe.py b/scripts/standalone_probe.py new file mode 100644 index 000000000..6a8b397a2 --- /dev/null +++ b/scripts/standalone_probe.py @@ -0,0 +1,695 @@ +#!/usr/bin/env python3 +"""Run a platform's standalone emulator on a game and record what it did. + +The libretro counterpart is core_probe.py: this script records the same way, +in the same measurements/.jsonl, for the release binaries the +platforms install beside RetroArch. + + resolve the release asset a platform's installer picks, by its own rule + fetch download it into .cache/standalone/, by sha256 + run one execution, printed as JSON + check the same execution under several variants of one BIOS file + +The binary is the asset the platform downloads, unpacked from its AppImage +without running it. Each repetition starts a private Xvfb, runs the emulator +in core_probe's allowlist sandbox under strace, and samples the screen at a +fixed interval from the X server's framebuffer file. The BIOS files are +staged in a directory the emulator's settings name as its BIOS folder, the +way the platform configures it, and the files it opened there come from +strace. + +The verdict of a repetition is one of boots, bios_error, crash, hang or +unknown, by the oracle's rules: a reference screen reached, an error string +of the emulator in its log, the process dying, a uniform screen to the end. +""" + +from __future__ import annotations + +import argparse +import contextlib +import datetime as dt +import hashlib +import json +import os +import re +import shutil +import signal +import struct +import subprocess +import sys +import tempfile +import threading +import time +import urllib.request +from collections.abc import Sequence +from dataclasses import dataclass, field +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import core_probe as cp + +ROOT = Path(__file__).resolve().parent.parent +CACHE = ROOT / ".cache" / "standalone" +DEFAULT_WORK_DIR = ROOT / "tmp" / "standalone_probe" +RECORD_VERSION = 4 +SCREEN = (1280, 720) +OBSERVABLES = ("verdict", "screens", "frame_hashes", "uniform_frames") +SCRIPT = "scripts/standalone_probe.py" + + +# -------------------------------------------------------------------------- +# What each emulator is, how the platform installs it, how it is driven +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Emulator: + """A standalone emulator as one platform installs it. + + `asset_*` reproduce the installer's choice: EmuDeck's getReleaseURLGH + takes the first asset, over GitHub's release list in its own order + (pre-releases included), whose name contains `asset_contains` and ends + with `asset_endswith`. + """ + + name: str + platform: str + installer_ref: str + repo: str + asset_endswith: str + asset_contains: str + binary: str + settings_file: str + error_patterns: tuple[str, ...] + settings: dict[str, dict[str, str]] = field(default_factory=dict) + bios_setting: tuple[str, str] = ("", "") + args: tuple[str, ...] = () + + +EMULATORS: dict[str, Emulator] = { + "duckstation": Emulator( + name="duckstation", + platform="emudeck", + installer_ref="dragoonDorise/EmuDeck functions/EmuScripts/emuDeckDuckStation.sh " + "(getReleaseURLGH \"stenzek/duckstation\" \"AppImage\" \"x64.\")", + repo="stenzek/duckstation", + asset_endswith="AppImage", + asset_contains="x64.", + binary="usr/bin/duckstation-qt", + # Core::SetDataRoot (src/core/core.cpp): $XDG_CONFIG_HOME/duckstation on Linux. + settings_file="data/duckstation/settings.ini", + bios_setting=("BIOS", "SearchDirectory"), + settings={ + "Main": {"SetupWizardIncomplete": "false", "ConfirmPowerOff": "false", + "StartFullscreen": "true"}, + "AutoUpdater": {"CheckAtStartup": "false"}, + "GPU": {"Renderer": "Software"}, + "Logging": {"LogLevel": "Info", "LogToConsole": "true", "LogToFile": "false"}, + "Console": {"Region": "Auto"}, + }, + args=("-batch", "-nogui", "-fullscreen", "--"), + # Strings of src/core/bios.cpp and src/core/system.cpp, as the + # release binary carries them. + error_patterns=("No BIOS image found for", "No BIOS images found", + "No BIOS images were found", "Failed to open BIOS"), + ), + "pcsx2": Emulator( + name="pcsx2", + platform="emudeck", + installer_ref="dragoonDorise/EmuDeck functions/EmuScripts/emuDeckPCSX2QT.sh " + "(getReleaseURLGH \"PCSX2/pcsx2\" \"Qt.AppImage\" \"\")", + repo="PCSX2/pcsx2", + asset_endswith="Qt.AppImage", + asset_contains="", + binary="usr/bin/pcsx2-qt", + # EmuFolders::SetDataDirectory (pcsx2/Pcsx2Config.cpp): $XDG_CONFIG_HOME/PCSX2 on Linux, + # its settings in inis/. + settings_file="data/PCSX2/inis/PCSX2.ini", + bios_setting=("Folders", "Bios"), + settings={ + # SETTINGS_VERSION of pcsx2/VMManager.cpp: another value resets every setting. + "UI": {"SettingsVersion": "1", "SetupWizardIncomplete": "false", "ConfirmShutdown": "false", + "StartFullscreen": "true"}, + "AutoUpdater": {"CheckAtStartup": "false"}, + # GSRendererType::SW + "EmuCore/GS": {"Renderer": "13"}, + "Logging": {"EnableSystemConsole": "true", "EnableVerbose": "false"}, + }, + args=("-batch", "-nogui", "-fullscreen", "-earlyconsolelog", "--"), + error_patterns=("Unable to auto locate a BIOS image", "BIOS image has not been selected", + "could not be read (permission denied?)"), + ), +} + + +# -------------------------------------------------------------------------- +# Pure helpers +# -------------------------------------------------------------------------- + +def select_asset(releases: Sequence[dict], endswith: str, contains: str) -> dict | None: + """The asset EmuDeck's getReleaseURLGH returns: the first, over the + releases in the API's order, whose name contains and ends as asked.""" + for release in releases: + for asset in release.get("assets") or (): + name = asset.get("name", "") + if contains in name and name.endswith(endswith): + return {"tag": release.get("tag_name"), "name": name, + "url": asset.get("browser_download_url"), + "prerelease": bool(release.get("prerelease"))} + return None + + +def appimage_offset(head: bytes) -> int: + """Where an AppImage's squashfs starts: the end of its ELF runtime, + read from the section header table, as the runtime computes it.""" + if head[:4] != b"\x7fELF": + raise cp.ProbeError("not an ELF file, not an AppImage") + if head[4] == 2: + shoff = struct.unpack_from(" str: + out = [] + for section, values in sections.items(): + out.append(f"[{section}]") + out.extend(f"{key} = {value}" for key, value in values.items()) + out.append("") + return "\n".join(out) + + +def emulator_settings(emu: Emulator, bios_dir: Path, overrides: Sequence[tuple[str, str]] + ) -> dict[str, dict[str, str]]: + """The settings written before a run: the emulator's own, the BIOS + folder, then each SECTION.KEY=VALUE override.""" + sections = {name: dict(values) for name, values in emu.settings.items()} + section, key = emu.bios_setting + sections.setdefault(section, {})[key] = str(bios_dir) + for name, value in overrides: + sec, _, k = name.rpartition(".") + if not sec or not k: + raise cp.ProbeError(f"--setting {name}: SECTION.KEY expected") + sections.setdefault(sec, {})[k] = value + return sections + + +def parse_xwd(data: bytes) -> tuple[int, int, bytes]: + """Width, height and RGB pixels of an XWD image, as Xvfb's -fbdir + keeps its screen: 32 bits per pixel, least significant byte first.""" + if len(data) < 100: + raise cp.ProbeError("framebuffer file shorter than an XWD header") + fields = struct.unpack_from(">25I", data, 0) + header_size, version = fields[0], fields[1] + width, height = fields[4], fields[5] + byte_order, bits_per_pixel, bytes_per_line, ncolors = fields[7], fields[11], fields[12], fields[19] + if version != 7 or bits_per_pixel != 32 or byte_order != 0: + raise cp.ProbeError(f"XWD version {version}, {bits_per_pixel} bpp, byte order " + f"{byte_order}: only version 7, 32 bpp, LSB first are read") + start = header_size + ncolors * 12 + rows = [data[start + y * bytes_per_line:start + y * bytes_per_line + width * 4] for y in range(height)] + packed = b"".join(rows) + if len(packed) != width * height * 4: + raise cp.ProbeError("framebuffer file shorter than its header says") + return width, height, cp.to_rgb(1, width, height, packed) + + + +# -------------------------------------------------------------------------- +# Release binaries +# -------------------------------------------------------------------------- + +def github_releases(repo: str) -> list[dict]: + request = urllib.request.Request(f"https://api.github.com/repos/{repo}/releases", + headers={"Accept": "application/vnd.github+json"}) + token = os.environ.get("GITHUB_TOKEN") + if token: + request.add_header("Authorization", f"Bearer {token}") + with urllib.request.urlopen(request, timeout=60) as response: + return json.load(response) + + +def fetch(emu: Emulator, cache: Path = CACHE) -> dict: + """Download the asset the platform picks, unpack it, and describe it.""" + asset = select_asset(github_releases(emu.repo), emu.asset_endswith, emu.asset_contains) + if not asset: + raise cp.ProbeError(f"{emu.repo}: no release asset matches the {emu.platform} rule") + folder = cache / emu.name + folder.mkdir(parents=True, exist_ok=True) + partial = folder / f"{asset['name']}.part" + with urllib.request.urlopen(asset["url"], timeout=600) as response, partial.open("wb") as out: + shutil.copyfileobj(response, out) + sha256 = cp.file_digest(partial, "sha256") + target = folder / sha256 / asset["name"] + target.parent.mkdir(parents=True, exist_ok=True) + partial.replace(target) + unpack(target) + meta = {**asset, "sha256": sha256, "repo": emu.repo, "platform": emu.platform, + "installer_ref": emu.installer_ref, "fetched_at": now_iso()} + (target.parent / "asset.json").write_text(json.dumps(meta, indent=2), encoding="utf-8") + return meta + + +def unpack(appimage: Path) -> Path: + """The AppImage's tree, unpacked with unsquashfs: the runtime is never run.""" + tree = appimage.parent / "tree" + if tree.is_dir(): + return tree + unsquashfs = shutil.which("unsquashfs") + if not unsquashfs: + raise cp.ProbeError("unsquashfs not found (squashfs-tools)") + with appimage.open("rb") as handle: + offset = appimage_offset(handle.read(64)) + scratch = appimage.parent / "tree.part" + shutil.rmtree(scratch, ignore_errors=True) + subprocess.run([unsquashfs, "-q", "-n", "-o", str(offset), "-d", str(scratch), str(appimage)], + check=True, stdout=subprocess.DEVNULL) + scratch.replace(tree) + return tree + + +def installed(emu: Emulator, sha256: str | None, cache: Path = CACHE) -> tuple[dict, Path]: + """The fetched asset to run: the one named by sha256, else the newest.""" + folder = cache / emu.name + metas = sorted(folder.glob("*/asset.json"), key=lambda p: p.stat().st_mtime, reverse=True) + for path in metas: + meta = json.loads(path.read_text(encoding="utf-8")) + if sha256 in (None, meta["sha256"]): + return meta, unpack(path.parent / meta["name"]) + raise cp.ProbeError(f"{emu.name}: no fetched asset{' ' + sha256 if sha256 else ''}, " + f"run `standalone_probe.py fetch {emu.name}`") + + +# -------------------------------------------------------------------------- +# One repetition +# -------------------------------------------------------------------------- + +def now_iso() -> str: + return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() + + +# The X server runs inside the sandbox, on the sandbox's own /tmp: the +# display is private to the run, and the host's /tmp/.X11-unix (read-only +# under WSLg) is never touched. -fbdir keeps the screen as an XWD file in the +# run's scratch folder, which the probe reads from outside. +DISPLAY_NUMBER = 1 +XVFB_WRAPPER = ( + f'Xvfb :{DISPLAY_NUMBER} -screen 0 {SCREEN[0]}x{SCREEN[1]}x24 -nolisten tcp -fbdir "$1" -noreset >/dev/null 2>&1 &\n' + 'i=0\n' + f'while [ ! -S /tmp/.X11-unix/X{DISPLAY_NUMBER} ]; do\n' + ' i=$((i + 1)); [ "$i" -gt 400 ] && exit 97; sleep 0.05\n' + 'done\n' + 'shift\n' + 'exec "$@"\n' +) + + +def with_display(argv: list[str], fbdir: Path) -> list[str]: + """The emulator's command, started once the sandbox's Xvfb listens.""" + return ["/bin/sh", "-c", XVFB_WRAPPER, "xvfb", str(fbdir), *argv] + + +def stop(proc: subprocess.Popen) -> None: + if proc.poll() is None: + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + proc.wait() + + +ANSI = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") + + +def read_lines(stream, out: list[dict], samples: list) -> None: + """The emulator's console, line by line, each with the sample count + reached when it was written, colour escapes removed.""" + for raw in iter(stream.readline, b""): + text = ANSI.sub("", raw.decode("utf-8", errors="replace")).rstrip() + if text: + out.append({"frame": len(samples), "level": "info", "text": text}) + + +def run_once(args: argparse.Namespace, emu: Emulator, tree: Path, system_dir: Path, scratch: Path, + png_dir: Path | None) -> dict: + home = scratch / "home" + data = scratch / "data" + for folder in (home, data): + folder.mkdir(parents=True, exist_ok=True) + settings = emulator_settings(emu, system_dir.resolve(), args.setting) + settings_path = scratch / emu.settings_file + settings_path.parent.mkdir(parents=True, exist_ok=True) + settings_path.write_text(ini_text(settings), encoding="utf-8") + content = Path(args.content).resolve() + argv = [str(tree / emu.binary)] + [a.format(data=data.resolve()) for a in emu.args] + [str(content)] + trace = scratch / "strace.txt" + strace = shutil.which("strace") if not args.no_trace else None + if strace: + argv = [strace, "-f", "-qq", "-s", "4096", "-e", "trace=openat,open", "-o", str(trace)] + argv + samples: list[tuple[float, str, bool]] = [] + logs: list[dict] = [] + if not shutil.which("Xvfb"): + raise cp.ProbeError("Xvfb not found") + fbdir = scratch / "fb" + fbdir.mkdir(parents=True, exist_ok=True) + screen_file = fbdir / "Xvfb_screen0" + env = {"HOME": str(home.resolve()), "XDG_CONFIG_HOME": str(data.resolve()), + "XDG_DATA_HOME": str(data.resolve()), "XDG_CACHE_HOME": str((home / ".cache").resolve()), + "DISPLAY": f":{DISPLAY_NUMBER}", "QT_QPA_PLATFORM": "xcb", "LIBGL_ALWAYS_SOFTWARE": "1", + "PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"} + prefix, isolation = cp.sandbox_argv([scratch], [tree, system_dir, content.parent]) + if args.require_sandbox and isolation != cp.SANDBOX: + raise cp.ProbeError(f"refused to run the emulator outside a sandbox: {isolation}") + proc = subprocess.Popen(prefix + with_display(argv, fbdir.resolve()), stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, env=env, start_new_session=True) + reader = threading.Thread(target=read_lines, args=(proc.stdout, logs, samples), daemon=True) + reader.start() + try: + deadline = time.monotonic() + 30 + while not screen_file.exists(): + if proc.poll() is not None or time.monotonic() > deadline: + raise cp.ProbeError(f"Xvfb did not start in the sandbox (exit {proc.poll()}): " + f"{' / '.join(line['text'] for line in logs[-5:])}") + time.sleep(0.05) + started = time.monotonic() + ended = None + for index in range(args.samples): + target = started + (index + 1) * args.interval + while time.monotonic() < target and proc.poll() is None: + time.sleep(min(0.05, max(0.0, target - time.monotonic()))) + if proc.poll() is not None: + code = proc.returncode + ended = {"status": code if code >= 0 else None, + "signal": signal.Signals(-code).name if code < 0 else None} + break + width, height, rgb = parse_xwd(screen_file.read_bytes()) + sha1 = hashlib.sha1(rgb).hexdigest() + samples.append((round(time.monotonic() - started, 3), sha1, cp.uniform_frame(rgb, 3))) + if png_dir and not any(p.name.endswith(f"{sha1[:12]}.png") for p in png_dir.iterdir()): + (png_dir / f"{index:04d}-{sha1[:12]}.png").write_bytes(cp.png_bytes(width, height, rgb)) + wall = time.monotonic() - started + finally: + stop(proc) + reader.join(5) + places = [(str(system_dir.resolve()), ""), (str(scratch.resolve()), ""), + (str(content.parent), ""), (str(tree), ""), (str(ROOT), "")] + for line in logs: + line["text"] = cp.host_paths_hidden(line["text"], places) + rep: dict = { + "loaded": ended is None, + "library_name": emu.name, + "frames": max(len(samples), 1), + "wall_s": round(wall, 3), + "sample_interval_s": args.interval, + "sample_times_s": [t for t, _, _ in samples], + # The shapes core_probe gives them: [sample, sha1] pairs, and the + # digests of the samples that are one colour. + "observables": { + "frame_hashes": [[index, h] for index, (_, h, _) in enumerate(samples)], + "uniform_frames": sorted({h for _, h, u in samples if u}), + "screens": sorted({h for _, h, u in samples if not u}), + }, + "logs": logs, + "isolation": isolation, + } + if ended: + rep["exit"] = ended + if strace and trace.is_file(): + opened, probed = cp.files_under(cp.parse_strace(trace.read_text(encoding="utf-8", errors="replace")), + system_dir) + rep["files_opened"] = [{"path": rel, "sha1_after": cp.file_digest(system_dir / rel)} + for rel in opened if (system_dir / rel).is_file()] + rep["files_probed_absent"] = probed + rep["tracing"] = "strace" + else: + rep["files_opened"] = None + rep["files_probed_absent"] = None + rep["tracing"] = "disabled: --no-trace" if args.no_trace else "unavailable: strace not found" + return rep + + +# -------------------------------------------------------------------------- +# Variants and the record +# -------------------------------------------------------------------------- + +def run_variant(args: argparse.Namespace, emu: Emulator, tree: Path, oracle: dict, + base: dict[str, Path], variant: cp.Variant, work: Path, repetitions: int) -> dict: + system_dir = work / "bios" + staged = cp.stage_system_dir(base, variant, system_dir) + reps: list[dict] = [] + flags: list[str] = [] + for index in range(repetitions): + png_dir = None + if args.png_dir and index == 0: + png_dir = Path(args.png_dir) / variant.label + png_dir.mkdir(parents=True, exist_ok=True) + rep = run_once(args, emu, tree, system_dir, work / f"rep{index}", png_dir) + rep["verdict"], rep["verdict_evidence"] = cp.judge(rep, oracle) + rep["observables"]["verdict"] = rep["verdict"] + for entry in rep.get("files_opened") or []: + expected = staged.get(entry["path"]) + entry["sha1_staged"] = expected + if expected is not None and expected != entry["sha1_after"]: + flags.append(f"repetition {index}: {entry['path']} sha1 changed during the run") + reps.append(rep) + verdict, evidence = cp.variant_verdict(reps) + return {"label": variant.label, "system_files": dict(sorted(staged.items())), "repetitions": reps, + "flags": flags, "verdict": verdict, "verdict_evidence": evidence} + + +def frontend_identity() -> dict: + script = Path(__file__).resolve() + commit = dirty = None + try: + commit = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, capture_output=True, text=True, + check=True).stdout.strip() + dirty = bool(subprocess.run(["git", "status", "--porcelain", "--", str(script)], cwd=ROOT, + capture_output=True, text=True, check=True).stdout.strip()) + except (OSError, subprocess.CalledProcessError) as exc: + print(f"WARNING: frontend commit unknown, git failed: {exc}", file=sys.stderr) + return {"script": SCRIPT, "commit": commit, "script_modified": dirty, + "script_sha256": cp.file_digest(script, "sha256")} + + +def emulator_oracle(emu: Emulator, path: Path, content_sha1: str) -> dict: + """The oracle core_probe reads (tests/campaign/oracle.yml), looked up by + the content's sha1 and the emulator's name, with the emulator's own + error strings of this script added as patterns. Without the file the + oracle names this script, which then holds every pattern.""" + data, identity = cp.load_oracle(path) + entries = cp.oracle_for(data, content_sha1, emu.name) + own = [re.escape(text) for text in emu.error_patterns] + entries["error_patterns"] = list(dict.fromkeys(entries["error_patterns"] + own)) + if identity is None: + identity = {"file": SCRIPT, "sha256": cp.file_digest(Path(__file__).resolve(), "sha256")} + return {**identity, **entries} + + +def check_result(variants: list[dict], observable: str) -> dict: + """core_probe's comparison, with each variant's verdict beside it.""" + return {**cp.check_result(variants, observable), + "verdicts": {v["label"]: v["verdict"] for v in variants}} + + +def probe(args: argparse.Namespace) -> int: + emu = EMULATORS[args.emulator] + problems = validate(args) + if problems: + for problem in problems: + print(f"ERROR: {problem}", file=sys.stderr) + return 2 + meta, tree = installed(emu, args.sha256) + max_load = args.max_load if args.max_load is not None else (os.cpu_count() or 1) / 4 + load_before = cp.load_average() + if args.record and load_before and load_before[0] > max_load: + print(f"ERROR: load average {load_before[0]} exceeds {max_load}: the machine is busy, " + f"nothing recorded", file=sys.stderr) + return 3 + oracle = emulator_oracle(emu, Path(args.oracle) if args.oracle else cp.DEFAULT_ORACLE, + cp.file_digest(Path(args.content))) + repetitions = 1 if args.mode == "run" else args.repeat + base = {dest: Path(src) for dest, src in args.bios_file} + if args.mode == "check": + variants_in = [] + seen: set[str] = set() + for dest, src in args.vary: + label = Path(src).name if src else f"{dest}-absent" + while label in seen: + label += "+" + seen.add(label) + variants_in.append(cp.Variant(label, {dest: Path(src) if src else None})) + else: + variants_in = [cp.Variant("default")] + work = Path(tempfile.mkdtemp(prefix=f"{args.mode}-", dir=cp._ensure(Path(args.work_dir)))) + try: + variants = [run_variant(args, emu, tree, oracle, base, variant, work / f"v{i}", repetitions) + for i, variant in enumerate(variants_in)] + except (cp.ProbeError, OSError, subprocess.SubprocessError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + finally: + shutil.rmtree(work, ignore_errors=True) + load_after = cp.load_average() + record: dict = { + "record_version": RECORD_VERSION, + "emulator": emu.name, + "subject": None, + "supersedes": args.supersedes, + "kind": args.mode, + "recorded_at": now_iso(), + "core": {"file": meta["name"], "sha256": meta["sha256"], "repo": f"https://github.com/{meta['repo']}", + "commit": None, "build_command": None, "compiler": None, "library_name": emu.name, + "library_version": meta["tag"], "artifact": meta["url"]}, + "frontend": frontend_identity(), + "host": {**cp.host_identity(), "load_before": load_before, "load_after": load_after, + "max_load": max_load}, + "inputs": { + "content": cp.content_identity(Path(args.content), args.content_label), + "options": {f"{s}.{k}": v for s, values in + emulator_settings(emu, Path(""), args.setting).items() for k, v in values.items()}, + "env": {}, + "frames": args.samples, + "warmup": 0, + "repetitions": repetitions, + "presses": [], + "observables": list(OBSERVABLES), + "frame_hash_every": 1, + "joypad_ports": 0, + "oracle": oracle, + }, + "variants": variants, + "flags": [f"{v['label']}: {flag}" for v in variants for flag in v["flags"]], + } + if load_before and load_after and max(load_before[0], load_after[0]) > max_load: + record["flags"].append(f"load average reached {max(load_before[0], load_after[0])} " + f"(threshold {max_load})") + if args.mode == "check": + record["check"] = {"claim": args.claim, "method": args.method, "observable": args.observable, + "variable": args.vary[0][0], "result": check_result(variants, args.observable)} + from measurement_records import canonical, sha256_hex, slugify + + if args.subject: + record["subject"] = args.subject + elif args.mode == "check": + digest = sha256_hex(canonical([args.claim, args.method, args.observable, args.vary[0][0]]))[:8] + record["subject"] = f"{slugify(args.claim)}-{digest}" + else: + digest = sha256_hex(canonical([meta["sha256"], record["inputs"]["content"], + record["inputs"]["options"], args.bios_file]))[:8] + record["subject"] = f"run-{slugify(args.content_label or Path(args.content).name, 30)}-{digest}" + if args.record: + try: + lines, record = cp.write_record(Path(args.record), record) + except cp.ProbeError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + print(f"recorded {lines}", file=sys.stderr) + print(json.dumps(record, indent=2)) + return 0 + + +def validate(args: argparse.Namespace) -> list[str]: + problems = [] + if not Path(args.content).is_file(): + problems.append(f"--content {args.content}: not a file") + for dest, src in args.bios_file: + if not Path(src).is_file(): + problems.append(f"--bios-file {dest}={src}: not a file") + if args.oracle and not Path(args.oracle).is_file(): + problems.append(f"--oracle {args.oracle}: not a file") + if args.samples < 2 or args.interval <= 0: + problems.append("--samples must be >= 2 and --interval > 0") + if args.require_sandbox: + isolation = cp.sandbox_argv([])[1] + if isolation != cp.SANDBOX: + problems.append(f"--require-sandbox: {isolation}") + if args.record: + wrong = cp.not_a_checkout_root(Path(args.record)) + if wrong: + problems.append(wrong) + if args.subject and not re.fullmatch(r"[a-z0-9]+(?:[-_][a-z0-9]+)*", args.subject): + problems.append(f"--subject {args.subject!r}: lowercase letters, digits, - and _ only") + if args.mode == "check": + if len({dest for dest, _ in args.vary}) != 1: + problems.append("--vary must name the same DEST in every occurrence") + for dest, src in args.vary: + if src and not Path(src).is_file(): + problems.append(f"--vary {dest}={src}: not a file") + if args.observable not in OBSERVABLES: + problems.append(f"--observable {args.observable}: not one of {', '.join(OBSERVABLES)}") + return problems + + +# -------------------------------------------------------------------------- +# Command line +# -------------------------------------------------------------------------- + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + sub = parser.add_subparsers(dest="mode", required=True) + for name, text in (("resolve", "print the asset the platform's installer picks"), + ("fetch", "download and unpack it into .cache/standalone/")): + p = sub.add_parser(name, help=text) + p.add_argument("emulator", choices=sorted(EMULATORS)) + common = argparse.ArgumentParser(add_help=False, allow_abbrev=False) + common.add_argument("emulator", choices=sorted(EMULATORS)) + common.add_argument("--content", required=True, help="the game: a .cue, .iso or .chd") + common.add_argument("--content-label", help="identity of the content, its Redump or No-Intro name") + common.add_argument("--bios-file", action="append", default=[], metavar="DEST=SRC", + type=lambda t: cp.parse_assignment(t, "--bios-file"), + help="copy SRC into the staged BIOS folder as DEST") + common.add_argument("--setting", action="append", default=[], metavar="SECTION.KEY=VALUE", + type=lambda t: cp.parse_assignment(t, "--setting"), + help="an emulator setting written before the run") + common.add_argument("--sha256", help="the fetched asset to run (default: the newest fetched)") + common.add_argument("--samples", type=int, default=60, help="screen samples per repetition") + common.add_argument("--interval", type=float, default=1.0, help="seconds between two samples") + common.add_argument("--oracle", metavar="FILE", + help="the boot oracle (default: tests/campaign/oracle.yml, as core_probe)") + common.add_argument("--png-dir", help="write each distinct sampled screen of repetition 0 as PNG") + common.add_argument("--record", metavar="DIR", nargs="?", const=str(ROOT), + help="append the record to measurements/.jsonl in the checkout DIR") + common.add_argument("--subject", help="slug naming what the record tests") + common.add_argument("--supersedes", metavar="ID", help="id of an earlier record this one corrects") + common.add_argument("--max-load", type=float, default=None, + help="refuse to record above this 1-minute load average (default: CPUs / 4)") + common.add_argument("--work-dir", default=str(DEFAULT_WORK_DIR), help="scratch directory") + common.add_argument("--no-trace", action="store_true", help="do not run under strace") + common.add_argument("--require-sandbox", action="store_true", + help="refuse to run unless bubblewrap can isolate the emulator") + sub.add_parser("run", parents=[common], help="one execution") + check = sub.add_parser("check", parents=[common], help="compare variants of one BIOS file") + check.add_argument("--vary", action="append", required=True, metavar="DEST=SRC", + type=lambda t: cp.parse_assignment(t, "--vary"), + help="one variant per occurrence; an empty SRC leaves DEST absent") + check.add_argument("--observable", default="verdict", help="compared across variants") + check.add_argument("--claim", required=True, help="the statement under test") + check.add_argument("--method", required=True, help="how the run tests it") + check.add_argument("--repeat", type=int, default=2, help="runs of each variant") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + emu = EMULATORS[args.emulator] + if args.mode == "resolve": + asset = select_asset(github_releases(emu.repo), emu.asset_endswith, emu.asset_contains) + print(json.dumps({"emulator": emu.name, "platform": emu.platform, + "installer_ref": emu.installer_ref, "asset": asset}, indent=2)) + return 0 if asset else 1 + if args.mode == "fetch": + try: + print(json.dumps(fetch(emu), indent=2)) + except (cp.ProbeError, OSError, subprocess.CalledProcessError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + return 0 + return probe(args) + + +if __name__ == "__main__": + from common import run_cli + + run_cli(main) diff --git a/tests/test_standalone_probe.py b/tests/test_standalone_probe.py new file mode 100644 index 000000000..ba63d54a3 --- /dev/null +++ b/tests/test_standalone_probe.py @@ -0,0 +1,177 @@ +"""standalone_probe: the platform's asset choice, the AppImage offset, the +framebuffer reader, the settings it writes and the oracle it judges with.""" + +from __future__ import annotations + +import struct +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "scripts")) + +import core_probe as cp # noqa: E402 +import standalone_probe as sp # noqa: E402 + + +def xwd(width: int, height: int, pixels: bytes, ncolors: int = 0, pad: int = 0) -> bytes: + """An XWD file as Xvfb -fbdir writes it: version 7, 32 bpp, LSB first.""" + name = b"Xvfb main window\0" + header_size = 100 + len(name) + bytes_per_line = width * 4 + pad + fields = [header_size, 7, 2, 24, width, height, 0, 0, 32, 0, 32, 32, bytes_per_line, 4, + 0xFF0000, 0xFF00, 0xFF, 8, 256, ncolors, ncolors, width, height, 0, 0] + rows = b"".join(pixels[y * width * 4:(y + 1) * width * 4] + b"\0" * pad for y in range(height)) + return struct.pack(">25I", *fields) + name + b"\0" * (12 * ncolors) + rows + + +class AssetChoice(unittest.TestCase): + RELEASES = [ + {"tag_name": "preview", "prerelease": True, "assets": [ + {"name": "DuckStation-x64-SSE2.AppImage", "browser_download_url": "u1"}, + {"name": "DuckStation-x64.AppImage", "browser_download_url": "u2"}]}, + {"tag_name": "latest", "prerelease": False, "assets": [ + {"name": "DuckStation-x64.AppImage", "browser_download_url": "u3"}]}, + ] + + def test_first_match_in_api_order_like_getreleaseurlgh(self) -> None: + asset = sp.select_asset(self.RELEASES, "AppImage", "x64.") + self.assertEqual((asset["tag"], asset["name"], asset["url"]), + ("preview", "DuckStation-x64.AppImage", "u2")) + self.assertTrue(asset["prerelease"]) + + def test_no_match(self) -> None: + self.assertIsNone(sp.select_asset(self.RELEASES, "Qt.AppImage", "")) + + def test_every_emulator_names_its_installer_and_profile(self) -> None: + profiles = {p.stem for p in (ROOT / "emulators").glob("*.yml")} + for name, emu in sp.EMULATORS.items(): + self.assertEqual(name, emu.name) + self.assertIn(name, profiles) + self.assertIn("EmuDeck", emu.installer_ref) + self.assertTrue(emu.error_patterns) + + +class AppImage(unittest.TestCase): + def test_offset_is_the_end_of_the_section_headers(self) -> None: + head = bytearray(64) + head[:5] = b"\x7fELF\x02" + struct.pack_into(" None: + with self.assertRaises(cp.ProbeError): + sp.appimage_offset(b"hsqs" + bytes(60)) + + +class Framebuffer(unittest.TestCase): + def test_reads_rgb_and_skips_colormap_and_padding(self) -> None: + # Two pixels, BGRX: red then blue. + pixels = bytes((0, 0, 255, 0, 255, 0, 0, 0)) + width, height, rgb = sp.parse_xwd(xwd(2, 1, pixels, ncolors=3, pad=8)) + self.assertEqual((width, height), (2, 1)) + self.assertEqual(rgb, bytes((255, 0, 0, 0, 0, 255))) + + def test_refuses_other_layouts(self) -> None: + data = bytearray(xwd(1, 1, bytes(4))) + struct.pack_into(">I", data, 44, 16) + with self.assertRaises(cp.ProbeError): + sp.parse_xwd(bytes(data)) + + def test_truncated(self) -> None: + with self.assertRaises(cp.ProbeError): + sp.parse_xwd(xwd(4, 4, bytes(64))[:-10]) + + +class Settings(unittest.TestCase): + def test_bios_folder_and_overrides(self) -> None: + emu = sp.EMULATORS["duckstation"] + sections = sp.emulator_settings(emu, Path("/b"), [("GPU.Renderer", "Vulkan"), ("New.Key", "1")]) + self.assertEqual(sections["BIOS"]["SearchDirectory"], "/b") + self.assertEqual(sections["GPU"]["Renderer"], "Vulkan") + self.assertEqual(sections["New"]["Key"], "1") + self.assertEqual(emu.settings["GPU"]["Renderer"], "Software") + text = sp.ini_text(sections) + self.assertIn("[BIOS]\nSearchDirectory = /b\n", text) + + def test_override_needs_a_section(self) -> None: + with self.assertRaises(cp.ProbeError): + sp.emulator_settings(sp.EMULATORS["pcsx2"], Path("/b"), [("Renderer", "13")]) + + def test_display_wrapper_passes_the_command_through(self) -> None: + argv = sp.with_display(["/emu", "--", "game.cue"], Path("/fb")) + self.assertEqual(argv[:2], ["/bin/sh", "-c"]) + self.assertEqual(argv[3:], ["xvfb", "/fb", "/emu", "--", "game.cue"]) + self.assertIn(f"Xvfb :{sp.DISPLAY_NUMBER}", argv[2]) + + +class Oracle(unittest.TestCase): + def test_emulator_patterns_join_the_oracle_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "oracle.yml" + path.write_text("error_patterns:\n - {pattern: 'shared', source_ref: x}\n" + "cores:\n duckstation:\n error_screens:\n" + " - {sha1: '" + "a" * 40 + "', shows: 'no BIOS'}\n" + "content:\n - sha1: '" + "b" * 40 + "'\n references:\n" + " duckstation:\n frames:\n" + " - {sha1: '" + "c" * 40 + "', stage: bios, shows: 'logo'}\n", + encoding="utf-8") + oracle = sp.emulator_oracle(sp.EMULATORS["duckstation"], path, "b" * 40) + self.assertEqual(oracle["references"][0]["sha1"], "c" * 40) + self.assertEqual(oracle["error_screens"][0]["shows"], "no BIOS") + self.assertEqual(oracle["error_patterns"][0], "shared") + self.assertIn(r"No\ BIOS\ image\ found\ for", oracle["error_patterns"]) + self.assertEqual(len(oracle["sha256"]), 64) + + def test_without_the_file_the_oracle_names_the_script(self) -> None: + oracle = sp.emulator_oracle(sp.EMULATORS["pcsx2"], Path("/nonexistent/oracle.yml"), "0" * 40) + self.assertEqual(oracle["file"], sp.SCRIPT) + self.assertEqual(oracle["references"], []) + + def test_a_log_line_gives_bios_error_through_core_probe_judge(self) -> None: + oracle = sp.emulator_oracle(sp.EMULATORS["duckstation"], Path("/nonexistent"), "0" * 40) + rep = {"loaded": True, "observables": {"frame_hashes": [[0, "d" * 40]], "uniform_frames": []}, + "logs": [{"frame": 3, "level": "info", "text": "E: No BIOS image found for NTSC-J region."}]} + verdict, evidence = cp.judge(rep, oracle) + self.assertEqual((verdict, evidence["rule"]), ("bios_error", "error_pattern")) + + def test_an_early_exit_is_a_crash_and_a_black_tail_a_hang(self) -> None: + oracle = {"references": [], "error_patterns": [], "error_screens": []} + rep = {"loaded": False, "exit": {"status": 1, "signal": None}, "observables": {}, "logs": []} + self.assertEqual(cp.judge(rep, oracle)[0], "crash") + black = "e" * 40 + rep = {"loaded": True, "logs": [], "observables": { + "frame_hashes": [[0, "f" * 40], [1, black], [2, black]], "uniform_frames": [black]}} + self.assertEqual(cp.judge(rep, oracle)[0], "hang") + + def test_check_result_carries_the_verdicts(self) -> None: + def variant(label: str, verdict: str) -> dict: + return {"label": label, "flags": [], "verdict": verdict, "repetitions": [ + {"loaded": True, "observables": {"verdict": verdict}}]} + result = sp.check_result([variant("absent", "bios_error"), variant("scph5501.bin", "boots")], + "verdict") + self.assertEqual(result["status"], "differs") + self.assertEqual(result["verdicts"], {"absent": "bios_error", "scph5501.bin": "boots"}) + + +class CommandLine(unittest.TestCase): + def test_check_refuses_two_destinations(self) -> None: + with tempfile.NamedTemporaryFile(suffix=".cue") as content: + args = sp.build_parser().parse_args( + ["check", "duckstation", "--content", content.name, "--vary", "a.bin=", "--vary", "b.bin=", + "--claim", "c", "--method", "m"]) + self.assertIn("--vary must name the same DEST in every occurrence", sp.validate(args)) + + def test_unknown_observable(self) -> None: + with tempfile.NamedTemporaryFile(suffix=".cue") as content: + args = sp.build_parser().parse_args( + ["check", "pcsx2", "--content", content.name, "--vary", "a.bin=", "--observable", "fps", + "--claim", "c", "--method", "m"]) + self.assertTrue(any("--observable fps" in p for p in sp.validate(args))) + + +if __name__ == "__main__": + unittest.main()