"""The PortMaster port: its archive, its event reader and its screens. The screens are driven here with PortMasterDialog's functions replaced by stubs that record each command and answer from a script, and with a fake install.py that writes canned --events streams. What a device adds (pugwash drawing, the pad, mount, the network) is not exercised. """ from __future__ import annotations import hashlib import importlib.util import io import json import os import shutil import subprocess import sys import tempfile import unittest import zipfile from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent PORT = REPO_ROOT / "ports" / "RetroBIOS" sys.path.insert(0, str(REPO_ROOT / "scripts")) import build_port # 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) _spec = importlib.util.spec_from_file_location( "rbevents", PORT / "retrobios" / "rbevents.py") rbevents = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(rbevents) SCRIPTS = [PORT / "RetroBIOS.sh", PORT / "retrobios" / "screens.txt"] def _tempdir(test: unittest.TestCase) -> Path: scratch = REPO_ROOT / "tmp" scratch.mkdir(exist_ok=True) path = Path(tempfile.mkdtemp(dir=str(scratch))) test.addCleanup(shutil.rmtree, path, True) return path class EveryScriptParses(unittest.TestCase): def test_bash_n(self): for script in SCRIPTS: with self.subTest(script=script.name): proc = subprocess.run(["bash", "-n", str(script)], capture_output=True, text=True, check=False) self.assertEqual(proc.returncode, 0, proc.stderr) def test_only_dialog_functions_the_oldest_releases_have(self): """PortMaster 2024.02.26 and 8.5.23 define these in PortMasterDialog.txt, and have no funcs.txt: pm_message and the rest would be undefined.""" known = {"PortMasterDialogInit", "PortMasterDialog", "PortMasterDialogResult", "PortMasterDialogExit", "PortMasterIPCheck"} commands = {"register_set_info", "register_clear", "selection_list", "messages_begin", "message", "messages_end", "progress", "progress_clear", "check_runtime"} text = "\n".join(path.read_text(encoding="utf-8") for path in SCRIPTS) import re # noqa: PLC0415 used = set(re.findall(r"\b(PortMasterDialog\w*|PortMasterIPCheck)\b", text)) self.assertLessEqual(used - {"PortMasterDialog.txt"}, known) self.assertNotRegex(text, r"\bpm_message\b|\bpm_platform_helper\b") sent = set(re.findall(r'(?:PortMasterDialog|rb_answer) "(\w+)"', text)) self.assertLessEqual(sent, commands) self.assertNotIn("register_clear_info", text) class ThePortArchive(unittest.TestCase): def setUp(self): self.out = _tempdir(self) def _build(self, name: str = "RetroBIOS.zip") -> Path: return build_port.build_port(self.out / name) def test_layout_portmaster_installs(self): """One launch script at the top, one folder holding port.json: what harbourmaster's captain.check_port accepts without a warning.""" with zipfile.ZipFile(self._build()) as zf: names = zf.namelist() port = json.loads(zf.read("retrobios/port.json")) mode = zf.getinfo("RetroBIOS.sh").external_attr >> 16 self.assertEqual(names, sorted(names)) self.assertEqual([n for n in names if "/" not in n], ["RetroBIOS.sh"]) self.assertEqual({n.split("/")[0] for n in names if "/" in n}, {"retrobios"}) self.assertEqual([n for n in names if n.endswith(".sh")], ["RetroBIOS.sh"]) for member in ("retrobios/install.py", "retrobios/rbevents.py", "retrobios/screens.txt", "retrobios/gameinfo.xml", "retrobios/retrobios.md", "retrobios/licenses/LICENSE-retrobios.txt"): self.assertIn(member, names) self.assertEqual(port["version"], 4) self.assertEqual(port["name"], "retrobios.zip") self.assertEqual(port["items"], ["RetroBIOS.sh", "retrobios"]) self.assertEqual(port["attr"]["genres"], ["other"]) self.assertTrue(mode & 0o111, "the launch script is executable") def test_bundled_installer_is_the_pinned_one(self): with zipfile.ZipFile(self._build()) as zf: bundled = hashlib.sha256(zf.read("retrobios/install.py")).hexdigest() self.assertEqual(bundled, build_port.pinned_installer_sha256()) install_sh = (REPO_ROOT / "install.sh").read_text(encoding="utf-8") self.assertIn(f'DEFAULT_INSTALL_SHA256="{bundled}"', install_sh) def test_an_unpinned_installer_is_refused(self): root = _tempdir(self) for name in ("install.py", "install.sh", "LICENSE", "NOTICE"): shutil.copy(REPO_ROOT / name, root / name) shutil.copytree(PORT, root / "ports" / "RetroBIOS") with (root / "install.py").open("a") as fh: fh.write("\n") with self.assertRaises(ValueError): build_port.build_port(self.out / "x.zip", root) def test_reproducible(self): first = self._build("a.zip").read_bytes() second = self._build("b.zip").read_bytes() self.assertEqual(first, second) class EventRecords(unittest.TestCase): """rbevents turns --events into fields the FIFO carries intact.""" def test_fields_are_fifo_safe(self): self.assertEqual(rbevents.clean("a\tb\n[x]*? c"), "a b x c") out = io.StringIO() rbevents.record(out, "file", "", "a") self.assertEqual(out.getvalue(), "file\t-\ta\n") def test_follow_writes_the_outcome(self): stream = [ json.dumps({"event": "download", "platform": "p", "files": 1, "bytes": 2048}), "not json", json.dumps({"event": "downloaded", "dest": "x.bin", "ok": False, "reason": "HTTP 404", "size": 2048, "files_done": 1, "files_total": 1, "bytes_done": 0, "bytes_total": 2048}), json.dumps({"event": "summary", "mode": "install", "installed": 0, "up_to_date": 4, "pending": 1, "failed": [{"dest": "x.bin", "reason": "HTTP 404"}], "skipped": [], "left_out": [], "kept": [{"folder": "/b/retrobios-backup/1", "files": 1, "bytes": 2048}], "interrupted": False}), json.dumps({"event": "end", "code": 2}), ] out, summary = io.StringIO(), io.StringIO() rbevents.follow(stream, out, summary) self.assertEqual(out.getvalue().splitlines(), [ "download\t1\t2 KB", "file\t1\t1\tfailed\tx.bin\t0 B\t2 KB"]) self.assertEqual(summary.getvalue().splitlines(), [ "summary\t0\t4\t1\t0", "failed\tx.bin\tHTTP 404", "kept\t/b/retrobios-backup/1\t1\t2 KB", "end\t2"]) def test_a_system_id_that_is_no_slug_is_dropped(self): # The id becomes a bash array subscript in the launcher, where a # command substitution in it would run. def system(name): return json.dumps({"event": "system", "platform": "p", "system": name, "name": "N", "files": 1, "present": 0, "missing": 1, "wrong_hash": 0, "bytes_needed": 1}) stream = [ json.dumps({"event": "detect", "os": "linux", "os_name": "X", "platforms": [{"platform": "p", "name": "P", "bios_dir": "/b"}]}), system("$(touch pwned)"), system("a`id`"), system("sony-playstation"), ] out = io.StringIO() rbevents.status(stream, out) ids = [line.split("\t")[1] for line in out.getvalue().splitlines() if line.startswith("system\t")] self.assertEqual(ids, ["sony-playstation"]) def test_launcher_never_double_expands_a_subscript(self): text = (PORT / "retrobios" / "screens.txt").read_text() self.assertNotRegex(text, r'unset\s+"[^"]*\[\$') def test_a_run_without_end_says_so(self): out = io.StringIO() rbevents.status([], out) self.assertEqual(out.getvalue(), "end\tnone\n") FAKE_INSTALL = r''' import json, os, sys from pathlib import Path state = Path(os.environ["FAKE_STATE"]) scenario = os.environ["FAKE_SCENARIO"] keys = json.loads(os.environ["FAKE_KEYS"]) with (state / "calls.txt").open("a") as fh: fh.write(" ".join(sys.argv[1:]) + "\n") check = "--check" in sys.argv def emit(kind, **fields): assert sorted(fields) == sorted(keys[kind]), kind print(json.dumps({"event": kind, **fields}), flush=True) def end(code): emit("end", code=code) sys.exit(code) emit("start", schema=1, installer="fake", mode="check" if check else "install") if scenario == "nodetect": emit("detect", os="linux", os_name="Fake OS 1.0", platforms=[]) end(1) if scenario == "conflict": emit("error", stage="detect", reason="conflict", platform="", message="JELOS and Koriki are both detected here, and a machine runs one") end(1) emit("detect", os="linux", os_name="Fake OS 1.0", platforms=[ {"platform": "retroarch", "name": "RetroArch", "bios_dir": "/roms/bios"}]) if scenario == "offline": emit("error", stage="manifest", reason="network", platform="retroarch", message="the file list could not be fetched: [Errno 101] Network is unreachable") end(1) done = scenario == "complete" or (state / "installed").exists() if check: for system, name, files, present, missing, wrong, needed in ( ("sys-a", "Alpha*", 2, 2 if done else 0, 0 if done else 1, 0 if done else 1, 0 if done else 300), ("sys-b", "Bravo", 1, 1, 0, 0, 0), ): emit("system", platform="retroarch", system=system, name=name, files=files, present=present, missing=missing, wrong_hash=wrong, bytes_needed=needed) emit("check", platform="retroarch", name="RetroArch", bios_dir="/roms/bios", files=3, present=3 if done else 1, missing=0 if done else 1, wrong_hash=0 if done else 1, bytes_needed=0 if done else 300, data_missing=0, not_installable=1) emit("summary", mode="check", installed=0, up_to_date=3 if done else 1, pending=0 if done else 2, failed=[], skipped=[], left_out=[], kept=[], interrupted=False) end(0 if done else 1) emit("download", platform="retroarch", files=2, bytes=300) for n, (dest, size) in enumerate((("a.bin", 100), ("b.bin", 200)), 1): emit("downloaded", dest=dest, ok=True, reason="", size=size, files_done=n, files_total=2, bytes_done=100 if n == 1 else 300, bytes_total=300) (state / "installed").write_text("") emit("summary", mode="install", installed=2, up_to_date=1, pending=0, failed=[], skipped=[{"reason": "not_found", "label": "not found", "files": 1, "required": 1}], left_out=[], kept=[{"folder": "/roms/retrobios-backup/20261022-153000", "files": 1, "bytes": 100}], interrupted=False) end(0) ''' # PortMasterDialog.txt replaced: every call is a trace line, fields split by # \x1f, and each selection list is answered from the answers file. DRIVER = r''' trace() { printf '%s' "$1" shift local field for field in "$@"; do printf '\x1f%s' "$field"; done printf '\n' } >> "$RB_TRACE" PM_PID="" PM_DONE="$FAKE_STATE/pm_done" PortMasterDialogInit() { trace INIT "$@"; } PortMasterDialogExit() { trace EXIT; } PortMasterIPCheck() { [ -n "$FAKE_IP" ] && echo 192.168.0.1; } PortMasterDialog() { trace CMD "$@"; } PortMasterDialogResult() { trace ASK "$@" if [ "$1" = "check_runtime" ]; then echo "${FAKE_RUNTIME:-FAIL}" return fi local answer answer=$(head -n 1 "$RB_ANSWERS") sed -i 1d "$RB_ANSWERS" case "$answer" in "") echo CANCEL ;; WAIT:*) printf '%s' "${answer#WAIT:}" > "$PM_DONE"; echo WAIT ;; *) echo "$answer" ;; esac } source "$GAMEDIR/screens.txt" if [ -n "$FAKE_ARCH" ]; then rb_arch() { echo "$FAKE_ARCH"; } fi rb_main trace STATUS "$?" ''' class TheScreens(unittest.TestCase): """Which screens the port shows, scenario by scenario.""" def run_port(self, scenario: str, answers: list[str], **env: str) -> list[tuple]: work = _tempdir(self) (work / "install.py").write_text(FAKE_INSTALL) (work / "driver.sh").write_text(DRIVER) (work / "answers").write_text("".join(f"{a}\n" for a in answers)) bin_dir = work / "bin" bin_dir.mkdir() # The oldest Python install.py supports, where it is installed: the # event reader runs on whatever the device has. (bin_dir / "python3").symlink_to(shutil.which("python3.8") or sys.executable) self.work = work full_env = { **os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}", "HOME": str(work), "GAMEDIR": str(PORT / "retrobios"), "controlfolder": str(work / "PortMaster"), "ESUDO": "", "RB_STATE": str(work / "run"), "RB_LOG": str(work / "log.txt"), "RB_INSTALL": str(work / "install.py"), "RB_TRACE": str(work / "trace"), "RB_ANSWERS": str(work / "answers"), "FAKE_STATE": str(work), "FAKE_SCENARIO": scenario, "FAKE_KEYS": json.dumps(install.EVENT_KEYS), **env, } proc = subprocess.run(["bash", str(work / "driver.sh")], env=full_env, capture_output=True, text=True, timeout=120, check=False) self.assertEqual(proc.returncode, 0, proc.stderr) log = work / "log.txt" self.assertNotIn("Traceback", log.read_text() if log.exists() else "") return self._screens((work / "trace").read_text(encoding="utf-8")) def _screens(self, trace: str) -> list[tuple]: """INIT, EXIT, STATUS, each list shown with its entries, and the progress and message commands, in order.""" registers: dict[str, dict[str, tuple[str, str]]] = {} shown: list[tuple] = [] for line in trace.splitlines(): kind, *fields = line.split("\x1f") for field in fields: self.assertNotRegex(field, r"[*?\[\]\t]", line) if kind == "CMD" and fields[0] == "register_clear": registers[fields[1]] = {} elif kind == "CMD" and fields[0] == "register_set_info": info = dict(item.split(":", 1) for item in fields[3:]) registers.setdefault(fields[1], {})[fields[2]] = ( info["title"], info["description"]) elif kind == "ASK" and fields[0] == "selection_list": self.assertIn("--want-cancel", fields) name = fields[1].split("=", 1)[1] shown.append(("list", name, dict(registers[name]))) elif kind == "ASK": shown.append(("ask", *fields)) elif kind == "CMD" and fields[0] in ("progress", "message"): shown.append(("cmd", *fields)) elif kind in ("INIT", "EXIT", "STATUS"): shown.append((kind, *fields)) for screen in shown: if screen[0] == "list": titles = [title for title, _desc in screen[2].values()] self.assertTrue({"Back", "Quit", "Back to the menu"} & set(titles), f"{screen[1]} has no Back or Quit entry") return shown def lists(self, shown: list[tuple]) -> list[tuple[str, list[str]]]: return [(s[1], list(s[2])) for s in shown if s[0] == "list"] def calls(self) -> list[str]: return (self.work / "calls.txt").read_text().splitlines() def test_all_present(self): shown = self.run_port("complete", ["quit"]) self.assertEqual(self.lists(shown), [("menu", ["status", "check", "quit"])]) menu = shown[[s[0] for s in shown].index("list")][2] self.assertEqual(menu["status"][0], "Fake OS 1.0: all 3 BIOS files present") self.assertEqual(shown[0], ("INIT", "no-harbour")) self.assertEqual(shown[-2:], [("EXIT",), ("STATUS", "0")]) self.assertEqual(self.calls(), ["--check --events"]) def test_some_missing_update_everything(self): shown = self.run_port("missing", ["update", "back", "quit"]) self.assertEqual(self.lists(shown), [ ("menu", ["status", "update", "choose", "check", "quit"]), ("summary", ["info:result", "back", "quit", "info:kept:1", "info:skipped:not found"]), ("menu", ["status", "check", "quit"]), ]) menu = shown[[s[0] for s in shown].index("list")][2] self.assertEqual(menu["status"][0], "Fake OS 1.0: 1/3 BIOS files, 1 missing, 1 wrong hash") self.assertIn("Downloads 2 files, 300 B, to /roms/bios", menu["update"][1]) progress = [s for s in shown if s[:2] == ("cmd", "progress")] self.assertEqual(progress, [ ("cmd", "progress", "Starting", "0", "1"), ("cmd", "progress", "a.bin, 100 B of 300 B", "1", "2"), ("cmd", "progress", "b.bin, 300 B of 300 B", "2", "2"), ]) summary = [s for s in shown if s[:2] == ("list", "summary")][0][2] self.assertEqual(summary["info:result"][0], "Installed: 2 files") self.assertEqual(summary["info:kept:1"][0], "Kept 1 file you had (100 B)") self.assertIn("/roms/retrobios-backup/20261022-153000", summary["info:kept:1"][1]) self.assertEqual(self.calls(), [ "--check --events", "--events --platform retroarch", "--check --events --platform retroarch", ]) def test_some_missing_choose_systems(self): shown = self.run_port("missing", [ "choose", "update", "sys:sys-a", "sys:sys-a", "sys:sys-a", "update", "quit"]) systems = [s[2] for s in shown if s[:2] == ("list", "systems")] self.assertEqual(list(systems[0]), ["update", "back", "sys:sys-a"]) self.assertEqual(systems[0]["sys:sys-a"][0], "Alpha") self.assertEqual(systems[1]["sys:sys-a"][0], "Alpha") self.assertEqual(systems[2]["sys:sys-a"][0], "Alpha (selected)") self.assertEqual(systems[2]["update"][0], "Update the 1 selected") self.assertIn("1 system complete, not listed", systems[0]["back"][1]) self.assertEqual(self.calls()[1], "--events --platform retroarch --system sys-a") self.assertEqual(self.lists(shown)[-1][0], "summary") self.assertEqual(shown[-1], ("STATUS", "0")) def test_no_network(self): shown = self.run_port("offline", ["quit"]) self.assertEqual(self.lists(shown), [("notice", ["info", "check", "quit"])]) notice = shown[[s[0] for s in shown].index("list")][2] self.assertEqual(notice["info"][0], "No connection") self.assertIn("Connect to Wi-Fi", notice["info"][1]) self.assertIn("/roms/bios", notice["info"][1]) def test_detection_failure(self): shown = self.run_port("nodetect", ["check", "quit"]) self.assertEqual([name for name, _keys in self.lists(shown)], ["notice", "notice"]) notice = shown[[s[0] for s in shown].index("list")][2] self.assertEqual(notice["info"][0], "No BIOS folder found") self.assertIn("Fake OS 1.0", notice["info"][1]) self.assertIn("--dest", notice["info"][1]) self.assertEqual(len(self.calls()), 2) def test_two_systems_found(self): shown = self.run_port("conflict", ["check", "quit"]) notice = shown[[s[0] for s in shown].index("list")][2] self.assertEqual(notice["info"][0], "Two systems found") self.assertIn("JELOS and Koriki", notice["info"][1]) self.assertIn("--dest", notice["info"][1]) def test_python_too_old(self): for arch, env, words in ( ("armv7l", {}, ["3.7.3", "aarch64 only", "armv7l", "--dest"]), ("aarch64", {}, ["3.7.3", "needs a connection", "--dest"]), ("aarch64", {"FAKE_IP": "1", "FAKE_RUNTIME": "FAIL"}, ["3.7.3", "could not be installed", "--dest"]), ): with self.subTest(arch=arch, env=env): shown = self.run_old(arch, **env) lists = self.lists(shown) self.assertEqual(lists, [("notice", ["info", "quit"])]) notice = [s for s in shown if s[0] == "list"][0][2] self.assertEqual(notice["info"][0], "Python 3.8 or newer is needed") for word in words: self.assertIn(word, notice["info"][1]) self.assertEqual(shown[-1], ("STATUS", "1")) self.assertFalse((self.work / "calls.txt").exists()) asked_runtime = ("ask", "check_runtime", "python_3.11.squashfs") in shown self.assertEqual(asked_runtime, bool(env.get("FAKE_RUNTIME"))) def run_old(self, arch: str, **env: str) -> list[tuple]: """A python3 on PATH that is 3.7.3, ahead of every other.""" holder = _tempdir(self) fake = holder / "python3" fake.write_text('#!/bin/sh\n[ "$1" = "--version" ] && echo "Python 3.7.3"\nexit 1\n') fake.chmod(0o755) return self.run_port("complete", [], FAKE_ARCH=arch, PATH=f"{holder}{os.pathsep}{os.environ['PATH']}", **env) def test_an_answer_after_the_old_30_second_wait(self): """2024.02.26 and 8.5.23 stop waiting after 30 s and answer WAIT: the port reads the answer pugwash writes later.""" shown = self.run_port("complete", ["WAIT:quit"]) self.assertEqual(shown[-1], ("STATUS", "0")) self.assertEqual(len(self.lists(shown)), 1) if __name__ == "__main__": unittest.main()