feat(tools): run titles repeatably and report their outcome

.gitignore

@@ -99,6 +99,7 @@ downloads/
 
 *.json
 !demo/web/i18n/*.json
+!tools/titles.json
 
 !/assets/*.jpg
 !/assets/*.png

CMakeLists.txt

@@ -261,6 +261,9 @@ if(BUILD_TESTING)
         add_test(NAME performance_compare COMMAND ${Python3_EXECUTABLE}
             ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_performance_compare.py)
         set_tests_properties(performance_compare PROPERTIES TIMEOUT 20)
+        add_test(NAME title_run COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_title_run.py)
+        set_tests_properties(title_run PROPERTIES TIMEOUT 60)
     endif()
 endif()
 

docs/dev/TRACING.md

@@ -73,3 +73,7 @@ See [Build](BUILD.md#performance-trace) for the trace format, summary fields and
 ## Shader registers
 
 `APS5_TRACE_SHADER_REGS=all` prints register snapshots for every shader. Set it to a shader code address in hexadecimal to print one shader's registers.
+
+## Running titles repeatably
+
+`python tools/title_run.py --games <directory> [ID ...]` relinks and runs extracted title packages found under `--games` (a directory holding `eboot.bin` and `sce_sys/param.json`) and writes `summary.json` and a fixed-format `summary.md` under `--out` (default `build/title-runs/<commit>-<time>/`). Each title runs `--runs` times (default 2): the first with empty AnyPS5 and Mesa or NVIDIA shader caches, the next ones keeping them. AMD's Windows `VkCache` is not isolated. On Windows the title's windows open behind every other window, without focus and without being minimized, and can be brought forward from the taskbar; elsewhere run it under a virtual display if that matters. A run ends as `relink-failed`, `crashed`, `exited`, `stalled` (no present in the last 20 s, or too few presents for a frame to be dumped), `blank` (the last frame is one colour) or `rendering`; the first, middle and last dumped frames are kept. A screen that presents only when its view changes stops after a few presents and is reported `stalled` although the title still answers input; pressing a button through `tools/pad_input.py` with `APS5_PAD_INPUT_FILE` set tells it apart from a stuck title. `--compare <summary.json>` lists regressions, improvements and missing titles against an earlier run, limited to the IDs given when there are any. Titles are prepared in `<out>/work/<ID>/`, the same path on every run, so a firewall prompt for a title that opens a port appears once; answering Cancel does not affect the run. Per-title relink flags, run length and environment live in `tools/titles.json`.

tools/tests/test_title_run.py

@@ -0,0 +1,318 @@
+import json
+import os
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+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"]["fixed_base"], "0x400000")
+
+    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_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("<IiiHHIIiiII", 40, width, -height, 1, 32, 0, len(rows), 0, 0, 0, 0)
+    path.write_bytes(header + info + rows)
+    return path
+
+
+class AnalysisTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+
+    def test_first_error(self):
+        text = "Transferring control\n  what():  sysctl_nid_postfix: unsupported MIB 1.1\nFATAL: abort()\n"
+        self.assertEqual(title_run.first_error(text), "what():  sysctl_nid_postfix: unsupported MIB 1.1")
+        self.assertEqual(title_run.first_error("radv/ps5: the GPU has not reached submission 12 (at 11) in 10 s\nVkResult is ERROR_DEVICE_LOST\n"), "radv/ps5: the GPU has not reached submission 12 (at 11) in 10 s")
+        self.assertEqual(title_run.first_error("Converting\nFAIL: Forbidden syscall instruction at code offset 0x1\nInput: x\n"), "FAIL: Forbidden syscall instruction at code offset 0x1")
+        self.assertEqual(title_run.first_error("all fine\n"), "")
+
+    def test_warning_counts(self):
+        text = "[gpu] storage image 0x1 destroyed with GPU results pending\n" * 2
+        self.assertEqual(title_run.warning_counts(text)["destroyed with GPU results pending"], 2)
+
+    def test_read_summary(self):
+        path = self.root / "aps5-summary.jsonl"
+        lines = [{"presents": 300, "window": {"ms": 10000}, "gpu": "A", "driver": "1", "os": "W"},
+                 {"presents": 200, "window": {"ms": 10000}, "gpu": "B", "driver": "2", "os": "W"}]
+        path.write_text("".join(json.dumps(line) + "\n" for line in lines), encoding="utf-8")
+        self.assertEqual(title_run.read_summary(path), {"presents": 500, "fps": 25.0, "gpu": "B", "driver": "2", "os": "W"})
+        path.write_text(path.read_text(encoding="utf-8") + '{"presents": 3', encoding="utf-8")
+        self.assertEqual(title_run.read_summary(path)["presents"], 500)
+        path.write_text(path.read_text(encoding="utf-8") + "\n[1, 2]\n", encoding="utf-8")
+        self.assertEqual(title_run.read_summary(path)["presents"], 500)
+        self.assertEqual(title_run.read_summary(self.root / "missing.jsonl")["presents"], 0)
+
+    def test_frame_is_uniform(self):
+        black = bmp(self.root / "black.bmp", 8, 4, lambda x, y: (0, 0, 0))
+        noisy = bmp(self.root / "noisy.bmp", 8, 4, lambda x, y: (3, 2, 1) if x < 7 else (200, 200, 200))
+        nearly = bmp(self.root / "nearly.bmp", 8, 4, lambda x, y: (x, 0, 0))
+        self.assertTrue(title_run.frame_is_uniform(black))
+        self.assertFalse(title_run.frame_is_uniform(noisy))
+        self.assertTrue(title_run.frame_is_uniform(nearly))
+        truncated = self.root / "truncated.bmp"
+        truncated.write_bytes(black.read_bytes()[:40])
+        self.assertIsNone(title_run.frame_is_uniform(truncated))
+        cut = self.root / "cut.bmp"
+        cut.write_bytes(noisy.read_bytes()[:-20])
+        self.assertIsNone(title_run.frame_is_uniform(cut))
+
+    def test_classify(self):
+        cases = [
+            ((False, None, None, False, None), "relink-failed"),
+            ((True, 3221226505, None, False, None), "crashed"),
+            ((True, 0, None, False, None), "exited"),
+            ((True, None, None, False, None), "stalled"),
+            ((True, None, 25.0, True, False), "stalled"),
+            ((True, None, 5.0, False, None), "stalled"),
+            ((True, None, 5.0, True, True), "blank"),
+            ((True, None, 5.0, True, False), "rendering"),
+            ((True, None, 5.0, True, None), "rendering"),
+        ]
+        for arguments, outcome in cases:
+            self.assertEqual(title_run.classify(*arguments), outcome, arguments)
+
+
+def run(outcome, cache="cold", fps=0.0, error=""):
+    return {"outcome": outcome, "cache": cache, "fps": fps, "first_error": error}
+
+
+def summary(titles):
+    return {"build": {"commit": "abc1234", "dirty": False, "note": "main + #6814"},
+            "machine": {"os": "Windows 11", "gpu": "Radeon 860M", "driver": "32.0.13058.8001", "cpu": "Ryzen AI 7 350"},
+            "runs": 2, "seconds": 60, "titles": titles}
+
+
+class ReportTests(unittest.TestCase):
+    def test_cell(self):
+        runs = [run("crashed"), run("rendering", "warm"), run("crashed", "warm")]
+        self.assertEqual(title_run.cell(runs, "cold"), "crashed")
+        self.assertEqual(title_run.cell(runs, "warm"), "crashed 1/2, rendering 1/2")
+        self.assertEqual(title_run.cell([run("rendering", "warm"), run("rendering", "warm")], "warm"), "rendering 2/2")
+        self.assertEqual(title_run.cell([run("rendering")], "warm"), "-")
+
+    def test_worst_and_compare(self):
+        now = summary({"PPSA85300": {"name": "Light", "version": "0.4", "runs": [run("crashed"), run("crashed", "warm")]},
+                       "PPSA97358": {"name": "XPSemu", "version": "A2.1", "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.assertEqual(title_run.worst(now["titles"]["PPSA85300"]["runs"]), "crashed")
+        self.assertEqual(title_run.compare(now, before),
+                         ([("PPSA85300", "rendering", "crashed")], [("PPSA97358", "stalled", "rendering")]))
+
+    def test_render_markdown(self):
+        now = summary({"PPSA85300": {"name": "Light", "version": "0.4",
+                                     "runs": [run("crashed", error="what(): unsupported MIB 1.1 | x"), run("rendering", "warm", 31.0)]}})
+        before = summary({"PPSA85300": {"name": "Light", "version": "0.4", "runs": [run("rendering"), run("rendering", "warm")]}})
+        self.assertEqual(title_run.render_markdown(now, before), "\n".join([
+            "AnyPS5 `abc1234`, main + #6814. Windows 11, Radeon 860M, driver 32.0.13058.8001, CPU Ryzen AI 7 350. 2 runs per title, 60 s each.",
+            "",
+            "Regressions: Light (PPSA85300) rendering -> 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_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_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 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)")
+        session = self.start(window)
+        user32 = title_run.user32
+        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()