Files
libretro/tests/test_install_selection.py
T

541 lines
23 KiB
Python

"""The installer narrows a platform the way the pack builder does.
The one-liner can install a chosen set of systems, cores and regions. It runs
without the profiles or the database, from the manifest alone, so the
manifest records each file's systems and regional competition and the
installer replays the builder's region rule. These tests hold the two ends
to one answer.
"""
from __future__ import annotations
import functools
import hashlib
import http.server
import importlib.util
import json
import os
import random
import shlex
import shutil
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "scripts"))
import region # noqa: E402
_spec = importlib.util.spec_from_file_location("install", REPO_ROOT / "install.py")
install = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(install)
class OneRegionVocabulary(unittest.TestCase):
"""install.py cannot import region.py, so it carries a copy."""
def test_the_copy_matches(self):
self.assertEqual(install.REGION_TREE, dict(region.REGION_TREE))
self.assertEqual(install.REGION_ALIASES, region.ALIASES)
self.assertEqual(install.WORLD_REGION, region.WORLD)
self.assertEqual(install.REGIONS, region.REGIONS)
class TheRegionRuleIsTheBuilders(unittest.TestCase):
"""Random groups, both implementations, one answer."""
POOL = ["north-america", "europe", "japan", "uk", "canada", "brazil",
"south-korea", "world", "asia", "oceania"]
def _case(self, rng: random.Random):
groups: dict[str, list[tuple[str, str]]] = {}
index: dict[str, dict] = {}
entries: dict[str, dict] = {}
for g in range(rng.randint(1, 4)):
for m in range(rng.randint(1, 5)):
dest = f"d{g}_{m}" if rng.random() < 0.8 else f"shared{m}"
if dest not in entries:
regions = (
set()
if rng.random() < 0.2
else set(rng.sample(self.POOL, rng.randint(1, 2)))
)
entries[dest] = {"dest": dest, "regions": sorted(regions),
"region_groups": []}
index[dest] = {"regions": regions,
"has_untagged": not regions, "emulators": []}
groups.setdefault(f"g{g}", []).append((dest, dest))
if f"g{g}" not in entries[dest]["region_groups"]:
entries[dest]["region_groups"].append(f"g{g}")
for entry in entries.values():
if not entry["regions"]:
del entry["regions"], entry["region_groups"]
requested = rng.sample(["north-america", "europe", "japan", "uk", "canada"],
rng.randint(1, 3))
return groups, index, list(entries.values()), requested
def test_two_thousand_random_platforms(self):
rng = random.Random(20261010)
for trial in range(2000):
groups, index, entries, requested = self._case(rng)
with self.subTest(trial=trial):
self.assertEqual(
install.region_drops(entries, requested),
region.resolve_region_drops(groups, index, requested),
)
class Narrowing(unittest.TestCase):
FILES = [
{"dest": "scph5500.bin", "size": 1, "cores": None,
"systems": ["sony-playstation"], "regions": ["japan"],
"region_system_groups": ["sony-playstation"]},
{"dest": "scph5501.bin", "size": 1, "cores": None,
"systems": ["sony-playstation"], "regions": ["north-america"],
"region_system_groups": ["sony-playstation"]},
{"dest": "gba_bios.bin", "size": 1, "cores": None,
"systems": ["nintendo-gba"]},
{"dest": "pcsx/extra.bin", "size": 1, "cores": ["pcsx_rearmed"],
"systems": ["sony-playstation"]},
{"dest": "beetle/extra.bin", "size": 1, "cores": ["beetle_psx"],
"systems": ["sony-playstation"]},
]
OMITTED = [
{"dest": "scph5502.bin", "name": "scph5502.bin", "system": "sony-playstation",
"required": True, "reason": "not_found", "cores": None,
"regions": ["europe"], "region_system_groups": ["sony-playstation"]},
]
def _dests(self, *args):
files, omitted = install.narrow(list(self.FILES), list(self.OMITTED), *args)
return sorted(f["dest"] for f in files), sorted(o["dest"] for o in omitted)
def test_nothing_chosen_keeps_everything(self):
self.assertEqual(len(self._dests([], [], [])[0]), len(self.FILES))
def test_a_system(self):
files, omitted = self._dests(["nintendo-gba"], [], [])
self.assertEqual(files, ["gba_bios.bin"])
self.assertEqual(omitted, [])
def test_a_core_keeps_the_platform_list(self):
files, _ = self._dests(["sony-playstation"], ["pcsx_rearmed"], [])
self.assertEqual(files, ["pcsx/extra.bin", "scph5500.bin", "scph5501.bin"])
def test_a_region_keeps_one_bios(self):
files, omitted = self._dests([], [], ["north-america"])
self.assertNotIn("scph5500.bin", files)
self.assertIn("scph5501.bin", files)
self.assertEqual(omitted, [])
def test_an_unavailable_match_still_wins_as_the_builder_counts_it(self):
files, omitted = self._dests([], [], ["europe"])
self.assertNotIn("scph5500.bin", files)
self.assertNotIn("scph5501.bin", files)
self.assertEqual(omitted, ["scph5502.bin"])
def test_no_match_keeps_every_regional_file(self):
files, _ = self._dests([], [], ["brazil"])
self.assertIn("scph5500.bin", files)
self.assertIn("scph5501.bin", files)
def test_a_core_variant_competes_across_systems(self):
"""PicoDrive files its US Mega CD BIOS under sega-segacd and the EU
one under sega-mega-cd, one variant group. The pack groups every core
extra before keeping a system, so under --system sega-mega-cd
--region us the US member still wins and the EU one is dropped."""
files = [
{"dest": "us_scd.bin", "size": 1, "cores": ["picodrive"],
"systems": ["sega-segacd"], "regions": ["north-america"],
"region_groups": ["picodrive:variant:mcd"]},
{"dest": "eu_mcd.bin", "size": 1, "cores": ["picodrive"],
"systems": ["sega-mega-cd"], "regions": ["europe"],
"region_groups": ["picodrive:variant:mcd"]},
{"dest": "bios_CD_U.bin", "size": 1, "cores": None,
"systems": ["sega-mega-cd"]},
]
kept, _ = install.narrow(files, [], ["sega-mega-cd"], [], ["north-america"])
self.assertEqual([f["dest"] for f in kept], ["bios_CD_U.bin"])
def test_an_unkept_systems_declaration_does_not_compete(self):
"""Group `a` holds system a's own US file and a core extra filed
under a but owned by `shared`. Under --system shared the pack builds
group a from the extra alone: nothing matches, so the extra stays.
Counting a's own file would make it win and drop the extra."""
files = [
{"dest": "a/us.bin", "size": 1, "cores": None, "systems": ["a"],
"regions": ["north-america"], "region_system_groups": ["a"]},
{"dest": "core/jp.bin", "size": 1, "cores": ["core"],
"systems": ["shared"], "regions": ["japan"], "region_groups": ["a"]},
]
kept, _ = install.narrow(files, [], ["shared"], [], ["north-america"])
self.assertEqual([f["dest"] for f in kept], ["core/jp.bin"])
class CommandLineChoices(unittest.TestCase):
KNOWN = ["msx1,msx2,msxturbor", "nintendo-gba", "sony-playstation"]
def test_comma_lists_and_repeats(self):
self.assertEqual(
install.resolve_choices(["Sony-PlayStation,nintendo-gba"], self.KNOWN, "system"),
["sony-playstation", "nintendo-gba"],
)
def test_a_name_with_commas_is_one_name(self):
self.assertEqual(
install.resolve_choices(["msx1,msx2,msxturbor"], self.KNOWN, "system"),
["msx1,msx2,msxturbor"],
)
def test_an_unknown_name_is_refused(self):
with self.assertRaises(ValueError) as ctx:
install.resolve_choices(["sega-saturn"], self.KNOWN, "system")
self.assertIn("sega-saturn", str(ctx.exception))
self.assertIn("nintendo-gba", str(ctx.exception))
def test_an_empty_list_is_refused(self):
with self.assertRaises(ValueError):
install.resolve_regions([" , "])
with self.assertRaises(ValueError):
install.resolve_choices([","], self.KNOWN, "system")
def test_region_aliases_keep_their_order(self):
self.assertEqual(
install.resolve_regions(["jp,us", "eu,jp"]),
["japan", "north-america", "europe"],
)
with self.assertRaises(ValueError):
install.resolve_regions(["mars"])
def test_prompt_numbers(self):
self.assertEqual(install.parse_selection("3,1-2", 5), [1, 2, 3])
self.assertEqual(install.parse_selection("3,1", 5, ordered=True), [3, 1])
self.assertEqual(install.parse_selection("all", 3), [1, 2, 3])
for bad in ("0", "6", "4-2", "x", "1-"):
with self.subTest(text=bad), self.assertRaises(ValueError):
install.parse_selection(bad, 5)
class TheBoundaryChecksTheNewFields(unittest.TestCase):
def _manifest(self, **extra):
entry = {"dest": "a.bin", "sha1": "a" * 40, "size": 1,
"repo_path": "bios/a.bin", **extra}
return {"manifest_version": 2, "platform": "retroarch", "files": [entry]}
def test_well_formed(self):
install._validate_manifest(
self._manifest(systems=["x"], regions=["japan"], region_groups=["x"]),
"retroarch",
)
def test_malformed(self):
for bad in ({"systems": "x"}, {"regions": [1]}, {"region_groups": [""]},
{"region_system_groups": "x"}):
with self.subTest(field=bad), self.assertRaises(ValueError):
install._validate_manifest(self._manifest(**bad), "retroarch")
class _QuietHandler(http.server.SimpleHTTPRequestHandler):
def log_message(self, format, *args): # noqa: A002 - stdlib signature
return
class FakeRepository:
"""A loopback copy of the repository: one manifest and its payloads."""
PAYLOADS = {
"scph5500.bin": (b"jp bios", ["sony-playstation"], ["japan"], None),
"scph5501.bin": (b"us bios", ["sony-playstation"], ["north-america"], None),
"gba_bios.bin": (b"gba bios", ["nintendo-gba"], None, None),
"pcsx/extra.bin": (b"pcsx", ["sony-playstation"], None, ["pcsx_rearmed"]),
"mgba/extra.bin": (b"mgba", ["nintendo-gba"], None, ["mgba"]),
}
def __init__(self, selection: bool = True):
self.selection = selection
def __enter__(self) -> FakeRepository:
self.root = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp"))
(self.root / "install").mkdir()
(self.root / "bios").mkdir()
files = []
for name, (payload, systems, regions, cores) in self.PAYLOADS.items():
stored = name.replace("/", "_")
(self.root / "bios" / stored).write_bytes(payload)
entry = {
"dest": name,
"sha1": hashlib.sha1(payload).hexdigest(),
"sha256": hashlib.sha256(payload).hexdigest(),
"size": len(payload),
"repo_path": f"bios/{stored}",
"cores": cores,
}
if self.selection:
entry["systems"] = systems
if regions:
entry.update(regions=regions, region_system_groups=systems)
files.append(entry)
manifest = {"manifest_version": 2, "platform": "retroarch", "files": files}
(self.root / "install" / "retroarch.json").write_text(json.dumps(manifest))
handler = functools.partial(_QuietHandler, directory=str(self.root))
self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
self.dest = self.root / "dest"
return self
def __exit__(self, *exc) -> None:
self.httpd.shutdown()
self.httpd.server_close()
shutil.rmtree(self.root, ignore_errors=True)
def env(self) -> dict:
env = dict(os.environ)
env["RETROBIOS_BASE_URL"] = f"http://127.0.0.1:{self.httpd.server_address[1]}"
return env
def command(self, *options: str) -> list[str]:
return [sys.executable, str(REPO_ROOT / "install.py"), "--platform",
"retroarch", "--dest", str(self.dest), *options]
def run(self, *options: str) -> subprocess.CompletedProcess:
return subprocess.run(
self.command(*options), env=self.env(), capture_output=True,
text=True, timeout=60, check=False, stdin=subprocess.DEVNULL,
)
def installed(self) -> list[str]:
if not self.dest.exists():
return []
return sorted(
p.relative_to(self.dest).as_posix()
for p in self.dest.rglob("*") if p.is_file()
)
class TheOneLinerNarrows(unittest.TestCase):
def test_options_install_only_the_choice(self):
with FakeRepository() as repo:
proc = repo.run("--system", "sony-playstation", "--region", "us")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertIn("Narrowed 5 -> 2 files", proc.stdout)
self.assertEqual(repo.installed(), ["pcsx/extra.bin", "scph5501.bin"])
def test_a_core_choice_keeps_the_platform_list(self):
with FakeRepository() as repo:
proc = repo.run("--core", "mgba")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(
repo.installed(),
["gba_bios.bin", "mgba/extra.bin", "scph5500.bin", "scph5501.bin"],
)
def test_an_older_file_list_is_refused_not_ignored(self):
"""Without the fields, a region filter would match nothing and the
run would install everything while the user asked for one region."""
with FakeRepository(selection=False) as repo:
proc = repo.run("--region", "us")
self.assertEqual(proc.returncode, 1)
self.assertIn("records no systems or regions", proc.stderr)
self.assertEqual(repo.installed(), [])
def test_empty_names_are_a_usage_error(self):
with FakeRepository() as repo:
for option in (["--region", " , "], ["--system", ""], ["--core", ","]):
with self.subTest(option=option):
proc = repo.run(*option)
self.assertEqual(proc.returncode, 2, proc.stderr)
self.assertIn("needs at least one", proc.stderr)
def test_listing_follows_the_narrowing(self):
with FakeRepository() as repo:
proc = repo.run("--list-cores", "--system", "nintendo-gba")
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("mgba", proc.stdout)
self.assertNotIn("pcsx_rearmed", proc.stdout)
def test_an_unknown_system_is_refused_before_any_download(self):
with FakeRepository() as repo:
proc = repo.run("--system", "sega-saturn")
self.assertEqual(proc.returncode, 1)
self.assertIn("unknown system: sega-saturn", proc.stderr)
self.assertIn("nintendo-gba", proc.stderr)
self.assertEqual(repo.installed(), [])
def test_an_unknown_region_is_a_usage_error(self):
with FakeRepository() as repo:
proc = repo.run("--region", "mars")
self.assertEqual(proc.returncode, 2)
self.assertIn("unknown region: mars", proc.stderr)
def test_listing_systems_installs_nothing(self):
with FakeRepository() as repo:
proc = repo.run("--list-systems")
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("nintendo-gba", proc.stdout)
self.assertIn("3 files", proc.stdout)
self.assertEqual(repo.installed(), [])
def test_no_terminal_asks_nothing_and_installs_everything(self):
with FakeRepository() as repo:
proc = repo.run()
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertNotIn("choose systems", proc.stdout)
self.assertEqual(len(repo.installed()), 5)
@unittest.skipUnless(shutil.which("script"), "script(1) provides the terminal")
def test_the_custom_mode_at_a_terminal(self):
"""Typed answers: customise, PlayStation only, no core, US first."""
with FakeRepository() as repo:
command = " ".join(shlex.quote(part) for part in repo.command())
proc = subprocess.run(
["script", "-qec", command, "/dev/null"],
input="c\n2\n\n2\n", env=repo.env(), capture_output=True,
text=True, timeout=60, check=False,
)
out = proc.stdout
self.assertEqual(proc.returncode, 0, out + proc.stderr)
self.assertIn("choose systems, cores and regions", out)
self.assertIn("pcsx_rearmed", out)
self.assertIn("--system sony-playstation --region north-america", out)
self.assertEqual(repo.installed(), ["pcsx/extra.bin", "scph5501.bin"])
@unittest.skipUnless(shutil.which("script"), "script(1) provides the terminal")
def test_enter_installs_everything_at_a_terminal(self):
with FakeRepository() as repo:
command = " ".join(shlex.quote(part) for part in repo.command())
proc = subprocess.run(
["script", "-qec", command, "/dev/null"],
input="\n", env=repo.env(), capture_output=True,
text=True, timeout=60, check=False,
)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertEqual(len(repo.installed()), 5)
@unittest.skipUnless(shutil.which("script"), "script(1) provides the terminal")
def test_no_input_asks_nothing_at_a_terminal(self):
with FakeRepository() as repo:
command = " ".join(
shlex.quote(part) for part in repo.command("--no-input")
)
proc = subprocess.run(
["script", "-qec", command, "/dev/null"],
input="", env=repo.env(), capture_output=True,
text=True, timeout=60, check=False,
)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertNotIn("choose systems", proc.stdout)
self.assertEqual(len(repo.installed()), 5)
class InstallMatchesThePack(unittest.TestCase):
"""Narrowing the full manifest gives what the narrowed build carries."""
PLATFORM = "recalbox"
@classmethod
def setUpClass(cls):
db_path = REPO_ROOT / "database.json"
if not db_path.exists():
raise unittest.SkipTest("database.json is not built")
import generate_pack as gp
from common import load_emulator_profiles
cls.gp = gp
cls.db = json.loads(db_path.read_text(encoding="utf-8"))
cls.profiles = load_emulator_profiles(str(REPO_ROOT / "emulators"))
cls.full = cls._manifest(None)
@classmethod
def _manifest(cls, regions):
return cls.gp.generate_manifest(
cls.PLATFORM, str(REPO_ROOT / "platforms"), cls.db,
str(REPO_ROOT / "bios"), str(REPO_ROOT / "platforms" / "_registry.yml"),
emulators_dir=str(REPO_ROOT / "emulators"), emu_profiles=cls.profiles,
regions=regions, offline=True,
)
def _narrowed(self, systems, regions):
files, omitted = install.narrow(
self.full["files"], self.full["omitted_files"], systems, [], regions
)
return {f["dest"] for f in files}, {o["dest"] for o in omitted}
def test_regions(self):
for regions in (["north-america"], ["japan", "europe"]):
with self.subTest(regions=regions):
built = self._manifest(regions)
files, omitted = self._narrowed([], regions)
self.assertEqual(files, {f["dest"] for f in built["files"]})
self.assertEqual(omitted, {o["dest"] for o in built["omitted_files"]})
class InstallMatchesASystemPack(unittest.TestCase):
"""`install.py --system X --region R` against a real `generate_pack.py
--system X --region R` build. The cases are the ones a review reproduced:
ColecoVision's extras filed at destinations another system declares, and
PicoDrive's Mega CD variant group spread over two systems."""
PLATFORM = "retroarch"
CASES = (
(["coleco-colecovision"], []),
(["sega-mega-cd"], ["north-america"]),
(["sony-playstation"], ["japan"]),
)
@classmethod
def setUpClass(cls):
db_path = REPO_ROOT / "database.json"
if not db_path.exists():
raise unittest.SkipTest("database.json is not built")
import generate_pack as gp
from common import load_emulator_profiles, load_platform_config
cls.gp = gp
cls.db = json.loads(db_path.read_text(encoding="utf-8"))
cls.profiles = load_emulator_profiles(str(REPO_ROOT / "emulators"))
cls.base = load_platform_config(
cls.PLATFORM, str(REPO_ROOT / "platforms")
).get("base_destination", "")
cls.full = gp.generate_manifest(
cls.PLATFORM, str(REPO_ROOT / "platforms"), cls.db,
str(REPO_ROOT / "bios"), str(REPO_ROOT / "platforms" / "_registry.yml"),
emulators_dir=str(REPO_ROOT / "emulators"), emu_profiles=cls.profiles,
offline=True,
)
def _packed(self, systems, regions) -> set[str]:
import zipfile
with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp") as out:
zip_path = self.gp.generate_pack(
self.PLATFORM, str(REPO_ROOT / "platforms"), self.db,
str(REPO_ROOT / "bios"), out,
emulators_dir=str(REPO_ROOT / "emulators"),
emu_profiles=self.profiles, system_filter=systems,
regions=regions or None, offline=True,
)
with zipfile.ZipFile(zip_path) as zf:
names = set(zf.namelist())
prefix = f"{self.base}/" if self.base else ""
return {n[len(prefix):] if prefix and n.startswith(prefix) else n for n in names}
def test_each_case(self):
known = {f["dest"] for f in self.full["files"]}
for systems, regions in self.CASES:
with self.subTest(systems=systems, regions=regions):
files, _omitted = install.narrow(
self.full["files"], self.full["omitted_files"],
systems, [], regions,
)
self.assertEqual(
{f["dest"] for f in files}, self._packed(systems, regions) & known
)
if __name__ == "__main__":
unittest.main()