import json import os from pathlib import Path import struct import subprocess import sys import tempfile import time import unittest import unittest.mock sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import title_run class TitlesTests(unittest.TestCase): def setUp(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.root = Path(directory.name) def write(self, data): path = self.root / "titles.json" path.write_text(json.dumps(data), encoding="utf-8") return path def test_repository_file_is_valid(self): titles = title_run.load_titles(title_run.ROOT / "tools/titles.json") self.assertEqual(titles["PPSA85300"], {"relink": ["--registry"]}) def test_unknown_field_is_refused(self): with self.assertRaisesRegex(ValueError, "PPSA00001: unknown field rlink"): title_run.load_titles(self.write({"PPSA00001": {"rlink": []}})) def test_wrong_type_is_refused(self): with self.assertRaisesRegex(ValueError, "PPSA00001: seconds must be int"): title_run.load_titles(self.write({"PPSA00001": {"seconds": "60"}})) def test_non_string_items_are_refused(self): with self.assertRaisesRegex(ValueError, "PPSA00001: env must hold strings"): title_run.load_titles(self.write({"PPSA00001": {"env": {"APS5_X": 1}}})) with self.assertRaisesRegex(ValueError, "PPSA00001: relink must hold strings"): title_run.load_titles(self.write({"PPSA00001": {"relink": [1]}})) def test_pad_timeline_is_checked(self): events = [{"at_ms": 0, "hold_ms": 200, "buttons": ["cross"]}, {"at_ms": 400, "hold_ms": 300, "buttons": ["right", "cross"]}] titles = title_run.load_titles(self.write({"PPSA00001": {"pad": {"after_s": 3, "repeat": 2, "events": events}}})) self.assertEqual(titles["PPSA00001"]["pad"]["repeat"], 2) for pad in ({"after_s": 3}, {"events": events, "delay": 3}, {"events": [{"at_ms": 0, "hold_ms": 200, "buttons": ["jump"]}]}, {"events": events, "after_s": 3601}, {"events": events, "repeat": 0}): with self.assertRaises(ValueError): title_run.load_titles(self.write({"PPSA00001": {"pad": pad}})) def test_bad_id_is_refused(self): with self.assertRaisesRegex(ValueError, "NOT-AN-ID is not a title ID"): title_run.load_titles(self.write({"NOT-AN-ID": {}})) def test_name_is_refused(self): with self.assertRaisesRegex(ValueError, "PPSA00001: unknown field name"): title_run.load_titles(self.write({"PPSA00001": {"name": "Light"}})) def test_default_entry(self): self.assertEqual(title_run.relink_args({}, "windows"), ["--windows"]) self.assertEqual(title_run.relink_args({}, "linux"), []) def test_entry_on_windows_and_linux(self): entry = {"relink": ["--registry"], "fixed_base": "0x400000", "executables": ["picker/picker.bin"]} self.assertEqual(title_run.relink_args(entry, "windows"), ["--windows", "--registry", "--windows-fixed-base", "0x400000", "--executable", "picker/picker.bin"]) self.assertEqual(title_run.relink_args(entry, "linux"), ["--registry", "--linux-fixed-base", "0x400000", "--executable", "picker/picker.bin"]) def test_fixed_base_on_macos_is_refused(self): with self.assertRaisesRegex(ValueError, "fixed_base is not supported on macos"): title_run.relink_args({"fixed_base": "0x400000"}, "macos") def bmp(path, width, height, color): rows = b"".join(bytes(channel for x in range(width) for channel in (*color(x, y), 255)) for y in range(height)) header = struct.pack("<2sIHHI", b"BM", 54 + len(rows), 0, 0, 54) info = struct.pack(" crashed", "", "| Title | ID | Version | Cold | Warm | FPS | First error | Before |", "|---|---|---|---|---|---|---|---|", "| Light | PPSA85300 | 0.4 | crashed | rendering | 31.0 | what(): unsupported MIB 1.1 \\| x | rendering |", "", ])) self.assertNotIn("Before", title_run.render_markdown(now)) now["machine"] = {"os": "", "gpu": "", "driver": "", "cpu": "Ryzen AI 7 350"} self.assertTrue(title_run.render_markdown(now).startswith("AnyPS5 `abc1234`, main + #6814. CPU Ryzen AI 7 350. 2 runs")) def test_render_markdown_skips_runs_without_an_error(self): now = summary({"PPSA01325": {"name": "Playroom", "version": "01.905.000", "runs": [run("stalled"), run("crashed", "warm", error="vkQueueSubmit recorder: Vulkan result -4")]}}) self.assertIn("| stalled | crashed | 0.0 | vkQueueSubmit recorder: Vulkan result -4 |", title_run.render_markdown(now)) def test_render_markdown_lists_missing_titles(self): now = summary({"PPSA85300": {"name": "Light", "version": "0.4", "runs": [run("rendering"), run("rendering", "warm")]}}) before = summary({"PPSA85300": {"name": "Light", "version": "0.4", "runs": [run("rendering"), run("rendering", "warm")]}, "PPSA97358": {"name": "XPSemu", "version": "A2.1", "runs": [run("stalled"), run("stalled", "warm")]}}) self.assertIn("Missing: XPSemu (PPSA97358)", title_run.render_markdown(now, before)) def package(root, title_id, version="01.000.000", name="Game"): (root / "sce_sys").mkdir(parents=True) (root / "eboot.bin").write_bytes(b"\x7fELF" + bytes(60)) params = {"titleId": title_id, "contentVersion": version, "localizedParameters": {"defaultLanguage": "en-US", "en-US": {"titleName": name}}} (root / "sce_sys/param.json").write_text(json.dumps(params), encoding="utf-8") return root class PreparationTests(unittest.TestCase): def setUp(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.root = Path(directory.name) def test_a_running_title_is_left_alone(self): package(self.root / "games/PPSA00001", "PPSA00001") out = self.root / "out" marker = out / "work/PPSA00001/app0/eboot.bin" marker.parent.mkdir(parents=True) marker.write_bytes(b"running") held = title_run.lock_work(out / "work/PPSA00001.lock") self.addCleanup(held.close) self.assertIsNotNone(held) with self.assertRaisesRegex(SystemExit, "already running in .*: PPSA00001"): title_run.main(["--games", str(self.root / "games"), "--out", str(out), "PPSA00001"]) self.assertEqual(marker.read_bytes(), b"running") held.close() again = title_run.lock_work(out / "work/PPSA00001.lock") self.assertIsNotNone(again) again.close() def test_discover(self): package(self.root / "a/src/PPSA00001", "PPSA00001", "01.000.110", "Light") package(self.root / "b/PPSA00002", "PPSA00002") found = title_run.discover([self.root]) self.assertEqual(sorted(found), ["PPSA00001", "PPSA00002"]) self.assertEqual(found["PPSA00001"]["version"], "01.000.110") self.assertEqual(found["PPSA00001"]["name"], "Light") def test_source_root_follows_the_build(self): build = self.root / "build" build.mkdir() self.assertEqual(title_run.source_root(build), title_run.ROOT) (build / "CMakeCache.txt").write_text("CMAKE_BUILD_TYPE:STRING=Release\nCMAKE_HOME_DIRECTORY:INTERNAL=C:/src/AnyPS5\n", encoding="utf-8") self.assertEqual(title_run.source_root(build), Path("C:/src/AnyPS5")) def test_discover_refuses_duplicates(self): package(self.root / "a/PPSA00001", "PPSA00001") package(self.root / "b/PPSA00001", "PPSA00001") with self.assertRaisesRegex(ValueError, "PPSA00001 found more than once"): title_run.discover([self.root]) def test_discover_refuses_a_package_without_title_id(self): package(self.root / "a/PPSA00001", "PPSA00001") param = self.root / "a/PPSA00001/sce_sys/param.json" param.write_text(json.dumps({"contentVersion": "01.000.000"}), encoding="utf-8") with self.assertRaisesRegex(ValueError, "param.json: no titleId"): title_run.discover([self.root]) def test_discover_refuses_a_title_id_that_is_not_one(self): for index, title_id in enumerate(("../../outside", str(self.root / "elsewhere"), "PPSA0001")): package(self.root / str(index) / "PPSA00001", title_id) with self.assertRaisesRegex(ValueError, "is not a title ID"): title_run.discover([self.root / str(index)]) def test_unwrap_or_copy_keeps_an_elf(self): source = self.root / "eboot.bin" source.write_bytes(b"\x7fELF" + bytes(60)) title_run.unwrap_or_copy(source, self.root / "out/eboot.elf") self.assertEqual((self.root / "out/eboot.elf").read_bytes(), source.read_bytes()) def test_link_tree_skips(self): source = package(self.root / "pkg", "PPSA00001") (source / "sce_module").mkdir() (source / "sce_module/libc.prx").write_bytes(b"x") title_run.link_tree(source, self.root / "app0", {"sce_module"}) self.assertTrue((self.root / "app0/sce_sys/param.json").is_file()) self.assertTrue((self.root / "app0/eboot.bin").is_file()) self.assertFalse((self.root / "app0/sce_module").exists()) def wait_until(condition, seconds=10): deadline = time.monotonic() + seconds while time.monotonic() < deadline: if condition(): return True time.sleep(0.1) return False class PadTests(unittest.TestCase): def setUp(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) self.work = Path(directory.name) / "work" self.work.mkdir() def run_with(self, pad, seconds=1): seen = {} class FakeSession: def __init__(self, command, cwd, env, stdout, stderr, owned=None): seen["env"] = env def alive(self): return True def send_new_windows_to_back(self): pass def exit_code(self): return None def close(self): seen["closed"] = True def kill(self): pass with unittest.mock.patch.object(title_run, "Session", FakeSession), unittest.mock.patch.object(title_run, "CLOSE_SECONDS", 0): result = title_run.run_title(self.work / "eboot.exe", self.work, self.work.parent / "run1", seconds, {}, pad) return result, seen["env"] def test_timeline_is_queued_after_the_delay(self): stale = self.work / "pad-input.txt" stale.write_text("cross\n", encoding="utf-8") result, env = self.run_with({"after_s": 0, "repeat": 2, "events": [{"at_ms": 0, "hold_ms": 200, "buttons": ["right", "cross"]}]}) self.assertEqual(env["APS5_PAD_INPUT_FILE"], str(stale)) lines = stale.read_text(encoding="utf-8").splitlines() self.assertRegex(lines[0], r"^APS5_PAD_TIMELINE 1 [0-9]+ 2$") self.assertEqual(lines[1:], [f"0 200 {0x20 | 0x4000}", f"400 200 {0x20 | 0x4000}"]) self.assertIsNotNone(result["pad_queued_s"]) def test_no_pad_leaves_the_input_file_alone(self): with unittest.mock.patch.dict(os.environ, {"APS5_PAD_INPUT_FILE": str(self.work / "outside.txt")}): result, env = self.run_with(None) self.assertNotIn("APS5_PAD_INPUT_FILE", env) self.assertNotIn("pad_queued_s", result) def test_a_late_delay_queues_nothing(self): result, _ = self.run_with({"after_s": 30, "events": [{"at_ms": 0, "hold_ms": 200, "buttons": ["cross"]}]}) self.assertIsNone(result["pad_queued_s"]) self.assertFalse((self.work / "pad-input.txt").exists()) class SessionTests(unittest.TestCase): def start(self, code, **options): session = title_run.Session([sys.executable, "-c", code], Path.cwd(), dict(os.environ), subprocess.DEVNULL, subprocess.DEVNULL, **options) self.addCleanup(session.kill) return session def test_exit_code(self): session = self.start("import sys; sys.exit(3)") self.assertTrue(wait_until(lambda: not session.alive())) self.assertEqual(session.exit_code(), 3) @unittest.skipUnless(os.name == "nt", "LoadExec hand-over is Windows-specific") def test_exit_code_of_the_last_process(self): session = self.start("import subprocess, sys; subprocess.Popen([sys.executable, '-c', 'import time, sys; time.sleep(1); sys.exit(7)'])") self.assertTrue(wait_until(lambda: not session.alive())) self.assertEqual(session.exit_code(), 7) @unittest.skipUnless(os.name == "nt", "job process list is Windows-specific") def test_exit_code_ignores_processes_outside_owned(self): child = "import subprocess, time; time.sleep(1); subprocess.run(['cmd.exe', '/c', 'exit 0']); raise SystemExit(7)" session = self.start(f"import subprocess, sys; subprocess.Popen([sys.executable, '-c', {child!r}])", owned=Path(sys.executable).parent) self.assertTrue(wait_until(lambda: not session.alive())) self.assertEqual(session.exit_code(), 7) @unittest.skipUnless(os.name == "nt", "window placement is Windows-specific") def test_new_windows_go_behind_the_others_without_polling(self): window = ("import ctypes; user32 = ctypes.WinDLL('user32'); user32.CreateWindowExW.argtypes = [ctypes.c_uint, ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint] + [ctypes.c_int] * 4 + [ctypes.c_void_p] * 4; " "user32.CreateWindowExW(0x08000000, 'STATIC', 'title_run test', 0x10CF0000, 100, 100, 200, 200, None, None, None, None); message = ctypes.create_string_buffer(64)\n" "while user32.GetMessageW(message, None, 0, 0) > 0: user32.DispatchMessageW(message)") user32 = title_run.user32 if not user32.FindWindowW("Progman", None): self.skipTest("no desktop shell in this session") session = self.start(window) user32.GetWindow.restype = title_run.wintypes.HWND user32.GetWindow.argtypes = (title_run.wintypes.HWND, title_run.wintypes.UINT) name = title_run.ctypes.create_unicode_buffer(256) def below(window): classes, window = set(), user32.GetWindow(window, 2) while window: if user32.IsWindowVisible(window) and user32.GetClassNameW(window, name, 256): classes.add(name.value) window = user32.GetWindow(window, 2) return classes self.assertTrue(wait_until(lambda: session.windows() and all(below(found) <= {"Progman", "WorkerW"} for found in session.windows()))) rect = title_run.wintypes.RECT() user32.GetWindowRect(session.windows()[0], title_run.ctypes.byref(rect)) self.assertEqual((rect.left, rect.top), (100, 100)) def test_children_keep_it_alive_and_are_killed(self): session = self.start("import subprocess, sys; subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])") self.assertTrue(wait_until(lambda: session.exit_code() == 0)) self.assertTrue(session.alive()) session.kill() self.assertTrue(wait_until(lambda: not session.alive())) if __name__ == "__main__": unittest.main()