chore: merge the engine options round

This commit is contained in:
Abdessamad Derraz committed 2026-10-23 19:25:34 +02:00
commit 6f958382b7
12 files changed
+1581 -47

No files matched your search

+76
View File
@@ -3864,6 +3864,7 @@ class HintsGiveTheUsersOwnCommandBack(unittest.TestCase):
system=None, core=None, emulator=None, region=None, target=None,
cores_dir=None, every_board=False, installed_cores=False,
standalone_copies=False, check=False, dry_run=False, no_game_data=False,
declared_only=False,
)
flags = list(argv)
while flags:
@@ -3908,6 +3909,13 @@ class HintsGiveTheUsersOwnCommandBack(unittest.TestCase):
self.assertIn("| sudo sh -s -- --platform retroarch --dest /media/ro "
"--system atari-lynx --region us", advice)
def test_the_platforms_own_list_is_kept_in_the_hint(self):
self._record("--declared-only", "--system", "atari-lynx")
install._INVOCATION.os_type = "linux"
with unittest.mock.patch.dict(os.environ, {"RETROBIOS_BOOTSTRAP": ""}):
command = install._INVOCATION.command("retroarch", None)
self.assertIn("--system atari-lynx --declared-only", command)
def test_windows_has_no_sudo(self):
self._record("--system", "atari-lynx")
install._INVOCATION.os_type = "windows"
@@ -5070,6 +5078,10 @@ class EventsDescribeTheRun(unittest.TestCase, _SyntheticTree):
"bios_dir": str(self.bios), "files": 4, "present": 1, "missing": 2,
"wrong_hash": 1, "bytes_needed": 245, "data_missing": 0,
"not_installable": 1},
# Every file of this list is the platform's own (cores: null).
{"event": "declared", "platform": "retroarch", "files": 4, "present": 1,
"missing": 2, "wrong_hash": 1, "bytes_needed": 245,
"dests": ["a.bin", "b.bin", "c.bin", "d.bin"]},
]
SKIPPED = [{"reason": "not_found", "label": "not found", "files": 1, "required": 1}]
@@ -5153,6 +5165,70 @@ class EventsDescribeTheRun(unittest.TestCase, _SyntheticTree):
for key in keys:
self.assertIn(f"`{key}`", row, f"{kind}.{key}")
def test_a_run_is_valid_against_the_published_schema(self):
try:
from jsonschema import Draft202012Validator # noqa: PLC0415
except ImportError:
self.skipTest("jsonschema is not installed")
schema = json.loads(EVENT_SCHEMA_PATH.read_text(encoding="utf-8"))
validator = Draft202012Validator(schema)
for argv in (("--check",), ()):
_proc, events = self._run(*argv)
for event in events:
errors = [error.message for error in validator.iter_errors(event)]
self.assertEqual(errors, [], event)
self.assertTrue(list(validator.iter_errors({"event": "file", "platform": "x"})))
self.assertTrue(list(validator.iter_errors(
{"event": "end", "code": 0, "extra": 1})))
EVENT_SCHEMA_PATH = REPO_ROOT / "schemas" / "install-events.schema.json"
class TheEventSchemaFileIsEventKeys(unittest.TestCase):
"""The table in the wiki was the only description of --events a reader
outside Python could check a stream against."""
def _generator(self):
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import generate_event_schema # noqa: PLC0415
return generate_event_schema
def test_the_committed_file_is_the_generated_one(self):
generator = self._generator()
self.assertEqual(EVENT_SCHEMA_PATH.read_text(encoding="utf-8"),
generator.render(generator.build_schema()),
"run python scripts/generate_event_schema.py")
def test_each_kind_has_exactly_its_keys(self):
schema = json.loads(EVENT_SCHEMA_PATH.read_text(encoding="utf-8"))
self.assertEqual(schema["properties"]["event"]["enum"], list(install.EVENT_KEYS))
for kind, keys in install.EVENT_KEYS.items():
shape = schema["$defs"][kind]
self.assertEqual(shape["required"], ["event", *keys], kind)
self.assertEqual(list(shape["properties"]), ["event", *keys], kind)
self.assertIs(shape["additionalProperties"], False, kind)
start = schema["$defs"]["start"]["properties"]["schema"]
self.assertEqual(start, {"const": install.EVENT_SCHEMA})
def test_a_key_without_a_type_stops_the_generator(self):
generator = self._generator()
with unittest.mock.patch.dict(install.EVENT_KEYS, {"new": ("untyped_key",)}), \
unittest.mock.patch.object(generator, "EVENT_KEYS", install.EVENT_KEYS), \
self.assertRaises(KeyError):
generator.build_schema()
def test_the_check_command_fails_on_a_stale_file(self):
generator = self._generator()
scratch = Path(_tempdir(self)) / "events.schema.json"
scratch.write_text("{}\n", encoding="utf-8")
with contextlib.redirect_stdout(io.StringIO()), \
contextlib.redirect_stderr(io.StringIO()):
self.assertEqual(generator.main(["--check", "--output", str(scratch)]), 1)
self.assertEqual(generator.main(["--output", str(scratch)]), 0)
self.assertEqual(generator.main(["--check", "--output", str(scratch)]), 0)
class NetworkFailuresInPlainWords(unittest.TestCase):
"""Offline, the run printed "[Errno -2] Name or service not known". A
+167
View File
@@ -21,6 +21,7 @@ import sys
import tempfile
import time
import unittest
import unittest.mock
import zipfile
from pathlib import Path
@@ -589,5 +590,171 @@ class TheSummaryNamesTheBulk(unittest.TestCase):
self.assertLess(text.index("Done."), text.index("arcade pulls"))
class OfflineMakesNoRequest(Scratch):
"""A handheld with no Wi-Fi ran --check against a loopback address to
learn its platform, since nothing said "make no request"."""
def _served(self, *options: str, env: dict | None = None):
import test_install_selection # noqa: PLC0415
hits: list[str] = []
with FakeRepository() as repo, unittest.mock.patch.object(
test_install_selection._QuietHandler, "log_message",
lambda handler, *args: hits.append(handler.path)):
started = time.monotonic()
proc = subprocess.run(
repo.command(*options), env={**repo.env(), **(env or {})},
capture_output=True, text=True, timeout=60, check=False,
stdin=subprocess.DEVNULL)
elapsed = time.monotonic() - started
installed = repo.installed()
return proc, hits, elapsed, installed
def test_check_names_the_platform_and_stops_at_the_file_list(self):
for options, env in (((("--offline",)), None),
((), {"RETROBIOS_INSTALL_OFFLINE": "1"})):
with self.subTest(options=options, env=env):
proc, hits, elapsed, installed = self._served(
*options, "--check", "--events", env=env)
self.assertEqual(hits, [])
self.assertLess(elapsed, 30)
self.assertEqual(proc.returncode, 1, proc.stderr)
events = [json.loads(line) for line in proc.stdout.splitlines()]
self.assertEqual([e["event"] for e in events],
["start", "detect", "error", "end"])
self.assertEqual(events[1]["platforms"][0]["platform"], "retroarch")
self.assertEqual((events[2]["stage"], events[2]["reason"]),
("manifest", "offline"))
self.assertIn("no network request", events[2]["message"])
self.assertIn("--manifest", proc.stderr)
self.assertNotIn("attempts", proc.stderr)
self.assertEqual(installed, [])
def test_a_copy_lands_and_the_rest_fails_at_once(self):
old = self.root / "old"
old.mkdir()
(old / "scph5501.bin").write_bytes(US)
proc = self.run_installer("--offline", "--manifest", str(self.list),
"--from", str(old), "--no-input")
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertEqual((self.dest / "scph5501.bin").read_bytes(), US)
self.assertFalse((self.dest / "gba_bios.bin").exists())
self.assertIn("no network request", proc.stdout + proc.stderr)
self.assertIn("run without --offline", proc.stdout + proc.stderr)
self.assertNotIn("trying again", proc.stderr)
def test_modes_that_need_the_network_refuse_it(self):
for extra in (["--target", "switch"], ["--list-targets"]):
with self.subTest(extra=extra):
proc = self.run_installer("--offline", *extra)
self.assertEqual(proc.returncode, 2, proc.stderr)
self.assertIn("--offline", proc.stderr)
def test_a_listing_and_a_pack_need_no_network_anyway(self):
proc = self.run_installer("--offline", "--list-platforms")
self.assertEqual(proc.returncode, 0, proc.stderr)
class ANetworkThatDroppedFailsFast(unittest.TestCase):
"""A dropped network cost 93 s before the line-mode error: three tries
of a 30-second timeout spent waiting for a connection that never opened."""
def test_the_connection_is_bounded_and_the_read_is_not(self):
calls: list[tuple] = []
class Sock:
def settimeout(self, value):
calls.append(("read", value))
def create(address, timeout, source_address=None):
calls.append(("connect", timeout))
return Sock()
with unittest.mock.patch.object(install.socket, "create_connection", create):
install._open_socket(("example.org", 443), 60)
install._open_socket(("example.org", 443), 2)
self.assertEqual(calls, [("connect", install.CONNECT_TIMEOUT), ("read", 60),
("connect", 2), ("read", 2)])
def test_three_tries_that_never_connect_end_within_fifteen_seconds(self):
waited = sum(install._retry_delay(n, None) for n in range(1, install.MAX_RETRIES))
self.assertLessEqual(install.MAX_RETRIES * install.CONNECT_TIMEOUT + waited, 15)
def test_a_slow_answer_on_an_open_connection_still_arrives(self):
import http.server # noqa: PLC0415
import threading # noqa: PLC0415
delay = install.CONNECT_TIMEOUT + 1
class Slow(http.server.BaseHTTPRequestHandler):
def do_GET(self): # noqa: N802 - stdlib name
time.sleep(delay)
body = b'{"ok": true}'
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
return
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Slow)
threading.Thread(target=server.serve_forever, daemon=True).start()
self.addCleanup(server.server_close)
self.addCleanup(server.shutdown)
install.install_bounded_opener()
self.addCleanup(install.urllib.request.install_opener, None)
url = f"http://127.0.0.1:{server.server_address[1]}/list.json"
started = time.monotonic()
self.assertEqual(install._fetch_json(url, 1024, "file list", attempts=1), {"ok": True})
self.assertGreaterEqual(time.monotonic() - started, delay - 0.5)
class ADryRunOpensOnItsTotals(Scratch):
"""A full --dry-run printed 599 lines, the total scrolled away above
eight file names for each of 66 systems."""
def setUp(self):
super().setUp()
files = [
{"dest": f"sys{n:02d}/file{k}.bin", "sha1": _sha1(b"%d-%d" % (n, k)),
"size": 1000 * (n + 1), "repo_path": f"bios/sys{n:02d}_{k}", "cores": None,
"systems": [f"system-{n:02d}"]}
for n in range(40) for k in range(3)
]
self.list.write_text(json.dumps({"manifest_version": 2, "platform": "retroarch",
"files": files}))
def _plan(self, *options: str) -> list[str]:
proc = subprocess.run(
[sys.executable, str(REPO_ROOT / "install.py"), "--platform", "retroarch",
"--dest", str(self.dest), "--manifest", str(self.list), "--dry-run",
"--no-input", *options],
env={**_offline_env(), "LINES": "24", "COLUMNS": "80"},
capture_output=True, text=True, timeout=120, check=False,
stdin=subprocess.DEVNULL)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
lines = proc.stdout.splitlines()
return lines[next(i for i, line in enumerate(lines)
if line.startswith("Would download")):]
def test_the_total_comes_first_and_the_largest_systems_after(self):
plan = self._plan()
self.assertTrue(plan[0].startswith("Would download 120 files"), plan[0])
self.assertLessEqual(len(plan), 24, "\n".join(plan))
self.assertIn("system-39", plan[2])
self.assertTrue(any("30 more systems" in line and "--verbose" in line
for line in plan), "\n".join(plan))
self.assertFalse(any("file0.bin" in line for line in plan))
self.assertEqual(plan[-1].strip(), "Nothing was written. Run without --dry-run to install.")
def test_verbose_lists_every_system_and_its_files(self):
plan = self._plan("--verbose")
self.assertTrue(plan[0].startswith("Would download 120 files"), plan[0])
for n in range(40):
self.assertTrue(any(line.strip().startswith(f"system-{n:02d}") for line in plan), n)
self.assertTrue(any("sys00/file0.bin" in line for line in plan))
if __name__ == "__main__":
unittest.main()
+192
View File
@@ -1296,5 +1296,197 @@ class AnUnknownNativeNameIsRefusedBeforeAnyWork(unittest.TestCase):
self.assertEqual(json.loads(proc.stdout)["batocera"]["systems"], ["sony-playstation"])
class ThePlatformsOwnListAlone(unittest.TestCase):
"""A setup that checks its own list asked --check about 14,508 files on
Batocera, 353 of them its own, and had no option to ask about those:
--declared-only keeps the entries the platform declares (cores: null)."""
FILES = [
{"dest": "a.bin", "size": 1, "cores": None, "systems": ["sys-a"],
"alternatives": [{"sha1": "2" * 40, "size": 2, "cores": ["core-b"]}]},
{"dest": "b.bin", "size": 1, "cores": None, "systems": ["sys-a"],
"sha1": "1" * 40,
"slot_choices": [{"sha1": "3" * 40, "size": 3, "cores": ["core-b"]}],
"declared_choices": [{"sha1": "1" * 40, "size": 1, "systems": ["sys-a"]}]},
{"dest": "core/c.bin", "size": 1, "cores": ["core-b"], "systems": ["sys-a"]},
]
OMITTED = [
{"dest": "d.bin", "name": "d.bin", "system": "sys-a", "systems": ["sys-a"],
"required": True, "reason": "not_found", "cores": None},
{"dest": "core/e.bin", "name": "e.bin", "system": "sys-a", "systems": ["sys-a"],
"required": True, "reason": "not_found", "cores": ["core-b"]},
]
def test_the_cut(self):
files, omitted = install.keep_declared(list(self.FILES), list(self.OMITTED))
self.assertEqual([f["dest"] for f in files], ["a.bin", "b.bin"])
self.assertEqual([o["dest"] for o in omitted], ["d.bin"])
# The platform answers its own slot, and no core copy is left to
# stand in for it when a later narrowing drops the system.
self.assertEqual(files[1]["sha1"], "1" * 40)
for entry in files:
self.assertIsNone(entry["cores"])
self.assertFalse({"alternatives", "slot_choices", "declared_choices"} & set(entry))
self.assertEqual(install.available_choices(files, omitted)["cores"], [])
def test_install_check_and_dry_run_apply_it(self):
with FakeRepository() as repo:
proc = repo.run("--declared-only")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("5 -> 3 files", proc.stdout)
self.assertEqual(repo.installed(),
["gba_bios.bin", "scph5500.bin", "scph5501.bin"])
for mode in ("--check", "--dry-run"):
with self.subTest(mode=mode), FakeRepository() as repo:
proc = repo.run("--declared-only", mode)
self.assertIn("5 -> 3 files", proc.stdout, proc.stderr)
self.assertIn("0/3 present", proc.stdout)
self.assertEqual(repo.installed(), [])
def test_it_composes_with_the_other_narrowings(self):
with FakeRepository() as repo:
proc = repo.run("--declared-only", "--system", "sony-playstation",
"--region", "us")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(repo.installed(), ["scph5501.bin"])
with FakeRepository() as repo:
listing = repo.run("--declared-only", "--list-systems")
self.assertEqual(listing.returncode, 0, listing.stderr)
row = next(line for line in listing.stdout.splitlines()
if "sony-playstation" in line)
self.assertIn("2 files", row)
def test_a_local_file_list_applies_it(self):
with FakeRepository() as repo:
proc = repo.run("--declared-only", "--dry-run", "--manifest",
str(repo.root / "install" / "retroarch.json"))
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("5 -> 3 files", proc.stdout)
def test_every_mode_that_cannot_apply_it_refuses_it(self):
"""Applied or refused, never ignored: each of these would have run
as if the option was not there."""
refused = [
["--list-platforms"],
["--list-targets"],
["--list-cores"],
["--list-emulators"],
["--core", "mgba"],
["--standalone-copies"],
["--tui"],
]
with FakeRepository() as repo:
for extra in refused:
with self.subTest(extra=extra):
proc = repo.run("--declared-only", *extra)
self.assertEqual(proc.returncode, 2, proc.stdout + proc.stderr)
self.assertIn("--declared-only", proc.stderr)
self.assertEqual(repo.installed(), [])
proc = subprocess.run(
[sys.executable, str(REPO_ROOT / "install.py"), "--restore",
str(REPO_ROOT / "tmp"), "--declared-only"],
capture_output=True, text=True, timeout=60, check=False,
stdin=subprocess.DEVNULL)
self.assertEqual(proc.returncode, 2)
self.assertIn("takes no --declared-only", proc.stderr)
def test_events_count_the_platforms_own_files(self):
with FakeRepository() as repo:
proc = repo.run("--check", "--events")
events = [json.loads(line) for line in proc.stdout.splitlines()]
declared = [e for e in events if e["event"] == "declared"]
self.assertEqual(declared, [{
"event": "declared", "platform": "retroarch", "files": 3, "present": 0,
"missing": 3, "wrong_hash": 0, "bytes_needed": 22,
"dests": ["scph5500.bin", "scph5501.bin", "gba_bios.bin"],
}])
kinds = [e["event"] for e in events]
self.assertEqual(kinds.index("declared"), kinds.index("check") + 1)
narrowed = [json.loads(line) for line in
repo.run("--check", "--events", "--declared-only").stdout.splitlines()]
check = next(e for e in narrowed if e["event"] == "check")
self.assertEqual(check["files"], 3)
class AnIdAnotherSetupUses(unittest.TestCase):
"""The Neo Geo card offers `--system snk-neogeo`, and RetroArch, which
files neogeo.zip under arcade, answered "Did you mean: snk-neogeo-cd?"."""
KNOWN = ["arcade", "snk-neogeo-cd", "sony-playstation"]
HOLDERS = {"snk-neogeo": ["arcade"]}
def _refusal(self, typed: str, holders=None) -> str:
with self.assertRaises(ValueError) as ctx:
install.resolve_choices([typed], self.KNOWN, "system",
holders=self.HOLDERS if holders is None else holders)
return str(ctx.exception)
def test_the_system_holding_its_file_is_suggested_never_taken(self):
for typed in ("snk-neogeo", "SNK-NeoGeo", "neogeo", "mvs", "aes"):
with self.subTest(typed=typed):
text = self._refusal(typed)
self.assertIn("Did you mean: arcade?", text)
self.assertIn("neogeo.zip", text)
self.assertNotIn("snk-neogeo-cd", text)
def test_a_typo_still_gets_the_closest_names(self):
self.assertIn("Did you mean: snk-neogeo-cd", self._refusal("snk-neogeo-dc", {}))
def test_holders_are_read_from_the_file_list(self):
files = [
{"dest": "neogeo.zip", "systems": ["arcade"]},
{"dest": "fbneo/neogeo.zip", "systems": ["arcade"]},
{"dest": "neocd/neocd.bin", "systems": ["snk-neogeo-cd"]},
{"dest": "Other/neogeo.zip", "systems": []},
]
self.assertEqual(install.key_file_holders(files, []), {"snk-neogeo": ["arcade"]})
own = [{"dest": "neogeo.zip", "systems": ["snk-neogeo"]}]
self.assertEqual(install.key_file_holders(own, []), {})
def test_retroarch_suggests_arcade(self):
never = REPO_ROOT / "tmp" / "never"
proc = subprocess.run(
[sys.executable, str(REPO_ROOT / "install.py"), "--platform", "retroarch",
"--manifest", str(REPO_ROOT / "install" / "retroarch.json"),
"--dest", str(never), "--system", "snk-neogeo", "--dry-run", "--no-input"],
capture_output=True, text=True, timeout=120, check=False,
stdin=subprocess.DEVNULL)
self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
self.assertIn("Did you mean: arcade?", proc.stderr)
self.assertNotIn("snk-neogeo-cd", proc.stderr)
self.assertFalse(never.exists())
class TheTitleASetupPrints(unittest.TestCase):
"""The installer and the site print "SNK Neo Geo CD": typed back, the
title was refused."""
MANIFEST = {
"system_names": {"sony-playstation": ["psx"]},
"system_titles": {"snk-neogeo-cd": "SNK Neo Geo CD", "arcade": "Arcade games",
"sony-playstation": "Sony PlayStation",
"twin-a": "Twin", "twin-b": "Twin"},
}
KNOWN = ["arcade", "snk-neogeo-cd", "sony-playstation", "twin-a", "twin-b"]
def _resolve(self, typed: str) -> list[str]:
return install.resolve_choices([typed], self.KNOWN, "system",
install.system_aliases(self.MANIFEST))
def test_a_title_is_taken_for_its_id(self):
self.assertEqual(self._resolve("SNK Neo Geo CD"), ["snk-neogeo-cd"])
self.assertEqual(self._resolve("arcade games"), ["arcade"])
self.assertEqual(self._resolve("psx"), ["sony-playstation"])
def test_a_title_two_systems_share_is_not_taken(self):
with self.assertRaises(ValueError):
self._resolve("Twin")
def test_the_platforms_own_name_wins_over_a_title(self):
manifest = {"system_names": {"arcade": ["Sony PlayStation"]},
"system_titles": {"sony-playstation": "Sony PlayStation"}}
self.assertEqual(install.system_aliases(manifest)["Sony PlayStation"], "arcade")
if __name__ == "__main__":
unittest.main()
+4 -4
View File
@@ -94,10 +94,6 @@ class SemanticManifestValidation(unittest.TestCase):
self.assertIn("total_files does not equal len(files)", errors[0])
if __name__ == "__main__":
unittest.main()
try:
import jsonschema
except ImportError:
@@ -122,3 +118,7 @@ class DataRootIsKeyedBySystem(unittest.TestCase):
self.assertTrue(self._errors({"amiga": {"path": "x", "source_ref": "y"}}))
self.assertTrue(self._errors({"windows": {"path": "%APPDATA%/X"}}))
self.assertTrue(self._errors({}))
if __name__ == "__main__":
unittest.main()