"""core_probe runs a libretro core headless and records what it measured. The parsers are checked on fixed inputs. One test builds a ten-line libretro core with the C compiler, when there is one, and runs the whole probe on it: staging, the worker process, strace, the record and its schema. """ from __future__ import annotations import argparse import contextlib import io import json import os import shutil import struct import subprocess import sys import tempfile import unittest import unittest.mock from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) import core_probe import measurement_records STRACE_SAMPLE = """\ 4242 openat(AT_FDCWD, "/sys/bios/card.bin", O_RDONLY) = 5 4242 openat(AT_FDCWD, "/sys/bios/missing.bin", O_RDONLY) = -1 ENOENT (No such file or directory) [pid 4243] open("/sys/bios/sub/x.rom", O_RDONLY|O_CLOEXEC) = 7 4242 openat(AT_FDCWD, "/elsewhere/game.cue", O_RDONLY) = 3 4242 read(3, "...", 10) = 10 """ FAKE_CORE = r""" #include #include #include #include #include #include struct retro_system_info { const char *n, *v, *e; bool full, block; }; struct retro_game_geometry { unsigned bw, bh, mw, mh; float a; }; struct retro_system_timing { double fps, rate; }; struct retro_system_av_info { struct retro_game_geometry g; struct retro_system_timing t; }; struct retro_log_callback { void (*log)(int, const char *, ...); }; typedef bool (*env_t)(unsigned, void *); typedef void (*video_t)(const void *, unsigned, unsigned, size_t); typedef size_t (*batch_t)(const int16_t *, size_t); typedef int16_t (*input_t)(unsigned, unsigned, unsigned, unsigned); static env_t env; static video_t video; static batch_t batch; static input_t input; static unsigned ports, frames; static struct retro_log_callback logger; static uint16_t pixels[4 * 2]; static unsigned char card; void retro_set_environment(env_t e) { bool t = true; env = e; env(18, &t); } void retro_set_video_refresh(video_t v) { video = v; } void retro_set_audio_sample(void *f) { (void)f; } void retro_set_audio_sample_batch(batch_t b) { batch = b; } void retro_set_input_poll(void *f) { (void)f; } void retro_set_input_state(input_t f) { input = f; } void retro_set_controller_port_device(unsigned port, unsigned device) { (void)port; if (device == 1) ports++; } void retro_get_system_info(struct retro_system_info *i) { i->n = "Fake"; i->v = "1"; i->e = ""; i->full = true; i->block = false; } void retro_get_system_av_info(struct retro_system_av_info *i) { memset(i, 0, sizeof *i); i->t.fps = 60.0; i->t.rate = 44100.0; } void retro_init(void) {} void retro_deinit(void) {} bool retro_load_game(const void *g) { const char *dir = NULL; char path[4096]; struct retro_log_callback cb; const char *tag = getenv("CORE_PROBE_TEST_TAG"); (void)g; env(9, &dir); snprintf(path, sizeof path, "%s/card.bin", dir); FILE *f = fopen(path, "rb"); card = 0; if (f) { int c = fgetc(f); card = c < 0 ? 0 : (unsigned char)c; fclose(f); } if (env(27, &cb)) { logger = cb; cb.log(1, "card %s byte %d\n", f ? "present" : "absent", (int)card); cb.log(1, "tag %s\n", tag ? tag : "unset"); } return true; } void retro_unload_game(void) {} void retro_run(void) { int16_t s[4] = {1, 2, 3, 4}; if (frames++ == 0 && logger.log) logger.log(1, "ports %u\n", ports); input(0, 1, 0, 8); for (int i = 0; i < 8; i++) pixels[i] = card; video(pixels, 4, 2, 8); batch(s, 2); } size_t retro_serialize_size(void) { return 1; } bool retro_serialize(void *d, size_t n) { (void)n; *(unsigned char *)d = card; return true; } """ def mednafen_state(sections: dict[str, dict[str, bytes]]) -> bytes: body = b"" for section, entries in sections.items(): chunk = b"".join(bytes([len(name)]) + name.encode() + struct.pack(" Path: """A folder made the root of a git checkout, as --record requires.""" path.mkdir(parents=True, exist_ok=True) subprocess.run(["git", "init", "-q", str(path)], check=True) return path class ParsingTests(unittest.TestCase): def test_assignment(self) -> None: self.assertEqual(core_probe.parse_assignment("a=b=c", "--option"), ("a", "b=c")) self.assertEqual(core_probe.parse_assignment("dest=", "--vary"), ("dest", "")) for bad in ("novalue", "=x"): with self.assertRaises(argparse.ArgumentTypeError): core_probe.parse_assignment(bad, "--option") def test_press(self) -> None: self.assertEqual(core_probe.parse_press("120:start"), (120, 3, 1)) self.assertEqual(core_probe.parse_press("5:A:10"), (5, 8, 10)) for bad in ("x:a", "1:turbo", "1:a:0", "-1:a", "1"): with self.assertRaises(argparse.ArgumentTypeError): core_probe.parse_press(bad) def test_strace(self) -> None: calls = core_probe.parse_strace(STRACE_SAMPLE) self.assertEqual([c["path"] for c in calls], ["/sys/bios/card.bin", "/sys/bios/missing.bin", "/sys/bios/sub/x.rom", "/elsewhere/game.cue"]) self.assertEqual([c["ok"] for c in calls], [True, False, True, True]) self.assertEqual(calls[1]["errno"], "ENOENT") opened, absent = core_probe.files_under(calls, Path("/sys/bios")) self.assertEqual(opened, ["card.bin", "sub/x.rom"]) self.assertEqual(absent, ["missing.bin"]) def test_printf(self) -> None: name = core_probe.ctypes.create_string_buffer(b"mpeg_card.bin") address = core_probe.ctypes.addressof(name) self.assertEqual(core_probe.render_printf(b"open %s: %d%% %x\n", [address, 50, 255]), "open mpeg_card.bin: 50% ff") self.assertEqual(core_probe.render_printf(b"%d", [0xFFFFFFFF]), "-1") self.assertEqual(core_probe.render_printf(b"%.2f fps %u", [7]), " fps 7") self.assertEqual(core_probe.render_printf(b"%s", [0]), "(null)") def test_mednafen_state(self) -> None: state = b"\x00" * 7 + mednafen_state({"MPEG": {"Present": b"\x01", "AuthState": b"\x02"}, "CDB": {"Word": b"\x34\x12"}}) self.assertEqual(core_probe.mednafen_state_vars(state), {"MPEG.Present": "01", "MPEG.AuthState": "02", "CDB.Word": "3412"}) self.assertEqual(core_probe.mednafen_state_vars(b"no state"), {}) def test_elf_comment(self) -> None: self.assertIsInstance(core_probe.elf_comment(Path(sys.executable).resolve()), list) with tempfile.NamedTemporaryFile(dir=REPO_ROOT / "tmp" / "tests") as handle: handle.write(b"not an elf") handle.flush() self.assertEqual(core_probe.elf_comment(Path(handle.name)), []) def test_env_file_recorded_by_name_and_hash(self) -> None: library = Path(sys.executable).resolve() described = core_probe.describe_env_value(str(library)) self.assertEqual(described, f"{library.name} sha256:{core_probe.file_digest(library, 'sha256')}") self.assertEqual(core_probe.describe_env_value("2000-01-01 00:00:00"), "2000-01-01 00:00:00") def test_uniform_frame(self) -> None: self.assertTrue(core_probe.uniform_frame(b"", 2)) self.assertTrue(core_probe.uniform_frame(b"\x01\x02" * 8, 2)) self.assertFalse(core_probe.uniform_frame(b"\x01\x02" * 7 + b"\x01\x03", 2)) self.assertFalse(core_probe.uniform_frame(b"\x00\x00\x00\x00\x00\x00\x00\x01", 4)) def test_cue_tracks(self) -> None: with tempfile.TemporaryDirectory(dir=REPO_ROOT / "tmp" / "tests") as tmp: cue = Path(tmp, "d.cue") cue.write_text('FILE "a b.bin" BINARY\n TRACK 01 MODE2/2352\nFILE c.bin BINARY\n', encoding="utf-8") self.assertEqual([p.name for p in core_probe.cue_tracks(cue)], ["a b.bin", "c.bin"]) class ResultTests(unittest.TestCase): @staticmethod def variant(label: str, value: str, loaded: bool = True) -> dict: return {"label": label, "flags": [], "repetitions": [{"loaded": loaded, "observables": {"video_sha1": value}}]} def test_same_differs_inconclusive(self) -> None: same = core_probe.check_result([self.variant("a", "1"), self.variant("b", "1")], "video_sha1") self.assertEqual(same["status"], "same") differs = core_probe.check_result([self.variant("a", "1"), self.variant("b", "2")], "video_sha1") self.assertEqual(differs["status"], "differs") self.assertEqual(len(differs["groups"]), 2) flagged = core_probe.check_result([self.variant("a", "1", loaded=False), self.variant("b", "1")], "video_sha1") self.assertEqual(flagged["status"], "inconclusive") def test_other_observable_does_not_void_the_check(self) -> None: variant = {"label": "a", "flags": ["repetition 1 disagrees with repetition 0 on state_sha1"], "repetitions": [{"loaded": True, "observables": {"video_sha1": "1", "state_sha1": "x"}}, {"loaded": True, "observables": {"video_sha1": "1", "state_sha1": "y"}}]} result = core_probe.check_result([variant], "video_sha1") self.assertEqual(result["status"], "same") self.assertIn("Other flags", result["statement"]) self.assertEqual(core_probe.check_result([variant], "state_sha1")["status"], "inconclusive") def test_host_paths_hidden(self) -> None: places = [("/tmp/x/system", ""), ("/tmp/x", "")] self.assertEqual(core_probe.host_paths_hidden("/tmp/x/system/bios.bin NOT FOUND", places), "/bios.bin NOT FOUND") def test_timing_stats(self) -> None: stats = core_probe.timing_stats([1.0, 2.0, 3.0, 4.0, 10.0]) self.assertEqual((stats["median"], stats["mean"], stats["min"], stats["max"]), (3.0, 4.0, 1.0, 10.0)) self.assertEqual(stats["raw"], [1.0, 2.0, 3.0, 4.0, 10.0]) def test_refusals(self) -> None: parser = core_probe.build_parser() core = sys.executable bench = parser.parse_args(["bench", "--core", core, "--record", "x", "--repeat", "3"]) self.assertIn("a recorded benchmark needs --repeat >= 5", core_probe.validate_args(bench)) check = parser.parse_args(["check", "--core", core, "--vary", "a=", "--vary", "b=", "--observable", "video_sha1", "--claim", "c", "--method", "m"]) self.assertTrue(any("same DEST" in p for p in core_probe.validate_args(check))) unknown = parser.parse_args(["check", "--core", core, "--vary", "a=", "--observable", "nope", "--claim", "c", "--method", "m"]) self.assertTrue(any("--observable nope" in p for p in core_probe.validate_args(unknown))) @unittest.skipUnless(shutil.which("cc"), "no C compiler") class FakeCoreTests(unittest.TestCase): """The whole probe against a core compiled from FAKE_CORE.""" @classmethod def setUpClass(cls) -> None: cls.tmp = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests")) source = cls.tmp / "fake.c" source.write_text(FAKE_CORE, encoding="utf-8") cls.core = cls.tmp / "fake_libretro.so" subprocess.run(["cc", "-shared", "-fPIC", "-o", str(cls.core), str(source)], check=True) cls.cards = [] for byte in (b"\x01", b"\x01", b"\x02"): path = cls.tmp / f"card{len(cls.cards)}.bin" path.write_bytes(byte) cls.cards.append(path) @classmethod def tearDownClass(cls) -> None: shutil.rmtree(cls.tmp, ignore_errors=True) def probe(self, argv: list[str]) -> tuple[int, dict | None]: out = io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()): code = core_probe.main(argv + ["--work-dir", str(self.tmp / "work")]) return code, json.loads(out.getvalue()) if code == 0 else None def test_run_record(self) -> None: record_dir = checkout(self.tmp / "records") code, record = self.probe(["check", "--core", str(self.core), "--frames", "3", "--vary", f"card.bin={self.cards[0]}", "--vary", f"card.bin={self.cards[1]}", "--vary", f"card.bin={self.cards[2]}", "--vary", "card.bin=", "--observable", "frame_sha1", "--claim", "c", "--method", "m", "--record", str(record_dir), "--max-load", "1000", "--env", "CORE_PROBE_TEST_TAG=set-by-test", "--press", "0:a:2"]) self.assertEqual(code, 0) result = record["check"]["result"] self.assertEqual(result["status"], "differs") self.assertEqual(sorted(len(g["variants"]) for g in result["groups"]), [1, 1, 2]) first = record["variants"][0]["repetitions"][0] self.assertEqual(first["logs"][0]["text"], "card present byte 1") self.assertEqual(first["logs"][1]["text"], "tag set-by-test") self.assertEqual(first["logs"][2]["text"], "ports 2") self.assertEqual((first["input_queries"], first["input_pressed"]), (3, 2)) self.assertEqual(record["inputs"]["env"], {"CORE_PROBE_TEST_TAG": "set-by-test"}) self.assertEqual(record["variants"][3]["repetitions"][0]["logs"][0]["text"], "card absent byte 0") if first["tracing"] == "strace": self.assertEqual([f["path"] for f in first["files_opened"]], ["card.bin"]) self.assertEqual(record["variants"][3]["repetitions"][0]["files_probed_absent"], ["card.bin"]) self.assertEqual(first["frames"], 3) self.assertEqual(record["emulator"], "fake") self.assertEqual([p.relative_to(record_dir).as_posix() for p in record_dir.rglob("*") if p.is_file() and ".git" not in p.parts], ["measurements/fake.jsonl"]) line = json.loads((record_dir / "measurements" / "fake.jsonl").read_text()) self.assertEqual(measurement_records.check_line(line), []) # The line is the record the run printed, id included. self.assertEqual(line, record) try: from jsonschema import Draft202012Validator except ImportError: return schema = json.loads((REPO_ROOT / "schemas" / "core-probe-record.schema.json").read_text()) self.assertEqual(list(Draft202012Validator(schema).iter_errors(line)), []) def test_second_run_is_appended_beside_the_first(self) -> None: record_dir = checkout(self.tmp / "appended") argv = ["run", "--core", str(self.core), "--frames", "1", "--no-trace", "--max-load", "1000", "--record", str(record_dir), "--subject", "smoke"] self.assertEqual(self.probe(argv)[0], 0) lines = record_dir / "measurements" / "fake.jsonl" first = lines.read_text() code, record = self.probe(argv) self.assertEqual(code, 0) self.assertTrue(lines.read_text().startswith(first)) ids = [json.loads(line)["id"] for line in lines.read_text().splitlines()] self.assertEqual(len(ids), 2) code, corrected = self.probe(argv + ["--supersedes", ids[0]]) self.assertEqual((code, corrected["supersedes"]), (0, ids[0])) code, _ = self.probe(argv + ["--supersedes", "2000-01-01-000000000000"]) self.assertEqual(code, 2) self.assertEqual(len(lines.read_text().splitlines()), 3) def test_record_defaults_to_this_checkout(self) -> None: parser = core_probe.build_parser() args = parser.parse_args(["run", "--core", sys.executable, "--record", "--frames", "1"]) self.assertEqual(Path(args.record), REPO_ROOT) self.assertIsNone(parser.parse_args(["run", "--core", sys.executable]).record) def test_bench_stats(self) -> None: code, record = self.probe(["bench", "--core", str(self.core), "--frames", "2", "--warmup", "1", "--repeat", "5", "--no-trace", "--max-load", "1000"]) self.assertEqual(code, 0) self.assertEqual(len(record["timing"]["ms_per_frame"]["raw"]), 5) self.assertEqual(record["variants"][0]["repetitions"][0]["tracing"], "disabled: --no-trace") self.assertEqual(record["flags"], []) def test_busy_machine_refused(self) -> None: refused = checkout(self.tmp / "refused") code, _ = self.probe(["run", "--core", str(self.core), "--frames", "1", "--record", str(refused), "--max-load", "-1"]) self.assertEqual(code, 3) self.assertFalse((refused / "measurements").exists()) def test_record_names_a_checkout_root(self) -> None: # --record measurements wrote measurements/measurements/.jsonl. root = checkout(self.tmp / "rooted") (root / "measurements").mkdir() for wrong in (root / "measurements", self.tmp / "plain"): wrong.mkdir(exist_ok=True) err = io.StringIO() with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(err): code = core_probe.main(["run", "--core", str(self.core), "--frames", "1", "--no-trace", "--record", str(wrong), "--work-dir", str(self.tmp / "work")]) self.assertEqual(code, 2) self.assertIn("not a git checkout root", err.getvalue()) self.assertIn("--record .", err.getvalue()) self.assertFalse((wrong / "measurements" / "fake.jsonl").exists()) self.assertIsNone(core_probe.not_a_checkout_root(root)) self.assertIsNone(core_probe.not_a_checkout_root(REPO_ROOT)) def test_the_core_sees_only_the_env_it_was_given(self) -> None: os.environ["CORE_PROBE_TEST_TAG"] = "from-the-host" os.environ["GITHUB_TOKEN"] = "secret" try: code, record = self.probe(["run", "--core", str(self.core), "--frames", "1", "--no-trace"]) finally: del os.environ["CORE_PROBE_TEST_TAG"] del os.environ["GITHUB_TOKEN"] self.assertEqual(code, 0) rep = record["variants"][0]["repetitions"][0] self.assertEqual(rep["logs"][1]["text"], "tag unset") self.assertTrue(rep["isolation"] == "bwrap" or rep["isolation"].startswith("environment only")) class Sandbox(unittest.TestCase): """What a core run under bubblewrap can reach: an allowlist.""" def setUp(self) -> None: if not core_probe.sandbox_argv([])[0]: self.skipTest("no working bwrap on this host") self.tmp = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests")) self.addCleanup(shutil.rmtree, self.tmp, True) self.out = self.tmp / "out" self.inputs = self.tmp / "inputs" self.inputs.mkdir() (self.inputs / "bios.bin").write_text("BIOS") self.elsewhere = self.tmp / "elsewhere" self.elsewhere.mkdir() def inside(self, script: str, **kwargs: object) -> subprocess.CompletedProcess: prefix, isolation = core_probe.sandbox_argv([self.out], [self.inputs], **kwargs) self.assertEqual(isolation, core_probe.SANDBOX) return subprocess.run(prefix + ["/bin/sh", "-c", script], capture_output=True, text=True, check=False, env={"PATH": "/usr/bin:/bin"}) def test_nothing_of_the_user_or_the_checkout_is_reachable(self) -> None: home = Path.home() hidden = [home / ".config", home / ".ssh", home / ".git-credentials", home / ".netrc", home / ".gitconfig", REPO_ROOT / ".git", REPO_ROOT / "scripts", REPO_ROOT / "bios", REPO_ROOT / "emulators", self.elsewhere, Path("/mnt/c"), Path("/root"), Path("/var"), Path("/etc/shadow"), Path("/opt")] script = "; ".join(f'test -e "{path}" && echo "LEAK {path}"' for path in hidden) # The checkout exists only as the path down to the bound folders. script += f'; ls -A /mnt /tmp; ls -A ~ 2>/dev/null; ls -A "{REPO_ROOT}"; echo end' proc = self.inside(script) self.assertNotIn("LEAK", proc.stdout) self.assertEqual(proc.stdout.split(), ["/mnt:", "/tmp:", "tmp", "end"]) def test_inputs_read_only_work_dir_writable(self) -> None: proc = self.inside(f'cat "{self.inputs}/bios.bin"; echo; echo x > "{self.inputs}/bios.bin" 2>/dev/null ' f'|| echo refused; echo ok > "{self.out}/w"; echo no > "{self.elsewhere}/w" ' f'2>/dev/null || echo gone') self.assertEqual(proc.stdout.split(), ["BIOS", "refused", "gone"]) self.assertEqual((self.inputs / "bios.bin").read_text(), "BIOS") self.assertEqual((self.out / "w").read_text(), "ok\n") self.assertFalse((self.elsewhere / "w").exists()) def test_no_network_unless_asked(self) -> None: devices = "tail -n +3 /proc/net/dev | cut -d: -f1" self.assertEqual(self.inside(devices).stdout.split(), ["lo"]) self.assertNotEqual(self.inside(devices, network=True).stdout.split(), ["lo"]) def test_a_probe_can_require_the_sandbox(self) -> None: args = core_probe.build_parser().parse_args( ["run", "--core", sys.executable, "--require-sandbox"]) self.assertEqual([p for p in core_probe.validate_args(args) if "sandbox" in p], []) with unittest.mock.patch.object(core_probe.shutil, "which", return_value=None): problems = core_probe.validate_args(args) self.assertIn("--require-sandbox: environment only: bwrap not found", problems) if __name__ == "__main__": unittest.main()