mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-01 15:43:45 -06:00
672 lines
30 KiB
Python
672 lines
30 KiB
Python
#!/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)
|
|
# An entry the core found under one of its names asks for nothing more:
|
|
# BlastEm tries tmss.md, then reads bios_MD.bin, both names of one entry.
|
|
read = {item["path"] for item in opened if not item["created"]}
|
|
satisfied = {entry["name"] for entry in libretro_entries(profile) if read & set(entry_paths(entry))}
|
|
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
|
|
if names and set(names) <= satisfied:
|
|
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())
|