mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-11-01 07:33:45 -06:00
chore: merge the automatic test campaign
This commit is contained in:
Abdessamad Derraz committed
commit
9c8dbcd99a
31 files changed
+4132
-49
No files matched your search
@@ -0,0 +1,42 @@
|
||||
# Test content of the real-condition campaign (scripts/campaign.py), per system.
|
||||
#
|
||||
# Each entry names a catalog entry, never a file: `name` is the exact Redump or
|
||||
# No-Intro name romfetch fetches and verifies (`romfetch get <romfetch> <name>`),
|
||||
# and `reason` says why this game serves the test. Content is read from
|
||||
# .cache/test-content/<romfetch>/<name>/ with its identity.json, and never
|
||||
# enters git. A record keeps only its identity (catalog name, sha1, tracks).
|
||||
#
|
||||
# region the territory of the disc or cartridge: the BIOS entry the
|
||||
# boot test varies is the one whose `region:` holds it, or one
|
||||
# without `region:`
|
||||
# load the file handed to the core: `cue` (the .cue sheet) or a
|
||||
# file name inside the folder, by default the single file
|
||||
# frames frames run per repetition, the same with and without the file
|
||||
# frame_hash_every frames between two sampled frame hashes
|
||||
# pinned never deleted by the campaign, whoever fetched it
|
||||
# boot_file (per system) the declared file the boot test varies, when a
|
||||
# profile declares it under this name or an alias
|
||||
#
|
||||
# The checks a game is used for are the boot test (no file against the
|
||||
# declared file), the files test (every declared file staged, which ones the
|
||||
# core opens) and the benchmark.
|
||||
|
||||
systems:
|
||||
sony-playstation:
|
||||
romfetch: psx
|
||||
# The file the boot test varies, when the profile declares it: the US
|
||||
# BIOS of a US disc, the same file for every core so the runs compare.
|
||||
boot_file: scph5501.bin
|
||||
content:
|
||||
- name: "Bust-A-Move 2 - Arcade Edition (USA)"
|
||||
catalog: redump
|
||||
reason: >-
|
||||
A US disc that goes through the BIOS shell and its licence screen, then
|
||||
reaches the publisher logos and the attract mode within a minute, with
|
||||
no button pressed. Several static screens, so a run with and a run
|
||||
without the BIOS can be compared frame by frame.
|
||||
region: north-america
|
||||
load: cue
|
||||
frames: 3600
|
||||
frame_hash_every: 10
|
||||
pinned: true
|
||||
@@ -19,6 +19,7 @@ import time
|
||||
import unittest
|
||||
import unittest.mock
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
||||
@@ -196,6 +197,100 @@ class Guards(Scratch):
|
||||
self.assertIsNone(record["last_success"])
|
||||
|
||||
|
||||
class StepEnvironment(Scratch):
|
||||
"""A step can run third-party code: it never inherits a token or a password."""
|
||||
|
||||
HOST = MappingProxyType({
|
||||
"PATH": "/usr/bin", "HOME": "/home/u", "LANG": "C.UTF-8", "LC_ALL": "C", "TZ": "UTC",
|
||||
"TMPDIR": "/t", "GITHUB_TOKEN": "t1", "GH_TOKEN": "t2", "ROMSTATION_USER": "u",
|
||||
"ROMSTATION_PASS": "p", "NPM_TOKEN": "t3", "AWS_SECRET_ACCESS_KEY": "s",
|
||||
"DB_PASSWORD": "p2", "ROMFETCH_CONTENT_DIR": "/c", "EDITOR": "vi"})
|
||||
|
||||
def test_the_allowlist(self):
|
||||
env = ar.safe_env(["ROMFETCH_CONTENT_DIR", "GITHUB_TOKEN"], {"RETROBIOS_AUTO": "1"},
|
||||
source=dict(self.HOST))
|
||||
self.assertEqual(env, {"PATH": "/usr/bin", "HOME": "/home/u", "LANG": "C.UTF-8",
|
||||
"LC_ALL": "C", "TZ": "UTC", "TMPDIR": "/t",
|
||||
"ROMFETCH_CONTENT_DIR": "/c", "RETROBIOS_AUTO": "1"})
|
||||
|
||||
def test_a_step_naming_a_secret_is_refused_at_load(self):
|
||||
for name in ("GITHUB_TOKEN", "ROMSTATION_PASS", "MY_SECRET"):
|
||||
with self.subTest(name=name), self.assertRaises(ValueError):
|
||||
ar.parse_jobs({"jobs": {"x": {"steps": [{"run": [PY, "-V"], "env": [name]}]}}})
|
||||
|
||||
def test_a_network_job_step_sees_no_token(self):
|
||||
dump = "import json, os; print(json.dumps(dict(os.environ)))"
|
||||
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "network": True,
|
||||
"steps": [{"run": [PY, "-c", dump], "stdout": "env.json",
|
||||
"env": ["ROMFETCH_CONTENT_DIR"]}]}})
|
||||
with unittest.mock.patch.dict(os.environ, {"GITHUB_TOKEN": "t1", "GH_TOKEN": "t2",
|
||||
"ROMSTATION_PASS": "p",
|
||||
"ROMFETCH_CONTENT_DIR": "/c", "EDITOR": "vi"}):
|
||||
self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK)
|
||||
given = ar.token_env()
|
||||
seen = json.loads((ctx.repo / "env.json").read_text())
|
||||
self.assertFalse({"GITHUB_TOKEN", "GH_TOKEN", "ROMSTATION_PASS", "EDITOR"} & set(seen))
|
||||
self.assertEqual((seen["ROMFETCH_CONTENT_DIR"], seen["RETROBIOS_AUTO"]), ("/c", "1"))
|
||||
self.assertEqual(seen[ar.LOCK_HELD_ENV], "1")
|
||||
# The push and gh are the phases that get it.
|
||||
self.assertEqual(given["GITHUB_TOKEN"], "t1")
|
||||
|
||||
|
||||
class GithubTokenPerStep(Scratch):
|
||||
"""GITHUB_TOKEN goes to first-party scripts that call the API, declared
|
||||
per step, and never to third-party code or the campaign."""
|
||||
|
||||
def job(self, run: list[str], name: str = "x") -> dict:
|
||||
return ar.parse_jobs({"jobs": {name: {"steps": [{"run": run, "github_token": True}]}}})
|
||||
|
||||
def test_allowed_on_the_scrapers_and_freshness_scripts(self):
|
||||
for run in ([PY, "-m", "scripts.scraper.mame_hash_scraper"],
|
||||
[PY, "scripts/check_freshness.py", "--json"],
|
||||
[PY, "scripts/refresh_stale.py", "--only", "data"]):
|
||||
with self.subTest(run=run):
|
||||
step = self.job(run)["x"].steps[0]
|
||||
self.assertTrue(step.github_token)
|
||||
with unittest.mock.patch.object(ar, "token_env", return_value={"GITHUB_TOKEN": "t1"}):
|
||||
self.assertEqual(ar.job_env(self.job(run)["x"], step)["GITHUB_TOKEN"], "t1")
|
||||
|
||||
def test_refused_on_any_other_step(self):
|
||||
for run in ([PY, "-c", "print(1)"], [PY, "scripts/campaign.py", "run"],
|
||||
[PY, "scripts/core_probe.py", "run"], ["sh", "-c", "env"],
|
||||
[PY, "scripts/not_tracked_here.py"], [PY, "-m", "scripts.scraper.not_tracked"],
|
||||
[PY, "-m", "os"], ["/usr/bin/python3", "scripts/check_freshness.py"]):
|
||||
with self.subTest(run=run), self.assertRaises(ValueError):
|
||||
self.job(run)
|
||||
|
||||
def test_refused_in_the_campaign_job_even_on_a_scraper(self):
|
||||
with self.assertRaisesRegex(ValueError, "campaign never receives"):
|
||||
self.job([PY, "-m", "scripts.scraper.mame_hash_scraper"], name="campaign")
|
||||
|
||||
def test_an_undeclared_step_gets_no_token(self):
|
||||
job = ar.parse_jobs({"jobs": {"x": {"network": True, "steps": [
|
||||
[PY, "-m", "scripts.scraper.mame_hash_scraper"]]}}})["x"]
|
||||
with unittest.mock.patch.object(ar, "token_env", return_value={"GITHUB_TOKEN": "t1"}), \
|
||||
unittest.mock.patch.dict(os.environ, {"GITHUB_TOKEN": "t1"}):
|
||||
self.assertNotIn("GITHUB_TOKEN", ar.job_env(job, job.steps[0]))
|
||||
|
||||
def test_the_committed_registry(self):
|
||||
jobs = ar.load_jobs()
|
||||
self.assertTrue(all(step.github_token for step in jobs["nightly"].steps))
|
||||
self.assertFalse(any(step.github_token for step in jobs["campaign"].steps + jobs["campaign"].gate))
|
||||
|
||||
|
||||
class ReportAlert(Scratch):
|
||||
def test_a_report_alert_is_raised_then_cleared(self):
|
||||
write = "import json, sys; json.dump({'alert': sys.argv[1]} if sys.argv[1] else {}, open('r.json', 'w'))"
|
||||
for message, expected in (("no sandbox", True), ("", False)):
|
||||
ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "report": "r.json",
|
||||
"steps": [[PY, "-c", write, message]]}})
|
||||
ar.run_job(ctx, "x")
|
||||
alerts = self.state(ctx).get("alerts", {})
|
||||
self.assertEqual("job:x" in alerts, expected)
|
||||
if expected:
|
||||
self.assertEqual(alerts["job:x"]["detail"], "no sandbox")
|
||||
|
||||
|
||||
class Retry(Scratch):
|
||||
def _runs(self, *outcomes: ar.Outcome):
|
||||
queue = list(outcomes)
|
||||
|
||||
@@ -0,0 +1,999 @@
|
||||
"""campaign.py runs the real-condition campaign: state, guards, content, rules.
|
||||
|
||||
No network and no real core. The rules and the profile editor are checked on
|
||||
synthetic records and profiles, the guards and the probe retries on a fake
|
||||
host, the content store on folders made in the test. One class builds a
|
||||
libretro core from C, when a compiler is there, and runs whole passes on a
|
||||
synthetic checkout: records, rules applied with their evidence, findings,
|
||||
resume and rerun on a changed core.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import unittest.mock
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
|
||||
import yaml
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
||||
|
||||
import campaign
|
||||
import campaign_profiles
|
||||
import campaign_rules as rules
|
||||
import measurement_records as mr
|
||||
import sitemeasurements
|
||||
|
||||
TMP = REPO_ROOT / "tmp" / "tests"
|
||||
|
||||
PROFILE = """\
|
||||
emulator: Fake
|
||||
type: libretro
|
||||
cores: [fake]
|
||||
systems: [fake-system]
|
||||
notes: >
|
||||
A profile for the tests.
|
||||
|
||||
files:
|
||||
# -- the BIOS --
|
||||
|
||||
- name: "bios.bin"
|
||||
description: "The BIOS"
|
||||
required: true
|
||||
size: 4
|
||||
validation: [size]
|
||||
aliases:
|
||||
- "Bios.bin"
|
||||
|
||||
- name: "other.bin"
|
||||
required: false
|
||||
"""
|
||||
|
||||
FAKE_CORE = r"""
|
||||
#include <stdbool.h>
|
||||
#include <stddef.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
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; };
|
||||
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 uint16_t pixels[4 * 2]; static int booted; static unsigned frame;
|
||||
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) { (void)f; }
|
||||
void retro_set_controller_port_device(unsigned port, unsigned device) { (void)port; (void)device; }
|
||||
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) {}
|
||||
static FILE *try_open(const char *dir, const char *name) {
|
||||
char path[4096]; snprintf(path, sizeof path, "%s/%s", dir, name); return fopen(path, "rb"); }
|
||||
bool retro_load_game(const void *g) {
|
||||
const char *dir = NULL; FILE *f; (void)g; env(9, &dir);
|
||||
f = try_open(dir, "bios.bin");
|
||||
if (!f) f = try_open(dir, "BIOS.BIN");
|
||||
#ifdef FAKE_HLE
|
||||
if (f) fclose(f);
|
||||
booted = 1;
|
||||
#else
|
||||
if (f) { fclose(f); booted = 1; } else booted = 0;
|
||||
#endif
|
||||
return true; }
|
||||
void retro_unload_game(void) {}
|
||||
void retro_run(void) {
|
||||
int16_t s[4] = {1, 2, 3, 4};
|
||||
#ifdef FAKE_HANG
|
||||
for (;;) {}
|
||||
#endif
|
||||
for (int i = 0; i < 8; i++) pixels[i] = booted ? (uint16_t)(frame / 2 + i) : 0;
|
||||
frame++;
|
||||
video(pixels, 4, 2, 8); batch(s, 2); }
|
||||
size_t retro_serialize_size(void) { return 0; }
|
||||
bool retro_serialize(void *d, size_t n) { (void)d; (void)n; return false; }
|
||||
"""
|
||||
|
||||
|
||||
def rep(frames: list[str], uniform: list[str] | None = None, loaded: bool = True) -> dict:
|
||||
return {"loaded": loaded, "observables": {
|
||||
"frame_hashes": [[i * 10, h] for i, h in enumerate(frames)],
|
||||
"uniform_frames": sorted(set(uniform or []))}}
|
||||
|
||||
|
||||
def boot_record(absent: list[dict], present: list[dict], rid: str = "2026-10-27-aaaaaaaaaaaa") -> dict:
|
||||
return {"id": rid, "variants": [{"label": "bios.bin-absent", "repetitions": absent},
|
||||
{"label": "bios.bin", "repetitions": present}]}
|
||||
|
||||
|
||||
class BootOutcome(unittest.TestCase):
|
||||
GAME = ("black", "logo", "title", "title2")
|
||||
|
||||
def test_boots_when_screens_are_shared(self) -> None:
|
||||
record = boot_record([rep(["black", "hle", "logo", "title"], ["black"])] * 2,
|
||||
[rep(self.GAME, ["black"])] * 2)
|
||||
boot = rules.boot_outcome(record, "bios.bin-absent", "bios.bin")
|
||||
self.assertEqual((boot.status, boot.shared), (rules.BOOTS, 2))
|
||||
|
||||
def test_one_colour_and_no_load_fail(self) -> None:
|
||||
black = boot_record([rep(["black"], ["black"]), rep([], [], loaded=False)],
|
||||
[rep(self.GAME, ["black"])])
|
||||
self.assertEqual(rules.boot_outcome(black, "bios.bin-absent", "bios.bin").status, rules.FAILS)
|
||||
|
||||
def test_a_shared_blank_screen_proves_nothing(self) -> None:
|
||||
# The black frame both runs start on is one colour: it is no game screen.
|
||||
record = boot_record([rep(["black", "error"], ["black"])], [rep(self.GAME, ["black"])])
|
||||
self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status,
|
||||
rules.INCONCLUSIVE)
|
||||
|
||||
def test_one_shared_screen_is_not_enough(self) -> None:
|
||||
record = boot_record([rep(["logo", "error"])], [rep(self.GAME, ["black"])])
|
||||
self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status,
|
||||
rules.INCONCLUSIVE)
|
||||
|
||||
def test_repetitions_that_disagree_settle_nothing(self) -> None:
|
||||
record = boot_record([rep(["logo", "title"]), rep(["black"], ["black"])],
|
||||
[rep(self.GAME, ["black"])])
|
||||
self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status,
|
||||
rules.INCONCLUSIVE)
|
||||
|
||||
def test_a_reference_that_shows_nothing_settles_nothing(self) -> None:
|
||||
record = boot_record([rep(["logo", "title"])], [rep(["black"], ["black"])])
|
||||
self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status,
|
||||
rules.INCONCLUSIVE)
|
||||
record = boot_record([rep(["logo", "title"])], [rep(self.GAME, loaded=False)])
|
||||
self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status,
|
||||
rules.INCONCLUSIVE)
|
||||
|
||||
|
||||
class RuleTable(unittest.TestCase):
|
||||
ENTRY = MappingProxyType({"name": "bios.bin", "required": True, "aliases": ["Bios.bin"]})
|
||||
RID = "2026-10-27-bbbbbbbbbbbb"
|
||||
|
||||
def test_boots_without_settles_hle_and_queues_required(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "x", 3), self.ENTRY, self.RID)
|
||||
self.assertEqual([(v.field, v.value, v.apply) for v in verdicts],
|
||||
[("hle_fallback", True, True), ("required", False, False)])
|
||||
self.assertEqual(verdicts[0].evidence, (self.RID,))
|
||||
optional = rules.rule_bios_need(rules.Boot(rules.BOOTS, "x", 3),
|
||||
{"name": "a", "required": False}, self.RID)
|
||||
self.assertEqual([v.field for v in optional], ["hle_fallback"])
|
||||
|
||||
def test_fails_without_settles_required_and_no_hle(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.FAILS, "x"), self.ENTRY, self.RID)
|
||||
self.assertEqual([(v.field, v.value, v.apply) for v in verdicts],
|
||||
[("required", True, True), ("hle_fallback", False, True)])
|
||||
|
||||
def test_inconclusive_is_queued(self) -> None:
|
||||
(verdict,) = rules.rule_bios_need(rules.Boot(rules.INCONCLUSIVE, "x"), self.ENTRY, self.RID)
|
||||
self.assertFalse(verdict.apply)
|
||||
|
||||
def test_path_and_alias(self) -> None:
|
||||
self.assertIsNone(rules.rule_path(self.ENTRY, "bios.bin", rules.Opened("bios.bin", self.RID)))
|
||||
moved = rules.rule_path(self.ENTRY, "bios.bin", rules.Opened("sys/bios.bin", self.RID))
|
||||
self.assertEqual((moved.field, moved.value, moved.apply), ("path", "sys/bios.bin", True))
|
||||
self.assertIsNone(rules.rule_alias(self.ENTRY, rules.Opened("Bios.bin", self.RID)))
|
||||
alias = rules.rule_alias(self.ENTRY, rules.Opened("BIOS.BIN", self.RID))
|
||||
self.assertEqual((alias.field, alias.value), ("aliases", ["Bios.bin", "BIOS.BIN"]))
|
||||
|
||||
def test_never_opened_is_flagged_not_applied(self) -> None:
|
||||
verdict = rules.rule_never_opened(self.ENTRY, [self.RID], "Game (USA)")
|
||||
self.assertFalse(verdict.apply)
|
||||
self.assertEqual(verdict.rule, "never_opened")
|
||||
|
||||
def test_variants(self) -> None:
|
||||
self.assertIsNone(rules.rule_variants("same", ["a", "b"], self.ENTRY, self.RID))
|
||||
self.assertFalse(rules.rule_variants("differs", ["b", "a"], self.ENTRY, self.RID).apply)
|
||||
|
||||
def test_agreement_and_evidence(self) -> None:
|
||||
self.assertTrue(rules.agrees({"name": "a"}, rules.Verdict("r", "hle_fallback", False, "", ())))
|
||||
self.assertFalse(rules.agrees({"name": "a"}, rules.Verdict("r", "hle_fallback", True, "", ())))
|
||||
self.assertTrue(rules.agrees(self.ENTRY, rules.Verdict("r", "aliases", ["Bios.bin"], "", ())))
|
||||
merged = rules.merge_evidence({"required": ["x"]}, rules.Verdict("r", "required", True, "", ("y", "x")))
|
||||
self.assertEqual(merged, {"required": ["x", "y"]})
|
||||
|
||||
|
||||
class ProfileEditor(unittest.TestCase):
|
||||
def test_replace_insert_and_keep_the_rest(self) -> None:
|
||||
out = campaign_profiles.set_fields(PROFILE, 0, {
|
||||
"required": False, "hle_fallback": True,
|
||||
"aliases": ["Bios.bin", "BIOS.BIN"], "evidence": {"hle_fallback": ["2026-10-27-cccccccccccc"]}})
|
||||
parsed = yaml.safe_load(out)
|
||||
self.assertEqual(parsed["files"][0]["hle_fallback"], True)
|
||||
self.assertEqual(parsed["files"][0]["aliases"], ["Bios.bin", "BIOS.BIN"])
|
||||
self.assertEqual(parsed["files"][1], {"name": "other.bin", "required": False})
|
||||
self.assertIn(" # -- the BIOS --", out)
|
||||
self.assertIn(' required: false\n hle_fallback: true\n', out)
|
||||
self.assertIn(' - "BIOS.BIN"\n', out)
|
||||
lines = out.splitlines()
|
||||
self.assertEqual(lines[lines.index(' evidence:') + 1], ' hle_fallback: ["2026-10-27-cccccccccccc"]')
|
||||
|
||||
def test_the_last_entry_and_a_no_op(self) -> None:
|
||||
out = campaign_profiles.set_fields(PROFILE, 1, {"path": "sub/other.bin"})
|
||||
self.assertEqual(yaml.safe_load(out)["files"][1]["path"], "sub/other.bin")
|
||||
self.assertEqual(campaign_profiles.set_fields(PROFILE, 1, {"required": False}), PROFILE)
|
||||
|
||||
def test_refusals(self) -> None:
|
||||
with self.assertRaises(campaign_profiles.ProfileEditError):
|
||||
campaign_profiles.set_fields(PROFILE, 2, {"required": False})
|
||||
with self.assertRaises(campaign_profiles.ProfileEditError):
|
||||
campaign_profiles.set_fields("emulator: X\n", 0, {"required": False})
|
||||
with self.assertRaises(campaign_profiles.ProfileEditError):
|
||||
campaign_profiles.set_fields(PROFILE, 0, {"name": "x.bin"})
|
||||
|
||||
def test_every_profile_reads_as_its_entries(self) -> None:
|
||||
# The text layout the editor reads holds in every profile it may edit.
|
||||
for path in sorted((REPO_ROOT / "emulators").glob("*.yml")):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
entries = (yaml.safe_load(text) or {}).get("files") or []
|
||||
if not entries or "\nfiles:\n" not in text:
|
||||
continue
|
||||
with self.subTest(profile=path.name):
|
||||
self.assertEqual(len(campaign_profiles.item_spans(text.splitlines())), len(entries))
|
||||
|
||||
|
||||
class FakeHost(campaign.Host):
|
||||
def __init__(self, loads: list[float] | None = None, free: float = 500.0) -> None:
|
||||
super().__init__()
|
||||
self.loads = list(loads or [0.0])
|
||||
self.now = 0.0
|
||||
self.slept: list[float] = []
|
||||
self.free = free
|
||||
self.calls: list[list[str]] = []
|
||||
self.results: list[object] = []
|
||||
self.loadavg = self._load
|
||||
self.sleep = self._sleep
|
||||
self.clock = lambda: self.now
|
||||
self.free_gb = lambda path: self.free
|
||||
self.cpu_count = 4
|
||||
self.run = self._run
|
||||
|
||||
def _load(self) -> tuple[float, float, float]:
|
||||
value = self.loads.pop(0) if len(self.loads) > 1 else self.loads[0]
|
||||
return (value, value, value)
|
||||
|
||||
def _sleep(self, seconds: float) -> None:
|
||||
self.slept.append(seconds)
|
||||
self.now += seconds
|
||||
|
||||
def _run(self, argv, **kwargs):
|
||||
self.calls.append(list(argv))
|
||||
result = self.results.pop(0) if len(self.results) > 1 else self.results[0]
|
||||
if isinstance(result, BaseException):
|
||||
raise result
|
||||
return result
|
||||
|
||||
|
||||
class Workspace(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
TMP.mkdir(parents=True, exist_ok=True)
|
||||
self.tmp = Path(tempfile.mkdtemp(dir=TMP))
|
||||
self.root = self.tmp / "repo"
|
||||
(self.root / "emulators").mkdir(parents=True)
|
||||
subprocess.run(["git", "init", "-q", str(self.root)], check=True)
|
||||
(self.root / "emulators" / "fake.yml").write_text(PROFILE, encoding="utf-8")
|
||||
bios = self.root / "bios" / "Fake" / "bios.bin"
|
||||
bios.parent.mkdir(parents=True)
|
||||
bios.write_bytes(b"BIOS")
|
||||
sha1 = hashlib.sha1(b"BIOS").hexdigest()
|
||||
db = {"files": {sha1: {"path": "bios/Fake/bios.bin", "name": "bios.bin", "size": 4,
|
||||
"sha1": sha1, "md5": hashlib.md5(b"BIOS").hexdigest()}},
|
||||
"indexes": {"by_name": {"bios.bin": [sha1]}, "by_md5": {}, "by_crc32": {},
|
||||
"by_sha256": {}, "by_path_suffix": {}}}
|
||||
(self.root / "database.json").write_text(json.dumps(db), encoding="utf-8")
|
||||
self.content_file = self.tmp / "content.yml"
|
||||
self.content_file.write_text(yaml.safe_dump({"systems": {"fake-system": {
|
||||
"romfetch": "fakesys", "content": [{
|
||||
"name": "Fake Game (World)", "catalog": "no-intro", "reason": "a test",
|
||||
"frames": 12, "frame_hash_every": 2}]}}}), encoding="utf-8")
|
||||
self.game = self.tmp / "content" / "fakesys" / "Fake Game (World)"
|
||||
self.paths = campaign.Paths(root=self.root, home=self.tmp / "home",
|
||||
content=self.tmp / "content", db=self.root / "database.json",
|
||||
content_file=self.content_file)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
shutil.rmtree(self.tmp, ignore_errors=True)
|
||||
|
||||
def put_game(self, folder: Path | None = None, payload: bytes = b"GAME") -> Path:
|
||||
folder = folder or self.game
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
(folder / "game.bin").write_bytes(payload)
|
||||
(folder / "identity.json").write_text(json.dumps({"status": "verified", "files": [
|
||||
{"name": "game.bin", "size": len(payload), "sha1": hashlib.sha1(payload).hexdigest()}]}))
|
||||
return folder
|
||||
|
||||
def make(self, host: campaign.Host | None = None, **settings: object) -> campaign.Campaign:
|
||||
values = {"delays": (1, 2), "echo": False, **settings}
|
||||
return campaign.Campaign(self.paths, campaign.Settings(**values), host or FakeHost(),
|
||||
lock_path=self.tmp / "auto.lock")
|
||||
|
||||
|
||||
class StateAndResume(Workspace):
|
||||
def test_should_run(self) -> None:
|
||||
self.assertTrue(campaign.should_run(None, "a", "d"))
|
||||
done = {"status": "done", "core_sha256": "a", "profile_digest": "d"}
|
||||
self.assertFalse(campaign.should_run(done, "a", "d"))
|
||||
self.assertTrue(campaign.should_run(done, "b", "d"))
|
||||
self.assertTrue(campaign.should_run(done, "a", "e"))
|
||||
self.assertTrue(campaign.should_run({**done, "status": "pending"}, "a", "d"))
|
||||
self.assertFalse(campaign.should_run({**done, "status": "failed"}, "a", "d"))
|
||||
|
||||
def test_evidence_does_not_change_the_digest(self) -> None:
|
||||
profile = yaml.safe_load(PROFILE)
|
||||
before = campaign.profile_digest(profile)
|
||||
profile["files"][0]["evidence"] = {"required": ["2026-10-27-dddddddddddd"]}
|
||||
self.assertEqual(campaign.profile_digest(profile), before)
|
||||
profile["files"][0]["required"] = False
|
||||
self.assertNotEqual(campaign.profile_digest(profile), before)
|
||||
|
||||
def test_state_survives_a_reload(self) -> None:
|
||||
c = self.make()
|
||||
c.pending("fake", "fake/fake-system/x/files", "content: romfetch not found")
|
||||
campaign.save_state(self.paths.state, c.state)
|
||||
again = self.make()
|
||||
self.assertEqual(again.state["checks"]["fake"]["fake/fake-system/x/files"]["status"], "pending")
|
||||
|
||||
def test_budget(self) -> None:
|
||||
self.assertEqual(campaign.parse_budget("6h"), 21600)
|
||||
self.assertEqual(campaign.parse_budget("90m"), 5400)
|
||||
self.assertEqual(campaign.parse_budget("30"), 30)
|
||||
|
||||
|
||||
class Guards(Workspace):
|
||||
def test_busy_host_waits_then_runs(self) -> None:
|
||||
host = FakeHost(loads=[5.0, 5.0, 0.5])
|
||||
c = self.make(host, max_load_wait=600)
|
||||
c.deadline = 10_000
|
||||
c.guard()
|
||||
self.assertEqual(host.slept, [60.0, 60.0])
|
||||
|
||||
def test_busy_host_pauses(self) -> None:
|
||||
host = FakeHost(loads=[5.0])
|
||||
c = self.make(host, max_load_wait=120)
|
||||
c.deadline = 10_000
|
||||
with self.assertRaisesRegex(campaign.Pause, "busy host"):
|
||||
c.guard()
|
||||
|
||||
def test_full_disk_pauses(self) -> None:
|
||||
c = self.make(FakeHost(free=10.0), min_free_gb=30)
|
||||
c.deadline = 10_000
|
||||
with self.assertRaisesRegex(campaign.Pause, "disk"):
|
||||
c.guard()
|
||||
|
||||
def test_budget_pauses(self) -> None:
|
||||
c = self.make()
|
||||
c.deadline = 0
|
||||
with self.assertRaisesRegex(campaign.Pause, "budget"):
|
||||
c.guard()
|
||||
|
||||
def test_the_lock_is_skipped_under_autorun(self) -> None:
|
||||
lock = self.tmp / "held.lock"
|
||||
with campaign.autorun.global_lock(lock):
|
||||
with self.assertRaises(campaign.autorun.LockBusy), campaign.host_lock(lock):
|
||||
pass
|
||||
os.environ[campaign.LOCK_HELD_ENV] = "1"
|
||||
try:
|
||||
with campaign.host_lock(lock):
|
||||
pass
|
||||
finally:
|
||||
del os.environ[campaign.LOCK_HELD_ENV]
|
||||
|
||||
|
||||
class ChildEnvironment(unittest.TestCase):
|
||||
def test_probes_builds_and_romfetch_get_no_secret(self) -> None:
|
||||
host = {"GITHUB_TOKEN": "t", "GH_TOKEN": "t", "ROMSTATION_USER": "u", "ROMSTATION_PASS": "p",
|
||||
"ROMFETCH_CONTENT_DIR": "/c", "SSH_AUTH_SOCK": "/s", "PATH": "/usr/bin"}
|
||||
with unittest.mock.patch.dict(os.environ, host):
|
||||
probe, fetch = campaign.child_env(), campaign.child_env(romfetch=True)
|
||||
for env in (probe, fetch):
|
||||
self.assertFalse({"GITHUB_TOKEN", "GH_TOKEN", "ROMSTATION_USER", "ROMSTATION_PASS",
|
||||
"SSH_AUTH_SOCK"} & set(env))
|
||||
self.assertEqual(env["PATH"], "/usr/bin")
|
||||
self.assertNotIn("ROMFETCH_CONTENT_DIR", probe)
|
||||
self.assertEqual(fetch["ROMFETCH_CONTENT_DIR"], "/c")
|
||||
|
||||
def test_the_probe_is_started_with_that_environment(self) -> None:
|
||||
seen: list[dict] = []
|
||||
host = FakeHost()
|
||||
host.run = lambda argv, **kw: (seen.append(kw.get("env")),
|
||||
subprocess.CompletedProcess([], 0, '{"id": "x"}', ""))[1]
|
||||
with tempfile.TemporaryDirectory(dir=TMP) as tmp:
|
||||
paths = campaign.Paths(root=Path(tmp), home=Path(tmp) / "h", content=Path(tmp) / "c",
|
||||
db=Path(tmp) / "db.json", content_file=campaign.CONTENT_FILE)
|
||||
c = campaign.Campaign(paths, campaign.Settings(echo=False), host, lock_path=Path(tmp) / "l")
|
||||
c.deadline = 10_000
|
||||
with unittest.mock.patch.dict(os.environ, {"GITHUB_TOKEN": "t"}):
|
||||
c.probe(["run"], "x")
|
||||
self.assertNotIn("GITHUB_TOKEN", seen[0])
|
||||
|
||||
|
||||
class FailClosed(Workspace):
|
||||
"""No core and no build runs outside the sandbox unless asked."""
|
||||
|
||||
def test_a_pass_without_a_sandbox_is_refused(self) -> None:
|
||||
host = FakeHost()
|
||||
host.results = [subprocess.CompletedProcess([], 0, "{}", "")]
|
||||
c = self.make(host)
|
||||
with unittest.mock.patch.object(campaign.core_probe.shutil, "which", return_value=None):
|
||||
current = c.run(set(), None)
|
||||
self.assertEqual(current.result, "refused")
|
||||
self.assertIn("bwrap not found", current.detail)
|
||||
self.assertEqual(host.calls, [])
|
||||
self.assertEqual(mr.read_campaign(self.root, mr.CAMPAIGN_RUNS)[-1]["result"], "refused")
|
||||
|
||||
def test_the_override_lets_it_run(self) -> None:
|
||||
c = self.make(allow_unsandboxed=True)
|
||||
c.unit_cores = lambda *a, **k: []
|
||||
with unittest.mock.patch.object(campaign.core_probe.shutil, "which", return_value=None):
|
||||
self.assertEqual(c.run(set(), None).result, "complete")
|
||||
|
||||
def test_every_probe_requires_the_sandbox(self) -> None:
|
||||
core = campaign.Core("fake", self.tmp / "x.so", "a" * 64)
|
||||
self.assertIn("--require-sandbox", self.make().probe_args("run", core, "fake", "s", None, None))
|
||||
self.assertNotIn("--require-sandbox", self.make(allow_unsandboxed=True).probe_args(
|
||||
"run", core, "fake", "s", None, None))
|
||||
|
||||
def test_a_build_without_a_sandbox_is_refused(self) -> None:
|
||||
c = self.make()
|
||||
src = c.paths.home / "src" / campaign.slug("https://example.invalid/x")
|
||||
(src / ".git").mkdir(parents=True)
|
||||
(src / "Makefile").write_text("all:\n")
|
||||
host = c.host
|
||||
host.results = [subprocess.CompletedProcess([], 0, "", "")]
|
||||
with unittest.mock.patch.object(campaign.core_probe.shutil, "which", return_value=None), \
|
||||
self.assertRaisesRegex(campaign.Unavailable, "refused: no sandbox"):
|
||||
c.cores.compile("https://example.invalid/x", "a" * 40, self.tmp / "out")
|
||||
self.assertFalse(any(call[0] == "make" for call in host.calls))
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which("git") and shutil.which("make"), "no git or make")
|
||||
class BuildIsolation(unittest.TestCase):
|
||||
"""Builds run on an archive exported from bare mirrors that only host git
|
||||
touches: a build cannot reach the mirror, and host git runs nothing a
|
||||
repository's config or attributes name."""
|
||||
|
||||
MAIN = "https://example.invalid/team/main.git"
|
||||
SUB = "https://example.invalid/team/sub.git"
|
||||
|
||||
def setUp(self) -> None:
|
||||
TMP.mkdir(parents=True, exist_ok=True)
|
||||
self.tmp = Path(tempfile.mkdtemp(dir=TMP))
|
||||
self.addCleanup(shutil.rmtree, self.tmp, True)
|
||||
self.pwned = self.tmp / "PWNED"
|
||||
sub = self.repo("sub", {"sub.c": "int x;\n"})
|
||||
self.sub_sha = self.git(sub, "rev-parse", "HEAD").strip()
|
||||
self.mirrors = {self.SUB: self.bare(sub, "sub")}
|
||||
main_mirror = self.tmp / "mirrors" / "main.git"
|
||||
# A Makefile that tries to plant a hook and a config line in the mirror.
|
||||
makefile = (
|
||||
"all:\n"
|
||||
f"\t-mkdir -p {main_mirror}/hooks\n"
|
||||
f"\t-echo 'touch {self.pwned}' > {main_mirror}/hooks/post-checkout\n"
|
||||
f"\t-printf '[core]\\n\\tfsmonitor = touch {self.pwned}\\n' >> {main_mirror}/config\n"
|
||||
"\ttouch fake_libretro.so\n")
|
||||
main = self.repo("main", {"Makefile": makefile, ".gitmodules":
|
||||
'[submodule "lib"]\n\tpath = lib\n\turl = ../sub.git\n'})
|
||||
self.git(main, "update-index", "--add", "--cacheinfo", f"160000,{self.sub_sha},lib")
|
||||
self.git(main, "commit", "-q", "-m", "submodule")
|
||||
self.commit = self.git(main, "rev-parse", "HEAD").strip()
|
||||
self.mirrors[self.MAIN] = self.bare(main, "main")
|
||||
self.config = (main_mirror / "config").read_text()
|
||||
|
||||
def git(self, repo: Path, *args: str) -> str:
|
||||
return subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", "-C", str(repo), *args],
|
||||
check=True, capture_output=True, text=True).stdout
|
||||
|
||||
def repo(self, name: str, files: dict[str, str]) -> Path:
|
||||
path = self.tmp / "work" / name
|
||||
subprocess.run(["git", "init", "-q", str(path)], check=True)
|
||||
for rel, text in files.items():
|
||||
(path / rel).write_text(text)
|
||||
self.git(path, "add", *files)
|
||||
self.git(path, "commit", "-q", "-m", "one")
|
||||
return path
|
||||
|
||||
def bare(self, work: Path, name: str) -> Path:
|
||||
path = self.tmp / "mirrors" / f"{name}.git"
|
||||
subprocess.run(["git", "clone", "-q", "--bare", str(work), str(path)], check=True)
|
||||
return path
|
||||
|
||||
def test_the_export_holds_the_tree_and_its_submodule_and_no_git(self) -> None:
|
||||
tree = self.tmp / "tree"
|
||||
campaign.export_commit(self.mirrors.__getitem__, self.MAIN, self.commit, tree)
|
||||
self.assertEqual((tree / "lib" / "sub.c").read_text(), "int x;\n")
|
||||
self.assertTrue((tree / "Makefile").is_file())
|
||||
self.assertEqual([p for p in tree.rglob(".git")], [])
|
||||
|
||||
def test_a_build_cannot_reach_the_mirror(self) -> None:
|
||||
tree = self.tmp / "tree"
|
||||
prefix, isolation = campaign.core_probe.sandbox_argv([tree])
|
||||
if isolation != campaign.core_probe.SANDBOX:
|
||||
self.skipTest("no working bwrap on this host")
|
||||
campaign.export_commit(self.mirrors.__getitem__, self.MAIN, self.commit, tree)
|
||||
proc = subprocess.run(prefix + ["make"], cwd=tree, capture_output=True, text=True,
|
||||
check=False, env=campaign.child_env())
|
||||
self.assertEqual(proc.returncode, 0, proc.stderr)
|
||||
mirror = self.mirrors[self.MAIN]
|
||||
self.assertFalse((mirror / "hooks" / "post-checkout").exists())
|
||||
self.assertEqual((mirror / "config").read_text(), self.config)
|
||||
campaign.export_commit(self.mirrors.__getitem__, self.MAIN, self.commit, self.tmp / "again")
|
||||
self.assertFalse(self.pwned.exists())
|
||||
|
||||
def test_host_git_runs_nothing_a_repository_config_names(self) -> None:
|
||||
mirror = self.mirrors[self.MAIN]
|
||||
hook = mirror / "hooks" / "reference-transaction"
|
||||
hook.write_text(f"#!/bin/sh\ntouch {self.pwned}\n")
|
||||
hook.chmod(0o755)
|
||||
with (mirror / "config").open("a") as handle:
|
||||
handle.write(f"[core]\n\tfsmonitor = touch {self.pwned}\n\tsshCommand = touch {self.pwned}\n"
|
||||
f'[filter "x"]\n\tsmudge = touch {self.pwned}\n')
|
||||
(mirror / "info").mkdir(exist_ok=True)
|
||||
(mirror / "info" / "attributes").write_text("* filter=x\n")
|
||||
campaign.export_commit(self.mirrors.__getitem__, self.MAIN, self.commit, self.tmp / "tree")
|
||||
campaign.git_out("-C", str(mirror), "update-ref", "refs/x", self.commit)
|
||||
self.assertFalse(self.pwned.exists())
|
||||
# Plain git on the same mirror runs the planted hook: the test plants a live one.
|
||||
subprocess.run(["git", "-C", str(mirror), "update-ref", "refs/y", self.commit], check=True)
|
||||
self.assertTrue(self.pwned.exists())
|
||||
|
||||
def test_only_https_mirrors(self) -> None:
|
||||
c = campaign.Campaign(campaign.Paths(root=self.tmp, home=self.tmp / "home", content=self.tmp,
|
||||
db=self.tmp / "db.json", content_file=campaign.CONTENT_FILE),
|
||||
campaign.Settings(echo=False), FakeHost(), lock_path=self.tmp / "l")
|
||||
for url in ("file:///etc", "/home/x/repo", "ext::sh -c touch", "ssh://h/r", "git@h:r"):
|
||||
with self.subTest(url=url), self.assertRaises(campaign.Unavailable):
|
||||
c.cores.mirror(url)
|
||||
self.assertEqual(campaign.submodule_url(self.MAIN, "../sub.git"), self.SUB)
|
||||
self.assertEqual(campaign.git_env()["GIT_ALLOW_PROTOCOL"], "https")
|
||||
|
||||
def test_a_whole_build_wipes_old_clones_and_copies_a_regular_file(self) -> None:
|
||||
if campaign.core_probe.sandbox_argv([])[1] != campaign.core_probe.SANDBOX:
|
||||
self.skipTest("no working bwrap on this host")
|
||||
home = self.tmp / "home"
|
||||
(home / "src" / "old" / ".git").mkdir(parents=True)
|
||||
c = campaign.Campaign(campaign.Paths(root=self.tmp, home=home, content=self.tmp,
|
||||
db=self.tmp / "db.json", content_file=campaign.CONTENT_FILE),
|
||||
campaign.Settings(echo=False), campaign.Host(), lock_path=self.tmp / "l")
|
||||
c.cores.mirror = self.mirrors.__getitem__
|
||||
so, command = c.cores.compile(self.MAIN, self.commit, self.tmp / "out")
|
||||
self.assertEqual(so, self.tmp / "out" / "fake_libretro.so")
|
||||
self.assertIn("isolation: bwrap", command)
|
||||
self.assertFalse((home / "src").exists())
|
||||
|
||||
|
||||
class BuiltCore(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
TMP.mkdir(parents=True, exist_ok=True)
|
||||
self.tree = Path(tempfile.mkdtemp(dir=TMP))
|
||||
self.addCleanup(shutil.rmtree, self.tree, True)
|
||||
self.secret = self.tree.parent / f"{self.tree.name}-host-file"
|
||||
self.secret.write_text("host secret")
|
||||
self.addCleanup(self.secret.unlink)
|
||||
|
||||
def test_a_link_is_never_taken_for_the_core(self) -> None:
|
||||
(self.tree / "a_libretro.so").symlink_to(self.secret)
|
||||
with self.assertRaises(campaign.Unavailable):
|
||||
campaign.built_core(self.tree)
|
||||
(self.tree / "sub").mkdir()
|
||||
(self.tree / "sub" / "b_libretro.so").write_bytes(b"\x7fELF")
|
||||
self.assertEqual(campaign.built_core(self.tree), self.tree / "sub" / "b_libretro.so")
|
||||
|
||||
def test_copy_refuses_a_link(self) -> None:
|
||||
link = self.tree / "c_libretro.so"
|
||||
link.symlink_to(self.secret)
|
||||
with self.assertRaises(OSError):
|
||||
campaign.copy_regular(link, self.tree / "out.so")
|
||||
self.assertFalse((self.tree / "out.so").exists())
|
||||
|
||||
|
||||
class ProbeRuns(Workspace):
|
||||
def completed(self, code: int, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
|
||||
return subprocess.CompletedProcess([], code, stdout, stderr)
|
||||
|
||||
def probe(self, *results: object) -> tuple[campaign.ProbeResult, FakeHost]:
|
||||
host = FakeHost()
|
||||
host.results = list(results)
|
||||
c = self.make(host)
|
||||
c.deadline = 10_000
|
||||
return c.probe(["run", "--core", "x.so"], "fake-run"), host
|
||||
|
||||
def test_timeout_is_failed_and_not_retried(self) -> None:
|
||||
result, host = self.probe(subprocess.TimeoutExpired(["x"], 1))
|
||||
self.assertEqual((result.status, result.reason), ("failed", "timeout"))
|
||||
self.assertEqual(len(host.calls), 1)
|
||||
self.assertIn("killed after", Path(result.log).read_text())
|
||||
result, host = self.probe(self.completed(1, stderr="subprocess.TimeoutExpired: timed out after 900 seconds"))
|
||||
self.assertEqual((result.reason, len(host.calls)), ("timeout", 1))
|
||||
|
||||
def test_a_crashing_core_is_tried_three_times(self) -> None:
|
||||
result, host = self.probe(self.completed(1, stderr="worker exited -11"))
|
||||
self.assertEqual((result.status, result.reason), ("failed", "crashed 3 times"))
|
||||
self.assertEqual(len(host.calls), 3)
|
||||
|
||||
def test_a_crash_then_a_record(self) -> None:
|
||||
record = {"id": "2026-10-27-eeeeeeeeeeee"}
|
||||
result, host = self.probe(self.completed(1, stderr="worker exited -11"),
|
||||
self.completed(0, stdout=json.dumps(record)))
|
||||
self.assertEqual((result.status, result.record, len(host.calls)), ("done", record, 2))
|
||||
|
||||
def test_refusal_on_a_busy_host_pauses(self) -> None:
|
||||
with self.assertRaisesRegex(campaign.Pause, "busy host"):
|
||||
self.probe(self.completed(3, stderr="load average"))
|
||||
|
||||
|
||||
class Content(Workspace):
|
||||
def content(self, name: str = "Fake Game (World)", pinned: bool = False) -> campaign.Content:
|
||||
return campaign.Content("fake-system", "fakesys", name, "no-intro", "r", None, "", 12, 2,
|
||||
pinned, None)
|
||||
|
||||
def test_local_content_is_verified_by_its_identity(self) -> None:
|
||||
self.put_game()
|
||||
c = self.make()
|
||||
self.assertEqual(c.store.get(self.content()), self.game)
|
||||
(self.game / "game.bin").write_bytes(b"EDIT")
|
||||
with self.assertRaisesRegex(campaign.Unavailable, "does not match"):
|
||||
self.make().store.get(self.content())
|
||||
|
||||
def test_without_romfetch_the_check_waits_for_content(self) -> None:
|
||||
c = self.make(romfetch=str(self.tmp / "absent-romfetch"))
|
||||
with self.assertRaisesRegex(campaign.Unavailable, "romfetch not found"):
|
||||
c.store.get(self.content())
|
||||
|
||||
def test_romfetch_is_retried_on_a_source_error(self) -> None:
|
||||
fetcher = self.tmp / "romfetch"
|
||||
fetcher.write_text("#!/bin/sh\n")
|
||||
fetcher.chmod(0o755)
|
||||
host = FakeHost()
|
||||
error = json.dumps({"error": {"code": "network", "message": "m", "hint": "h"}})
|
||||
game = self.game
|
||||
|
||||
def fetched(*args, **kwargs):
|
||||
self.put_game(game)
|
||||
return subprocess.CompletedProcess([], 0, '{"status": "verified"}', "")
|
||||
host.results = [subprocess.CompletedProcess([], 3, error, ""), None]
|
||||
host.run = lambda argv, **kw: (host.calls.append(argv), fetched() if len(host.calls) == 2
|
||||
else host.results[0])[1]
|
||||
c = self.make(host, romfetch=str(fetcher))
|
||||
self.assertEqual(c.store.get(self.content()), self.game)
|
||||
self.assertEqual(host.slept, [1])
|
||||
self.assertEqual(host.calls[0][1:4], ["get", "fakesys", "Fake Game (World)"])
|
||||
self.assertTrue(c.state["content"]["fakesys/Fake Game (World)"]["fetched_by_campaign"])
|
||||
|
||||
def test_romfetch_not_found_names_its_code(self) -> None:
|
||||
fetcher = self.tmp / "romfetch"
|
||||
fetcher.write_text("#!/bin/sh\n")
|
||||
host = FakeHost()
|
||||
host.results = [subprocess.CompletedProcess(
|
||||
[], 1, json.dumps({"error": {"code": "no_entry", "message": "m", "hint": "h"}}), "")]
|
||||
c = self.make(host, romfetch=str(fetcher))
|
||||
with self.assertRaisesRegex(campaign.Unavailable, r"exit 1 \(no_entry\)"):
|
||||
c.store.get(self.content())
|
||||
self.assertEqual(len(host.calls), 1)
|
||||
|
||||
def test_shared_content_stays_until_its_last_user(self) -> None:
|
||||
c = self.make()
|
||||
self.put_game()
|
||||
content = self.content()
|
||||
c.state["content"][content.key] = {"fetched_by_campaign": True, "pinned": False, "size": 4}
|
||||
c.store.hold(content, "alpha")
|
||||
c.store.hold(content, "beta")
|
||||
c.store.release(content, "alpha")
|
||||
self.assertTrue(self.game.is_dir())
|
||||
c.store.release(content, "beta")
|
||||
self.assertFalse(self.game.exists())
|
||||
|
||||
def test_pinned_or_foreign_content_is_never_deleted(self) -> None:
|
||||
c = self.make()
|
||||
self.put_game()
|
||||
content = self.content(pinned=True)
|
||||
c.state["content"][content.key] = {"fetched_by_campaign": True, "pinned": True}
|
||||
c.store.hold(content, "alpha")
|
||||
c.store.release(content, "alpha")
|
||||
self.assertTrue(self.game.is_dir())
|
||||
c.state["content"][content.key] = {"fetched_by_campaign": False, "pinned": False}
|
||||
c.store.hold(content, "alpha")
|
||||
c.store.release(content, "alpha")
|
||||
self.assertTrue(self.game.is_dir())
|
||||
|
||||
def test_least_recently_used_goes_first_past_the_cap(self) -> None:
|
||||
c = self.make(cache_cap_gb=5e-9)
|
||||
for name, used in (("Old", "2026-01-01"), ("New", "2026-06-01"), ("Kept", "2025-01-01")):
|
||||
self.put_game(self.tmp / "content" / "fakesys" / name)
|
||||
c.state["content"][f"fakesys/{name}"] = {"fetched_by_campaign": True, "pinned": name == "Kept",
|
||||
"size": 4, "last_used": used}
|
||||
self.assertEqual(c.store.evict(keep=set()), ["fakesys/Old"])
|
||||
self.assertTrue((self.tmp / "content" / "fakesys" / "New").is_dir())
|
||||
self.assertTrue((self.tmp / "content" / "fakesys" / "Kept").is_dir())
|
||||
|
||||
def test_load_file(self) -> None:
|
||||
self.put_game()
|
||||
self.assertEqual(campaign.load_file(self.game, ""), self.game / "game.bin")
|
||||
with self.assertRaises(campaign.Unavailable):
|
||||
campaign.load_file(self.game, "cue")
|
||||
|
||||
|
||||
class Findings(Workspace):
|
||||
RID = "2026-10-27-ffffffffffff"
|
||||
|
||||
def setUp(self) -> None:
|
||||
super().setUp()
|
||||
self.c = self.make()
|
||||
self.c.current = campaign.Pass(started=campaign.now_iso())
|
||||
self.content = campaign.Content("fake-system", "fakesys", "Fake Game (World)", "no-intro",
|
||||
"r", None, "", 12, 2, False, None)
|
||||
self.core = campaign.Core("fake", self.tmp / "fake_libretro.so", "a" * 64)
|
||||
self.profile = self.root / "emulators" / "fake.yml"
|
||||
|
||||
def lines(self) -> list[dict]:
|
||||
return mr.read_campaign(self.root, mr.FINDINGS)
|
||||
|
||||
def entry(self, index: int = 0) -> dict:
|
||||
return yaml.safe_load(self.profile.read_text())["files"][index]
|
||||
|
||||
def test_a_settled_field_is_written_with_its_evidence(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID)
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts)
|
||||
entry = self.entry()
|
||||
self.assertEqual(entry["hle_fallback"], True)
|
||||
self.assertEqual(entry["evidence"], {"hle_fallback": [self.RID]})
|
||||
self.assertEqual(entry["required"], True)
|
||||
self.assertEqual([(line["field"], line["status"]) for line in self.lines()],
|
||||
[("hle_fallback", "applied"), ("required", "open")])
|
||||
self.assertIn("emulators/fake.yml", self.c.current.touched)
|
||||
# The same outcome a second time adds nothing.
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts)
|
||||
self.assertEqual(len(self.lines()), 2)
|
||||
|
||||
def test_an_agreeing_run_adds_evidence_only(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.FAILS, "nothing"), self.entry(), self.RID)
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts)
|
||||
entry = self.entry()
|
||||
self.assertEqual((entry["required"], entry["hle_fallback"]), (True, False))
|
||||
self.assertEqual(entry["evidence"], {"hle_fallback": [self.RID], "required": [self.RID]})
|
||||
self.assertEqual(self.lines(), [])
|
||||
|
||||
def test_a_later_agreement_resolves_the_open_finding(self) -> None:
|
||||
(verdict,) = rules.rule_bios_need(rules.Boot(rules.INCONCLUSIVE, "?"), self.entry(), self.RID)
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), [verdict])
|
||||
self.assertEqual(self.lines()[-1]["status"], "open")
|
||||
later = rules.Verdict("inconclusive", "hle_fallback", False, "agrees now", ("2026-10-28-000000000000",))
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), [later])
|
||||
self.assertEqual([line["status"] for line in self.lines()], ["open", "resolved"])
|
||||
|
||||
def test_never_opened_is_flagged(self) -> None:
|
||||
verdict = rules.rule_never_opened(self.entry(1), [self.RID], "Fake Game (World)")
|
||||
self.c.settle("fake", self.content, self.core, 1, self.entry(1), [verdict])
|
||||
self.assertEqual(self.lines()[0]["status"], "flagged")
|
||||
self.assertNotIn("evidence", self.entry(1))
|
||||
|
||||
def test_finding_lines_validate(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID)
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts)
|
||||
self.c.write_run_line(self.c.current)
|
||||
try:
|
||||
from jsonschema import Draft202012Validator
|
||||
except ImportError:
|
||||
self.skipTest("jsonschema is a tooling dependency")
|
||||
for name, (_, schema_name) in mr.CAMPAIGN_FILES.items():
|
||||
schema = json.loads((REPO_ROOT / "schemas" / schema_name).read_text())
|
||||
for line in mr.read_campaign(self.root, name):
|
||||
self.assertEqual(list(Draft202012Validator(schema).iter_errors(line)), [], name)
|
||||
|
||||
def test_the_batch_commit_plan(self) -> None:
|
||||
verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID)
|
||||
self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts)
|
||||
self.c.current.touched.add(mr.records_path("fake"))
|
||||
plan = self.c.plan_commits(set())
|
||||
self.assertEqual(plan, [
|
||||
("chore: add campaign measurements", ["measurements/fake.jsonl", "measurements/findings.jsonl"]),
|
||||
("chore: apply tested fields to fake", ["emulators/fake.yml"])])
|
||||
for subject, _ in plan:
|
||||
self.assertEqual(campaign.autorun.check_subject(subject), [], subject)
|
||||
# A file someone had changed before the pass is left to them.
|
||||
self.assertEqual(self.c.plan_commits({"emulators/fake.yml"})[1:], [])
|
||||
out = io.StringIO()
|
||||
with contextlib.redirect_stdout(out):
|
||||
self.assertEqual(self.c.commit_batch(dry_run=True), [])
|
||||
self.assertIn("would commit 'chore: add campaign measurements'", out.getvalue())
|
||||
|
||||
|
||||
class Registry(unittest.TestCase):
|
||||
def test_no_profile_takes_a_campaign_file_name(self) -> None:
|
||||
stems = {p.stem for p in (REPO_ROOT / "emulators").glob("*.yml")}
|
||||
self.assertEqual(stems & set(mr.CAMPAIGN_FILES), set())
|
||||
|
||||
def test_content_list(self) -> None:
|
||||
content = campaign.load_content(campaign.CONTENT_FILE)
|
||||
systems = {s for p in (REPO_ROOT / "emulators").glob("*.yml")
|
||||
for s in (yaml.safe_load(p.read_text()) or {}).get("systems") or ()}
|
||||
self.assertTrue(content)
|
||||
for system, items in content.items():
|
||||
self.assertIn(system, systems)
|
||||
for item in items:
|
||||
self.assertIn(item.catalog, ("redump", "no-intro"))
|
||||
self.assertGreater(len(item.reason), 40)
|
||||
self.assertGreater(item.frames, item.frame_hash_every)
|
||||
|
||||
def test_the_autorun_job(self) -> None:
|
||||
jobs = campaign.autorun.load_jobs()
|
||||
job = jobs["campaign"]
|
||||
self.assertTrue(job.lock and job.worktree and job.schedule)
|
||||
self.assertEqual(job.gate[0].argv, ("python", "scripts/campaign.py", "gate"))
|
||||
self.assertEqual({p for rule in job.commits for p in rule.paths},
|
||||
{"measurements/*.jsonl", "emulators/*.yml"})
|
||||
self.assertGreater(job.schedule, jobs["nightly"].schedule)
|
||||
|
||||
def test_autorun_tells_its_steps_the_lock_is_held(self) -> None:
|
||||
job = campaign.autorun.load_jobs()["campaign"]
|
||||
self.assertEqual(campaign.autorun.job_env(job)[campaign.LOCK_HELD_ENV], "1")
|
||||
|
||||
|
||||
class SitePage(unittest.TestCase):
|
||||
def test_last_pass_and_open_findings(self) -> None:
|
||||
summary = {
|
||||
"last_run": {"ended_at": "2026-10-27T10:00:00+00:00", "result": "complete", "detail": "",
|
||||
"counts": {"done": 12, "failed": 1, "pending": 3},
|
||||
"pending_reasons": {"standalone": 3},
|
||||
"failed": [{"emulator": "fake", "check": "fake/x/y/bench", "reason": "timeout"}]},
|
||||
"passes": 1, "applied": 2,
|
||||
"open": [{"emulator": "fake", "file": "bios.bin", "field": "required", "status": "open",
|
||||
"profile_says": True, "run_shows": False, "statement": "the core boots without it",
|
||||
"evidence": ["2026-10-27-aaaaaaaaaaaa"], "links": ["https://x/l#L1"]}],
|
||||
}
|
||||
page = "\n".join(sitemeasurements.campaign_section(summary))
|
||||
self.assertIn("12 tests done, 1 failed, 3 waiting", page)
|
||||
self.assertIn("| standalone | 3 |", page)
|
||||
self.assertIn("the profile says <code>true</code>, the run shows <code>false</code>", page)
|
||||
self.assertIn("[2026-10-27-aaaaaaaaaaaa](https://x/l#L1)", page)
|
||||
self.assertIn("<code>false</code>. The core boots without it. Records:", page)
|
||||
self.assertEqual(sitemeasurements.campaign_section({"last_run": None}), [])
|
||||
|
||||
def test_observed_values_cannot_leave_their_place(self) -> None:
|
||||
# A value comes from a core's output or a romfetch error: it may hold
|
||||
# backticks, MDX braces and HTML.
|
||||
hostile = "``}{import x}<script>alert(1)</script>`` | [a](javascript:x) *b*"
|
||||
summary = {
|
||||
"last_run": {"ended_at": "2026-10-27T10:00:00+00:00", "result": "paused", "detail": hostile,
|
||||
"counts": {"done": 0, "failed": 1, "pending": 1},
|
||||
"pending_reasons": {hostile: 1},
|
||||
"failed": [{"emulator": hostile, "check": hostile, "reason": hostile}]},
|
||||
"passes": 1, "applied": 0,
|
||||
"open": [{"emulator": "fake", "file": hostile, "field": hostile, "status": "open",
|
||||
"profile_says": {"k": hostile}, "run_shows": [hostile], "statement": hostile,
|
||||
"evidence": ["2026-10-27-aaaaaaaaaaaa"], "links": [None]}],
|
||||
}
|
||||
page = "\n".join(sitemeasurements.campaign_section(summary))
|
||||
for raw in ("<script", "{import", "}{", "``", "](javascript", "*b*", "| [a]"):
|
||||
self.assertNotIn(raw, page)
|
||||
self.assertIn("<script>alert(1)</script>", page)
|
||||
self.assertIn("{import x}", page)
|
||||
# The JSON dumps are escaped too: no brace reaches the page as syntax.
|
||||
self.assertNotIn("{", page)
|
||||
self.assertNotIn("}", page)
|
||||
|
||||
|
||||
@unittest.skipUnless(shutil.which("cc") and shutil.which("strace"), "no C compiler or no strace")
|
||||
class FakeCorePasses(Workspace):
|
||||
"""Whole passes against a core compiled from FAKE_CORE."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
TMP.mkdir(parents=True, exist_ok=True)
|
||||
cls.build = Path(tempfile.mkdtemp(dir=TMP))
|
||||
source = cls.build / "fake.c"
|
||||
source.write_text(FAKE_CORE, encoding="utf-8")
|
||||
# The core sees no variable of the host, so its behaviour is compiled in.
|
||||
cls.cores = {}
|
||||
for kind, flags in (("plain", []), ("hle", ["-DFAKE_HLE"]), ("hang", ["-DFAKE_HANG"])):
|
||||
cls.cores[kind] = cls.build / f"fake_{kind}_libretro.so"
|
||||
subprocess.run(["cc", "-shared", "-fPIC", *flags, "-o", str(cls.cores[kind]), str(source)],
|
||||
check=True)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
shutil.rmtree(cls.build, ignore_errors=True)
|
||||
|
||||
def passing(self, kind: str = "plain", core: Path | None = None,
|
||||
**settings: object) -> campaign.Campaign:
|
||||
self.put_game()
|
||||
# The core_probe threshold follows the host: a test never waits on the real load.
|
||||
host = campaign.Host(loadavg=lambda: (0.0, 0.0, 0.0), free_gb=lambda path: 500.0,
|
||||
cpu_count=1_000_000)
|
||||
values = {"bench_warmup": 2, "bench_frames": 4, "check_repeat": 2, "probe_timeout": 30,
|
||||
"delays": (1,), "echo": False, **settings}
|
||||
c = campaign.Campaign(self.paths, campaign.Settings(**values), host,
|
||||
lock_path=self.tmp / "auto.lock")
|
||||
so = core or self.cores[kind]
|
||||
c.unit_cores = lambda *a, **k: [campaign.Core("fake", so, campaign.file_sha256(so))]
|
||||
return c
|
||||
|
||||
def test_a_stand_in_is_proven_and_written(self) -> None:
|
||||
c = self.passing("hle")
|
||||
current = c.run(set(), None)
|
||||
self.assertEqual(current.result, "complete")
|
||||
records = [s.record for s in mr.read_records(self.root)]
|
||||
isolations = {rep["isolation"] for r in records for v in r["variants"] for rep in v["repetitions"]}
|
||||
self.assertEqual(isolations, {campaign.core_probe.SANDBOX})
|
||||
kinds = sorted(r["subject"].split("-")[0] for r in records)
|
||||
self.assertEqual(kinds, ["alias", "bench", "boot", "files"])
|
||||
boot = next(r for r in records if r["subject"].startswith("boot"))
|
||||
alias = next(r for r in records if r["subject"].startswith("alias"))
|
||||
entry = yaml.safe_load((self.root / "emulators" / "fake.yml").read_text())["files"][0]
|
||||
self.assertEqual(entry["hle_fallback"], True)
|
||||
self.assertEqual(entry["evidence"], {"aliases": [alias["id"]], "hle_fallback": [boot["id"]]})
|
||||
statuses = {(f["field"], f["status"]) for f in mr.read_campaign(self.root, mr.FINDINGS)}
|
||||
self.assertEqual(statuses, {("hle_fallback", "applied"), ("required", "open"),
|
||||
("aliases", "applied")})
|
||||
runs = mr.read_campaign(self.root, mr.CAMPAIGN_RUNS)
|
||||
self.assertEqual(runs[-1]["counts"]["done"], 4)
|
||||
# The game is pinned by nobody but was not fetched by the campaign: kept.
|
||||
self.assertTrue(self.game.is_dir())
|
||||
|
||||
# A second pass on the same core and profile runs nothing.
|
||||
again = self.passing("hle")
|
||||
self.assertEqual(again.run(set(), None).records, [])
|
||||
|
||||
# A new build of the core runs everything again.
|
||||
rebuilt = self.build / "fake2_libretro.so"
|
||||
shutil.copy2(self.cores["hle"], rebuilt)
|
||||
with rebuilt.open("ab") as handle:
|
||||
handle.write(b"\0")
|
||||
third = self.passing("hle", core=rebuilt)
|
||||
# BIOS.BIN is declared now, so no alias run follows the boot test.
|
||||
self.assertEqual(len(third.run(set(), None).records), 3)
|
||||
|
||||
def test_no_stand_in_settles_required_and_finds_an_alias(self) -> None:
|
||||
c = self.passing()
|
||||
c.run(set(), None)
|
||||
entry = yaml.safe_load((self.root / "emulators" / "fake.yml").read_text())["files"][0]
|
||||
self.assertEqual((entry["required"], entry["hle_fallback"]), (True, False))
|
||||
self.assertIn("BIOS.BIN", entry["aliases"])
|
||||
self.assertEqual(sorted(entry["evidence"]), ["aliases", "hle_fallback", "required"])
|
||||
alias = [f for f in mr.read_campaign(self.root, mr.FINDINGS) if f["field"] == "aliases"]
|
||||
self.assertEqual(alias[0]["status"], "applied")
|
||||
|
||||
def test_a_hung_core_is_killed_and_the_pass_moves_on(self) -> None:
|
||||
c = self.passing("hang", probe_timeout=2)
|
||||
c.run(set(), None)
|
||||
# The whole process group went with the run: no worker spins on.
|
||||
left = subprocess.run(["pgrep", "-f", str(self.cores["hang"])], capture_output=True, check=False)
|
||||
self.assertEqual(left.stdout, b"")
|
||||
checks = c.state["checks"]["fake"]
|
||||
self.assertTrue(checks)
|
||||
self.assertEqual({entry["status"] for entry in checks.values()}, {"failed"})
|
||||
self.assertEqual({entry["reason"] for entry in checks.values()}, {"timeout"})
|
||||
self.assertEqual(mr.read_records(self.root), [])
|
||||
|
||||
def test_missing_content_leaves_the_checks_pending(self) -> None:
|
||||
c = self.passing(romfetch=str(self.tmp / "no-romfetch"))
|
||||
shutil.rmtree(self.game)
|
||||
c.run(set(), None)
|
||||
reasons = {entry["reason"] for entry in c.state["checks"]["fake"].values()}
|
||||
self.assertEqual(len(reasons), 1)
|
||||
self.assertIn("romfetch not found", reasons.pop())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
+112
-7
@@ -11,12 +11,14 @@ 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
|
||||
@@ -99,6 +101,13 @@ def mednafen_state(sections: dict[str, dict[str, bytes]]) -> bytes:
|
||||
return b"MDFNSVST".ljust(32, b"\0") + body
|
||||
|
||||
|
||||
def checkout(path: Path) -> 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"))
|
||||
@@ -154,6 +163,12 @@ class ParsingTests(unittest.TestCase):
|
||||
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")
|
||||
@@ -238,7 +253,7 @@ class FakeCoreTests(unittest.TestCase):
|
||||
return code, json.loads(out.getvalue()) if code == 0 else None
|
||||
|
||||
def test_run_record(self) -> None:
|
||||
record_dir = self.tmp / "records"
|
||||
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=",
|
||||
@@ -262,11 +277,12 @@ class FakeCoreTests(unittest.TestCase):
|
||||
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()], ["measurements/fake.jsonl"])
|
||||
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, with its id.
|
||||
self.assertEqual({k: v for k, v in line.items() if k != "id"}, record)
|
||||
# The line is the record the run printed, id included.
|
||||
self.assertEqual(line, record)
|
||||
try:
|
||||
from jsonschema import Draft202012Validator
|
||||
except ImportError:
|
||||
@@ -275,7 +291,7 @@ class FakeCoreTests(unittest.TestCase):
|
||||
self.assertEqual(list(Draft202012Validator(schema).iter_errors(line)), [])
|
||||
|
||||
def test_second_run_is_appended_beside_the_first(self) -> None:
|
||||
record_dir = self.tmp / "appended"
|
||||
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)
|
||||
@@ -307,10 +323,99 @@ class FakeCoreTests(unittest.TestCase):
|
||||
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(self.tmp / "refused"), "--max-load", "-1"])
|
||||
str(refused), "--max-load", "-1"])
|
||||
self.assertEqual(code, 3)
|
||||
self.assertFalse((self.tmp / "refused").exists())
|
||||
self.assertFalse((refused / "measurements").exists())
|
||||
|
||||
def test_record_names_a_checkout_root(self) -> None:
|
||||
# --record measurements wrote measurements/measurements/<emulator>.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__":
|
||||
|
||||
Reference in new issue
Block a user