Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
52 kB · 1095 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096"""Tests for tools/build_accepted_fixture.py.
The accepted Rust-MOS fixtures must keep producing the exact bytes that wereaccepted (smoke ``dac6abb2...2560``, timeline ``bf5525d7...80b6``) even after thehost workspace version moves on. The build therefore composes a frozen buildroot from ``fixtures/accepted-sdk`` and refuses anything that would silentlychange the artifact.
These tests pin the safety properties without a container runtime or a network:the contract schema, the frozen SDK and fixture-source pins, symlink refusal, theexact container invocation and its root paths, staging versus publishing, and the"never publish on failure" rule. A container-backed check of the real hashes isopt-in through ``ULTIMATE_TOOL_ACCEPTED_FIXTURE_BUILD=1``.
A fake ``docker`` executable stands in for the pinned image. It records itsargv, writes a PRG/ELF/map into the composed root, and can be told to fail or toproduce the wrong bytes.
Fixture build outputs are generated and git-ignored, so a clean checkout hasnone. Each temporary repository therefore copies fixture sources only and seedsthe already-published outputs from the tracked ``examples/first-run`` snapshot,hash-checked against that snapshot's own record and, for the PRG, against theaccepted pin.
No accepted timeline artifact is tracked or reproducible here, so the timelinetests are mocks: they feed deterministic synthetic bytes and rewrite only thetemporary repository's contract copy to pin those bytes. They exercise theper-fixture hash routing and the verification modes, and claim no acceptedtimeline build. The tracked contract keeps pinning the real accepted timelineartifact, which ``ContractTests`` asserts directly."""
from __future__ import annotations
import hashlibimport jsonimport osimport shutilimport statimport subprocessimport sysimport tempfileimport unittestfrom pathlib import Pathfrom unittest import mock
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"sys.path.insert(0, str(TOOLS))
import build_accepted_fixture as baf # noqa: E402
ACCEPTED_SMOKE = "dac6abb2ea01752e94c4318463efa6c54942713ffb021b67f1a4e1d66fa02560"ACCEPTED_TIMELINE = "bf5525d7fcabfae8cdcb0d70df41b25c150ab9dff416c558cd94aba280a680b6"SDK_SOURCE = "b3a50613174c9be942961355fe86e50f2596fffb31e1cbcb3225d5e8601905e3"IMAGE_ID = "sha256:0225fce9df86792cf0fa459ba90d02cf3816c60b30e8cb0b2afad69ca1e211aa"REAL_BUILD_ENV = "ULTIMATE_TOOL_ACCEPTED_FIXTURE_BUILD"
# The approved first-run demo keeps a tracked copy of the accepted smoke build# outputs. A clean checkout has no ``fixtures/*/build`` directory at all (those# are generated and git-ignored), so tests that need already-published outputs# seed them from this recorded snapshot rather than from whatever this working# tree happens to hold.APPROVED_DEMO = ROOT / "examples/first-run"GENERATED_FIXTURE_DIRS = ("build", "target", ".ultimate-tool")SNAPSHOT_PINS = {"ultimate-tool-smoke.prg": ACCEPTED_SMOKE}
FAKE_DOCKER = '''#!/usr/bin/env python3"""Stand-in for the pinned rust-mos image.
Records every argv, tracks a container state marker so `ps`/`stop`/`rm` behavelike the real runtime, and makes `llvm-readelf` fail on a path that does notexist exactly as the real tool does."""import json, os, pathlib, sys, time
argv = sys.argv[1:]log = os.environ.get("FAKE_DOCKER_LOG")if log: with open(log, "a", encoding="utf-8") as handle: handle.write(json.dumps(argv) + "\\n")
mode = os.environ.get("FAKE_DOCKER_MODE", "ok")state = pathlib.Path(os.environ.get("FAKE_DOCKER_STATE_DIR", "/tmp"))
def marker(name): return state / ("container-" + name)
def env_code(key): return int(os.environ.get(key, "0"))
if "inspect" in argv: print(os.environ.get("FAKE_DOCKER_IMAGE_ID", "%s")) raise SystemExit(0)
if argv and argv[0] in ("stop", "rm"): name = argv[-1] code = env_code("FAKE_DOCKER_RM_EXIT" if argv[0] == "rm" else "FAKE_DOCKER_STOP_EXIT") if argv[0] == "rm" and code == 0 and marker(name).exists(): marker(name).unlink() raise SystemExit(code)
if argv and argv[0] == "ps": code = env_code("FAKE_DOCKER_PS_EXIT") if code: sys.stderr.write("fake docker: ps failed\\n") raise SystemExit(code) wanted = "" for item in argv: if item.startswith("name=^"): wanted = item[len("name=^"):].rstrip("$") if wanted and marker(wanted).exists(): print("deadbeef") raise SystemExit(0)
root = workdir = Nonetarget_dir = "target/line-tables"for index, item in enumerate(argv): if item == "--volume" and argv[index + 1].endswith(":/work"): root = pathlib.Path(argv[index + 1][:-len(":/work")]) if item == "--workdir": workdir = argv[index + 1] if item == "--target-dir": target_dir = argv[index + 1]if root is None or workdir is None: sys.stderr.write("fake docker: no composed root in argv\\n") raise SystemExit(2)
name = ""for index, item in enumerate(argv): if item == "--name": name = argv[index + 1]if name: marker(name).touch()
if "llvm-readelf" in argv: container_path = argv[-1] host_path = root / pathlib.Path(container_path).relative_to("/work") if not host_path.is_file(): sys.stderr.write( "llvm-readelf: error: '" + container_path + "': No such file or directory\\n" ) raise SystemExit(1) print(".debug_info") print(".debug_line") raise SystemExit(0)
if mode == "fail": sys.stderr.write("fake cargo failure\\n") raise SystemExit(1)
if mode == "block": # Outlives the harness deadline so the harness must time out and tear down. time.sleep(120) raise SystemExit(0)
fixture_dir = root / pathlib.Path(workdir).relative_to("/work")package = Nonefor line in (fixture_dir / "Cargo.toml").read_text().splitlines(): if line.startswith("name = "): package = line.split('"')[1] breakif package is None: sys.stderr.write("fake docker: no package name\\n") raise SystemExit(2)
payload = pathlib.Path(os.environ["FAKE_DOCKER_PRG"]).read_bytes()if mode == "tamper": payload = bytes([payload[0] ^ 0xFF]) + payload[1:]
release = fixture_dir / target_dir / "mos-c64-none/release"(release / "deps").mkdir(parents=True, exist_ok=True)(release / package).write_bytes(payload)if mode != "noelf": snake = package.replace("-", "_") (release / "deps" / f"{snake}-0123456789abcdef.elf").write_bytes(b"\\x7fELF fake")(fixture_dir / "build").mkdir(parents=True, exist_ok=True)(fixture_dir / "build" / f"{package}.map").write_text("fake map\\n")raise SystemExit(0)'''
def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest()
def approved_snapshot(name: str, expected: str | None = None) -> Path: """A tracked accepted build output, checked against the demo's own record.""" manifest = json.loads((APPROVED_DEMO / "snapshot.json").read_text()) recorded = { Path(entry["path"]).name: entry for entry in manifest["file_entries"] if entry["output"] } entry = recorded.get(name) if entry is None: raise AssertionError(f"the approved snapshot does not record the output {name}") path = APPROVED_DEMO / entry["path"] if not path.is_file(): raise AssertionError(f"the approved snapshot {path} is missing") observed, size = sha256(path), path.stat().st_size if observed != entry["sha256"] or size != entry["size"]: raise AssertionError( f"the approved snapshot {path} drifted: {observed} ({size} bytes), " f"recorded {entry['sha256']} ({entry['size']} bytes)" ) if expected is not None and observed != expected: raise AssertionError( f"the approved snapshot {path} is {observed}, not the accepted {expected}" ) return path
def seed_published_outputs(repo: Path, fixture: str = "rust-mos-smoke") -> Path: """Give a temp fixture the already-published outputs a developer would have.
The publish tests assert that a failed, refused or interrupted build leaves the published outputs byte-identical, so the temp fixture needs real ones to compare against. Every seeded file is hash-checked against the approved snapshot's record, and the PRG is additionally pinned to the accepted artifact, so a drifted snapshot cannot stand in for accepted bytes. """ build = repo / "fixtures" / fixture / "build" build.mkdir(parents=True, exist_ok=True) manifest = json.loads((APPROVED_DEMO / "snapshot.json").read_text()) for entry in manifest["file_entries"]: if not entry["output"]: continue name = Path(entry["path"]).name shutil.copy2(approved_snapshot(name, SNAPSHOT_PINS.get(name)), build / name) return build
def synthetic_timeline_payload(directory: Path, size: int = 2048) -> Path: """Deterministic stand-in bytes for the timeline fixture's PRG.
Mock only. These bytes are not, and cannot be made, the accepted timeline artifact: that artifact is produced by the pinned image and is not tracked, so no clean checkout has it. They exist so the harness logic around a second fixture stays testable without claiming an accepted build. """ marker = b"ultimate-tool synthetic timeline payload\n" body = bytes(range(256)) filler = (body * (size // len(body) + 1))[: max(size - len(marker), 0)] path = directory / "synthetic-timeline.prg" path.write_bytes(marker + filler) return path
def pin_temp_timeline_contract(repo: Path, payload: Path) -> None: """Pin the synthetic timeline bytes in the temporary repository's contract.
Only this throwaway copy is rewritten. The tracked contract under ``fixtures/accepted-sdk`` keeps pinning the real accepted timeline artifact, so a test can show the timeline fixture is validated against its own entry rather than the smoke pin, and that no acceptance claim moved. """ path = repo / "fixtures/accepted-sdk/accepted-sdk.json" document = json.loads(path.read_text()) fixture = document["fixtures"]["rust-mos-timeline"] fixture["artifact_sha256"] = sha256(payload) fixture["artifact_size"] = payload.stat().st_size path.write_text(json.dumps(document, indent=2, sort_keys=True) + "\n")
def make_temp_repo(tmp: Path, *, host_version: str = "0.2.0-preview.1") -> Path: """A throwaway repository with a moved-on host workspace version.
Only tracked fixture inputs are copied. ``build``, ``target`` and ``.ultimate-tool`` are generated and git-ignored, so a clean checkout does not have them and no test may lean on the ones this working tree holds; the already-published outputs a developer would have are then seeded explicitly from the approved snapshot. """ repo = tmp / "repo" (repo / "tools").mkdir(parents=True) shutil.copy2(TOOLS / "build_accepted_fixture.py", repo / "tools/build_accepted_fixture.py") shutil.copytree(ROOT / "fixtures/accepted-sdk", repo / "fixtures/accepted-sdk") # The frozen SDK copy is also present at its original path so the temp repo # mirrors the real layout without compiling anything. shutil.copytree(ROOT / "crates/ultimate-tool-c64", repo / "crates/ultimate-tool-c64") for fixture in ("rust-mos-smoke", "rust-mos-timeline"): shutil.copytree( ROOT / "fixtures" / fixture, repo / "fixtures" / fixture, ignore=shutil.ignore_patterns(*GENERATED_FIXTURE_DIRS), ) seed_published_outputs(repo) (repo / "Cargo.toml").write_text( "[workspace]\n" 'members = ["crates/ultimate-tool-c64"]\n' 'exclude = ["fixtures/rust-mos-smoke", "fixtures/rust-mos-timeline"]\n' 'resolver = "2"\n\n' "[workspace.package]\n" 'edition = "2024"\n' f'version = "{host_version}"\n' 'license = "GPL-3.0-or-later"\n' ) return repo
class FakeDockerHarness: """Runs the harness against a fake docker and exposes its argv log."""
def __init__(self, repo: Path, tmp: Path) -> None: self.repo = repo self.tmp = tmp self.work = repo / "target/accepted-fixture-build" self.state = tmp / "docker-state" self.state.mkdir(parents=True, exist_ok=True) self.log = tmp / "docker-argv.jsonl" self.fake = tmp / "fake-docker" self.fake.write_text(FAKE_DOCKER % IMAGE_ID) self.fake.chmod(0o755) self.prg = approved_snapshot("ultimate-tool-smoke.prg", ACCEPTED_SMOKE)
def run(self, *extra: str, fixture: str = "rust-mos-smoke", mode: str = "ok", image_id: str | None = None, payload: Path | None = None, env_extra: dict | None = None): env = dict(os.environ) env["FAKE_DOCKER_LOG"] = str(self.log) env["FAKE_DOCKER_MODE"] = mode env["FAKE_DOCKER_PRG"] = str(payload or self.prg) env["FAKE_DOCKER_IMAGE_ID"] = image_id or IMAGE_ID env["FAKE_DOCKER_STATE_DIR"] = str(self.state) if env_extra: env.update(env_extra) argv = [ sys.executable, str(self.repo / "tools/build_accepted_fixture.py"), "--fixture", fixture, "--docker", str(self.fake), "--work-parent", str(self.work), "--cache-dir", str(self.repo / "target/accepted-fixture-cargo-home"), "--json", *extra, ] return subprocess.run(argv, capture_output=True, text=True, env=env)
def invocations(self) -> list[list[str]]: if not self.log.exists(): return [] return [json.loads(line) for line in self.log.read_text().splitlines() if line]
def build_kwargs(self, **overrides) -> dict: """In-process arguments for baf.build_fixture against the fake docker.""" kwargs = dict( fixture_name="rust-mos-smoke", mode="build", contract_path=self.repo / "fixtures/accepted-sdk/accepted-sdk.json", work_parent=self.work, cache_dir=self.repo / "target/accepted-fixture-cargo-home", docker=str(self.fake), repository_root=self.repo, timeout=30.0, ) kwargs.update(overrides) return kwargs
def env_for_process(self, **extra) -> dict: env = dict(os.environ) env["FAKE_DOCKER_LOG"] = str(self.log) env["FAKE_DOCKER_MODE"] = extra.pop("mode", "ok") env["FAKE_DOCKER_PRG"] = str(extra.pop("payload", self.prg)) env["FAKE_DOCKER_IMAGE_ID"] = extra.pop("image_id", IMAGE_ID) env["FAKE_DOCKER_STATE_DIR"] = str(self.state) env.update({key: str(value) for key, value in extra.items()}) return env
class ContractTests(unittest.TestCase): def test_contract_pins_the_accepted_frozen_input(self) -> None: contract = baf.load_contract(ROOT / "fixtures/accepted-sdk/accepted-sdk.json") self.assertEqual(contract["record_kind"], baf.INPUT_RECORD_KIND) self.assertEqual(contract["sdk"]["version"], "0.1.0") self.assertEqual(contract["sdk"]["workspace_path"], "crates/ultimate-tool-c64") self.assertEqual(contract["workspace"]["version"], "0.1.0") self.assertEqual(contract["sdk"]["files"]["src/lib.rs"], SDK_SOURCE) self.assertEqual( contract["fixtures"]["rust-mos-smoke"]["artifact_sha256"], ACCEPTED_SMOKE ) self.assertEqual( contract["fixtures"]["rust-mos-timeline"]["artifact_sha256"], ACCEPTED_TIMELINE )
def test_frozen_sdk_copy_matches_the_repository_crate(self) -> None: for relative in ("Cargo.toml", "src/lib.rs"): self.assertEqual( sha256(ROOT / "fixtures/accepted-sdk" / relative), sha256(ROOT / "crates/ultimate-tool-c64" / relative), f"the frozen SDK copy of {relative} drifted from the repository crate", )
def test_fixture_source_pins_match_the_repository(self) -> None: for name in ("rust-mos-smoke", "rust-mos-timeline"): checks = baf.verify_fixture_sources( name, baf.load_contract(ROOT / "fixtures/accepted-sdk/accepted-sdk.json")["fixtures"][name] ) self.assertTrue(all(item["ok"] for item in checks), name)
def test_rejects_wrong_schema(self) -> None: with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "contract.json" document = json.loads((ROOT / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) document["schema"] = 2 path.write_text(json.dumps(document)) with self.assertRaises(baf.Refused): baf.load_contract(path)
def test_rejects_wrong_record_kind(self) -> None: with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "contract.json" document = json.loads((ROOT / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) document["record_kind"] = "something_else" path.write_text(json.dumps(document)) with self.assertRaises(baf.Refused): baf.load_contract(path)
def test_rejects_a_moved_frozen_version(self) -> None: with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "contract.json" document = json.loads((ROOT / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) document["sdk"]["version"] = "0.2.0-preview.1" path.write_text(json.dumps(document)) with self.assertRaises(baf.Refused) as caught: baf.load_contract(path) self.assertIn("0.1.0", str(caught.exception))
def test_rejects_a_workspace_that_is_not_the_frozen_member(self) -> None: with tempfile.TemporaryDirectory() as raw: path = Path(raw) / "contract.json" document = json.loads((ROOT / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) document["workspace"]["members"] = [ "crates/ultimate-tool-c64", "crates/ultimate-tool-core", ] path.write_text(json.dumps(document)) with self.assertRaises(baf.Refused): baf.load_contract(path)
def test_reads_the_host_workspace_version(self) -> None: with tempfile.TemporaryDirectory() as raw: manifest = Path(raw) / "Cargo.toml" manifest.write_text( "[workspace.package]\n" 'edition = "2024"\n' 'version = "0.2.0-preview.1"\n' ) self.assertEqual(baf.host_workspace_version(manifest), "0.2.0-preview.1") self.assertIsNone(baf.host_workspace_version(Path(raw) / "missing.toml"))
def test_rendered_workspace_pins_the_frozen_version(self) -> None: contract = baf.load_contract(ROOT / "fixtures/accepted-sdk/accepted-sdk.json") text = baf.render_workspace_manifest(contract) self.assertIn('version = "0.1.0"', text) self.assertIn('members = ["crates/ultimate-tool-c64"]', text) self.assertIn("unsafe_op_in_unsafe_fn", text)
class RefusalTests(unittest.TestCase): def setUp(self) -> None: self._tmpdir = tempfile.TemporaryDirectory() self.tmp = Path(self._tmpdir.name)
def tearDown(self) -> None: self._tmpdir.cleanup()
def test_tampered_frozen_sdk_is_refused(self) -> None: repo = make_temp_repo(self.tmp) target = repo / "fixtures/accepted-sdk/src/lib.rs" target.write_bytes(target.read_bytes() + b"// tampered\n") harness = FakeDockerHarness(repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("frozen SDK input", result.stdout) self.assertEqual(harness.invocations(), []) self.assertTrue((repo / "fixtures/rust-mos-smoke/build").exists())
def test_fixture_source_drift_is_refused(self) -> None: repo = make_temp_repo(self.tmp) target = repo / "fixtures/rust-mos-smoke/src/main.rs" target.write_bytes(target.read_bytes() + b"\n// drift\n") harness = FakeDockerHarness(repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("source drift", result.stdout) self.assertEqual(harness.invocations(), [])
def test_update_records_pins_without_accepting_them(self) -> None: repo = make_temp_repo(self.tmp) target = repo / "fixtures/rust-mos-smoke/src/main.rs" target.write_bytes(target.read_bytes() + b"\n// drift\n") harness = FakeDockerHarness(repo, self.tmp) result = harness.run("--update") self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) self.assertTrue(record["requires_reacceptance"]) self.assertIn("src/main.rs", record["changed"]) contract = json.loads((repo / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) self.assertEqual( contract["fixtures"]["rust-mos-smoke"]["artifact_sha256"], ACCEPTED_SMOKE, "the accepted artifact hash must not move when pins are refreshed", ) self.assertEqual( contract["fixtures"]["rust-mos-smoke"]["sources"]["src/main.rs"], sha256(target), ) # After the update the rebuilt (unaccepted) fixture compiles again. self.assertEqual(harness.run().returncode, 0)
def test_symlinked_fixture_input_is_refused(self) -> None: repo = make_temp_repo(self.tmp) link = repo / "fixtures/rust-mos-smoke/link.ld" real = repo / "fixtures/rust-mos-smoke/link.ld.real" link.rename(real) os.symlink(real.name, link) harness = FakeDockerHarness(repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("symlink", result.stdout) self.assertEqual(harness.invocations(), [])
def test_symlinked_sdk_input_is_refused(self) -> None: repo = make_temp_repo(self.tmp) source = repo / "fixtures/accepted-sdk/src/lib.rs" real = repo / "fixtures/accepted-sdk/src/lib.rs.real" source.rename(real) os.symlink(real.name, source) harness = FakeDockerHarness(repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("symlink", result.stdout)
def test_wrong_image_identity_is_refused(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) result = harness.run(image_id="sha256:" + "0" * 64) self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("pinned image", result.stdout)
def test_symlinked_cache_directory_is_refused(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) real = repo / "target/real-cache" real.mkdir(parents=True, exist_ok=True) link = repo / "target/linked-cache" os.symlink(real, link) result = harness.run("--cache-dir", str(link)) self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("symlink", result.stdout) self.assertEqual(harness.invocations(), [])
def test_cache_that_looks_like_a_host_cargo_home_is_refused(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) hostile = repo / "target/host-cargo" hostile.mkdir(parents=True, exist_ok=True) (hostile / "config.toml").write_text('[source.crates-io]\nreplace-with = "x"\n') result = harness.run("--cache-dir", str(hostile)) self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("config.toml", result.stdout) self.assertEqual(harness.invocations(), [], "nothing may be mounted")
def test_cache_outside_the_repository_is_refused(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) outside = self.tmp / "outside-cache" outside.mkdir(parents=True, exist_ok=True) result = harness.run("--cache-dir", str(outside)) self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("outside the repository root", result.stdout) self.assertEqual(list(outside.iterdir()), [])
class CompositionTests(unittest.TestCase): def setUp(self) -> None: self._tmpdir = tempfile.TemporaryDirectory() self.tmp = Path(self._tmpdir.name)
def tearDown(self) -> None: self._tmpdir.cleanup()
def test_dry_run_plans_the_exact_container_command(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) result = harness.run("--dry-run") self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) self.assertEqual(record["status"], "planned") argv = record["commands"][0] contract = json.loads((repo / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) self.assertIn(contract["toolchain"]["image"], argv) self.assertIn("/work/fixtures/rust-mos-smoke", argv) index = argv.index("--workdir") self.assertEqual(argv[index + 1], "/work/fixtures/rust-mos-smoke") mounts = [argv[i + 1] for i, item in enumerate(argv) if item == "--volume"] targets = sorted(item.split(":")[1] for item in mounts) self.assertEqual(targets, ["/tmp/ut-cargo", "/work"]) self.assertTrue( any(item.startswith("CARGO_HOME=") for item in argv), "the container needs an isolated CARGO_HOME", ) self.assertIn("--name", argv) self.assertTrue(argv[argv.index("--name") + 1].startswith("ultimate-tool-accepted-")) self.assertEqual(record["containers"], [], "a dry run starts no container") cargo = argv[argv.index(contract["toolchain"]["image"]) + 1:] self.assertEqual( cargo, ["cargo", "build", "--release", "--target", "mos-c64-none", "--target-dir", "target/line-tables"], ) self.assertFalse(harness.invocations(), "a dry run must not call docker") self.assertFalse(harness.work.exists() and any(harness.work.iterdir()))
def test_plan_never_mounts_host_home_or_credentials(self) -> None: repo = make_temp_repo(self.tmp) harness = FakeDockerHarness(repo, self.tmp) record = json.loads(harness.run("--dry-run").stdout) argv = record["commands"][0] mounts = [argv[i + 1] for i, item in enumerate(argv) if item == "--volume"] self.assertEqual(len(mounts), 2, mounts) home = str(Path.home()) for mount in mounts: self.assertNotIn(home, mount) self.assertNotIn(".docker", mount) self.assertNotIn(".cargo/config", mount) self.assertIn("/work", mounts[1].split(":")[1]) self.assertEqual(mounts[0].split(":")[1], "/tmp/ut-cargo")
def test_composed_root_places_the_sdk_at_its_original_path(self) -> None: repo = make_temp_repo(self.tmp) contract = baf.load_contract(repo / "fixtures/accepted-sdk/accepted-sdk.json") root = self.tmp / "manual-root" root.mkdir() baf.compose_root(root, contract, "rust-mos-smoke", repo) sdk = root / "crates/ultimate-tool-c64" self.assertEqual(sha256(sdk / "src/lib.rs"), SDK_SOURCE) self.assertIn('version.workspace = true', (sdk / "Cargo.toml").read_text()) self.assertIn('version = "0.1.0"', (root / "Cargo.toml").read_text()) self.assertTrue((root / "fixtures/rust-mos-smoke/src/main.rs").is_file()) self.assertTrue((root / "fixtures/rust-mos-smoke/link.ld").is_file()) self.assertTrue((root / "fixtures/rust-mos-smoke/.cargo/config.toml").is_file()) self.assertTrue((root / "fixtures/rust-mos-smoke/build.rs").is_file()) # Nothing from the fixture's build outputs is copied into the root. self.assertFalse(any((root / "fixtures/rust-mos-smoke/build").iterdir()))
def test_composed_root_refuses_a_symlinked_directory(self) -> None: repo = make_temp_repo(self.tmp) contract = baf.load_contract(repo / "fixtures/accepted-sdk/accepted-sdk.json") root = self.tmp / "manual-root2" root.mkdir() source = repo / "fixtures/rust-mos-smoke/.cargo" real = repo / "fixtures/rust-mos-smoke/cargo-config-real" source.rename(real) os.symlink(real.name, source) with self.assertRaises(baf.Refused): baf.compose_root(root, contract, "rust-mos-smoke", repo)
class PublishTests(unittest.TestCase): def setUp(self) -> None: self._tmpdir = tempfile.TemporaryDirectory() self.tmp = Path(self._tmpdir.name) self.repo = make_temp_repo(self.tmp) self.build = self.repo / "fixtures/rust-mos-smoke/build" self.published = self.build / "ultimate-tool-smoke.prg" self.before = sha256(self.published)
def tearDown(self) -> None: self._tmpdir.cleanup()
def _snapshot(self) -> dict: return { item.name: (sha256(item), os.lstat(item).st_size) for item in sorted(self.build.iterdir()) }
def test_successful_build_publishes_only_verified_bytes(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) prg = record["artifacts"]["prg"] self.assertTrue(prg["matches_expected"]) self.assertEqual(prg["sha256"], ACCEPTED_SMOKE) self.assertEqual(prg["size"], 4932) for entry in record["published"]: self.assertTrue( entry["path"].startswith(str(self.build) + os.sep), entry["path"], ) published = {item["key"] for item in record["published"]} self.assertEqual(published, {"prg", "elf", "map"}) self.assertEqual(sha256(self.published), ACCEPTED_SMOKE) leftovers = [item.name for item in self.build.iterdir() if item.name.startswith(".")] self.assertEqual(leftovers, [], "no partially written files may be left behind")
def test_failed_build_leaves_published_outputs_untouched(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run(mode="fail") self.assertEqual(result.returncode, 1, result.stdout + result.stderr) self.assertIn("cargo build failed", result.stdout) self.assertEqual(self._snapshot(), before) self.assertEqual(sha256(self.published), self.before)
def test_wrong_artifact_is_refused_and_never_published(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run(mode="tamper") self.assertEqual(result.returncode, 1, result.stdout + result.stderr) self.assertIn("not the accepted artifact", result.stdout) self.assertEqual(self._snapshot(), before) self.assertEqual(sha256(self.published), self.before)
def test_temporary_roots_are_removed_and_kept_on_request(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) harness.run() self.assertEqual([item.name for item in harness.work.iterdir()], []) harness.run("--keep-work") self.assertEqual(len([item for item in harness.work.iterdir()]), 1) for item in harness.work.iterdir(): shutil.rmtree(item)
def test_work_parent_outside_the_repository_is_refused(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) outside = self.tmp / "elsewhere" outside.mkdir(parents=True, exist_ok=True) result = harness.run("--work-parent", str(outside)) self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("outside the repository root", result.stdout) self.assertEqual(list(outside.iterdir()), [])
def test_timeline_fixture_uses_its_own_recorded_hash(self) -> None: """Mock, not an acceptance claim: synthetic bytes and a test-local pin.
The timeline fixture must be validated against its own contract entry, not the smoke artifact's hash. The tracked contract still pins the real accepted timeline artifact; only this throwaway copy is rewritten. """ payload = synthetic_timeline_payload(self.tmp) pin_temp_timeline_contract(self.repo, payload) expected = sha256(payload) self.assertNotEqual(expected, ACCEPTED_SMOKE, "the routing must be distinguishable") harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run(fixture="rust-mos-timeline", payload=payload) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) self.assertEqual(record["fixture"], "rust-mos-timeline") self.assertEqual(record["artifacts"]["prg"]["sha256"], expected) self.assertEqual(record["artifacts"]["prg"]["expected_sha256"], expected) self.assertTrue(record["artifacts"]["prg"]["matches_expected"]) self.assertEqual( sha256(self.repo / "fixtures/rust-mos-timeline/build/ultimate-tool-timeline.prg"), expected, ) # No acceptance moved: the tracked contract and the smoke pin are intact. production = json.loads((ROOT / "fixtures/accepted-sdk/accepted-sdk.json").read_text()) self.assertEqual( production["fixtures"]["rust-mos-timeline"]["artifact_sha256"], ACCEPTED_TIMELINE ) self.assertEqual( production["fixtures"]["rust-mos-smoke"]["artifact_sha256"], ACCEPTED_SMOKE )
def test_verify_reproducible_builds_twice_in_separate_target_dirs(self) -> None: """Mock: the two-run check is about harness plumbing, not acceptance.""" payload = synthetic_timeline_payload(self.tmp) pin_temp_timeline_contract(self.repo, payload) harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run("--mode", "verify-reproducible", fixture="rust-mos-timeline", payload=payload) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) target_dirs = [] for argv in harness.invocations(): if "--target-dir" in argv: target_dirs.append(argv[argv.index("--target-dir") + 1]) self.assertEqual(target_dirs, ["target/line-tables", "target/reproduce"]) record = json.loads(result.stdout) self.assertIn({"check": "reproducible_rebuild", "ok": True}, record["checks"])
def test_verify_debug_compares_stripped_and_line_table_builds(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run("--mode", "verify-debug") self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) checks = {item["check"] for item in record["checks"]} self.assertIn("debug_builds_match", checks) self.assertIn("elf_sections", checks) target_dirs = [] for argv in harness.invocations(): if "--target-dir" in argv: target_dirs.append(argv[argv.index("--target-dir") + 1]) self.assertIn("target/verify-release", target_dirs) self.assertIn("target/verify-lines", target_dirs)
def test_verify_debug_for_the_timeline_fixture_expects_no_crash_site(self) -> None: """Mock: the timeline fixture declares no debug-build comparison.""" payload = synthetic_timeline_payload(self.tmp) pin_temp_timeline_contract(self.repo, payload) harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run("--mode", "verify-debug", fixture="rust-mos-timeline", payload=payload) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) checks = {item["check"] for item in record["checks"]} self.assertIn("elf_sections", checks) self.assertNotIn("debug_builds_match", checks) target_dirs = [] for argv in harness.invocations(): if "--target-dir" in argv: target_dirs.append(argv[argv.index("--target-dir") + 1]) self.assertEqual(target_dirs, ["target/line-tables"])
def test_missing_linked_elf_fails_before_publishing(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run(mode="noelf") self.assertEqual(result.returncode, 1, result.stdout + result.stderr) self.assertIn("freshly linked fixture ELF", result.stdout) self.assertEqual(self._snapshot(), before)
def test_symlinked_build_directory_is_refused(self) -> None: outside = self.tmp / "elsewhere" outside.mkdir() self.build.rename(self.tmp / "build-real") os.symlink(outside, self.build) harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("symlink", result.stdout) self.assertEqual(list(outside.iterdir()), [], "nothing may be written outside")
def test_symlinked_destination_artifact_is_refused(self) -> None: outside = self.tmp / "captured.prg" outside.write_bytes(b"untouched") for item in self.build.iterdir(): item.unlink() os.symlink(outside, self.published) harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run() self.assertEqual(result.returncode, 2, result.stdout + result.stderr) self.assertIn("symlink", result.stdout) self.assertEqual(outside.read_bytes(), b"untouched")
def test_container_deadline_tears_the_container_down(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run("--timeout", "2", mode="block") self.assertEqual(result.returncode, 1, result.stdout + result.stderr) self.assertIn("deadline", result.stdout) self.assertIn("absence confirmed", result.stdout) invocations = harness.invocations() record = json.loads(result.stdout) recorded = record["containers"] self.assertEqual(record["status"], "failed") for argv in invocations: if argv[:1] == ["stop"] or argv[:1] == ["rm"]: self.assertIn(argv[-1], recorded) self.assertEqual( [item.name for item in harness.state.glob("container-*")], [], "the removed container must leave no state behind", ) self.assertEqual(self._snapshot(), before)
def test_remove_failure_keeps_the_root_and_reports_unresolved(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) env = harness.env_for_process(mode="block", FAKE_DOCKER_RM_EXIT="1") with mock.patch.dict(os.environ, env, clear=False): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs(timeout=2.0)) error = caught.exception retained = Path(error.retained_root) self.assertTrue(retained.is_dir(), "the composed root must be retained") self.assertIn("could not be confirmed gone", str(error)) self.assertNotIn("absence confirmed", str(error)) self.assertTrue(list(harness.state.glob("container-*")), "the surviving container state must still be visible") self.assertEqual(self._snapshot(), before, "nothing may be published") record = getattr(error, "record") self.assertEqual(record["status"], "unresolved") self.assertEqual(record["retained_root"], str(retained)) shutil.rmtree(retained, ignore_errors=True)
def _interrupt_first_run(self, harness: "FakeDockerHarness"): """Interrupt the first container run after the runtime has started it.""" original = baf.ContainerRunner._spawn state = {"done": False}
def spawn(self, argv, timeout): arguments = list(argv) if arguments[1:2] == ["run"] and not state["done"]: state["done"] = True name = arguments[arguments.index("--name") + 1] (harness.state / f"container-{name}").touch() raise KeyboardInterrupt() return original(self, argv, timeout)
return spawn
def test_interrupt_with_unconfirmed_teardown_retains_the_root(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) env = harness.env_for_process(FAKE_DOCKER_RM_EXIT="1") with mock.patch.dict(os.environ, env, clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", self._interrupt_first_run(harness)): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs()) self.assertIn("KeyboardInterrupt", str(caught.exception)) self.assertTrue(Path(caught.exception.retained_root).is_dir()) self.assertEqual(self._snapshot(), before) shutil.rmtree(caught.exception.retained_root, ignore_errors=True)
def test_interrupt_with_confirmed_teardown_propagates(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) env = harness.env_for_process() with mock.patch.dict(os.environ, env, clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", self._interrupt_first_run(harness)): with self.assertRaises(KeyboardInterrupt): baf.build_fixture(**harness.build_kwargs()) self.assertEqual(self._snapshot(), before, "nothing may be published") leftovers = [item for item in harness.work.glob("*")] if harness.work.exists() else [] self.assertEqual(leftovers, [], "a confirmed teardown must not retain the root")
def test_executed_containers_are_recorded_exactly_once(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) result = harness.run("--mode", "verify-debug") self.assertEqual(result.returncode, 0, result.stdout + result.stderr) record = json.loads(result.stdout) names, commands = record["containers"], record["commands"] self.assertEqual(len(names), len(commands)) self.assertEqual(len(names), len(set(names)), "no duplicated container records") for name, argv in zip(names, commands): self.assertEqual(argv[argv.index("--name") + 1], name) executed = [argv[argv.index("--name") + 1] for argv in harness.invocations() if argv[:1] == ["run"]] self.assertEqual(sorted(executed), sorted(names), "the recorded names must be exactly the executed containers")
def test_readelf_missing_path_is_detected(self) -> None: harness = FakeDockerHarness(self.repo, self.tmp) env = harness.env_for_process() with mock.patch.dict(os.environ, env, clear=False): contract = baf.load_contract(self.repo / "fixtures/accepted-sdk/accepted-sdk.json") root = self.tmp / "probe-root" root.mkdir() runner = baf.ContainerRunner( contract, str(harness.fake), root, contract["fixtures"]["rust-mos-smoke"], self.repo / "target/accepted-fixture-cargo-home", ) with self.assertRaises(baf.BuildFailed) as caught: baf.elf_section_check(runner, root / "missing" / "absent.elf", [".debug_line"]) self.assertIn("cannot read sections", str(caught.exception))
def _interrupt_stage(self, harness: "FakeDockerHarness", stage: str): """Interrupt the first container run, then the named teardown stage.""" original = baf.ContainerRunner._spawn state = {"run": False, "stage": False}
def spawn(self, argv, timeout): arguments = list(argv) if arguments[1:2] == ["run"] and not state["run"]: state["run"] = True name = arguments[arguments.index("--name") + 1] (harness.state / f"container-{name}").touch() raise KeyboardInterrupt() if arguments[1:2] == [stage] and not state["stage"]: state["stage"] = True raise KeyboardInterrupt() return original(self, argv, timeout)
return spawn
def test_interrupt_during_stop_still_confirms_removal(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) with mock.patch.dict(os.environ, harness.env_for_process(), clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", self._interrupt_stage(harness, "stop")): with self.assertRaises(KeyboardInterrupt): baf.build_fixture(**harness.build_kwargs()) self.assertEqual(self._snapshot(), before, "nothing may be published") leftovers = [item for item in harness.work.glob("*")] if harness.work.exists() else [] self.assertEqual(leftovers, [], "remove still ran, so the root is not retained")
def test_interrupt_during_remove_retains_the_root(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) with mock.patch.dict(os.environ, harness.env_for_process(), clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", self._interrupt_stage(harness, "rm")): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs()) self.assertTrue(Path(caught.exception.retained_root).is_dir()) self.assertIn("could not be confirmed gone", str(caught.exception)) self.assertEqual(self._snapshot(), before) shutil.rmtree(caught.exception.retained_root, ignore_errors=True)
def test_interrupt_during_presence_check_retains_the_root(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) with mock.patch.dict(os.environ, harness.env_for_process(), clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", self._interrupt_stage(harness, "ps")): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs()) self.assertTrue(Path(caught.exception.retained_root).is_dir()) self.assertEqual(self._snapshot(), before) shutil.rmtree(caught.exception.retained_root, ignore_errors=True)
def test_failed_presence_query_retains_the_root(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) env = harness.env_for_process(mode="block", FAKE_DOCKER_PS_EXIT="1") with mock.patch.dict(os.environ, env, clear=False): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs(timeout=2.0)) self.assertTrue(Path(caught.exception.retained_root).is_dir()) self.assertIn("could not be confirmed gone", str(caught.exception)) self.assertEqual(self._snapshot(), before) shutil.rmtree(caught.exception.retained_root, ignore_errors=True)
def test_timed_out_presence_query_retains_the_root(self) -> None: before = self._snapshot() harness = FakeDockerHarness(self.repo, self.tmp) original = baf.ContainerRunner._spawn
def spawn(self, argv, timeout): arguments = list(argv) if arguments[1:2] == ["ps"]: raise subprocess.TimeoutExpired(arguments, timeout) return original(self, argv, timeout)
with mock.patch.dict(os.environ, harness.env_for_process(mode="block"), clear=False): with mock.patch.object(baf.ContainerRunner, "_spawn", spawn): with self.assertRaises(baf.Unresolved) as caught: baf.build_fixture(**harness.build_kwargs(timeout=2.0)) self.assertTrue(Path(caught.exception.retained_root).is_dir()) self.assertEqual(self._snapshot(), before) shutil.rmtree(caught.exception.retained_root, ignore_errors=True)
@unittest.skipUnless( os.environ.get(REAL_BUILD_ENV) == "1", f"set {REAL_BUILD_ENV}=1 to build both fixtures in the pinned image",)class RealBuildTests(unittest.TestCase): """The container-backed acceptance check, opt-in because it needs Docker."""
def test_both_fixtures_reproduce_the_accepted_hashes(self) -> None: expected = {"rust-mos-smoke": ACCEPTED_SMOKE, "rust-mos-timeline": ACCEPTED_TIMELINE} for fixture, digest in expected.items(): with self.subTest(fixture=fixture): record = baf.build_fixture(fixture_name=fixture, mode="build") self.assertEqual(record["artifacts"]["prg"]["sha256"], digest) self.assertEqual(record["frozen_workspace_version"], "0.1.0") self.assertTrue(record["artifacts"]["prg"]["matches_expected"]) published = {item["key"]: item["sha256"] for item in record["published"]} self.assertEqual(published["prg"], digest) self.assertTrue(record["containers"])
def test_fixture_verification_modes_hold_in_the_pinned_image(self) -> None: cases = ( # fixture, mode, required checks, expected ELF sections (None: no check) ("rust-mos-smoke", "verify-debug", {"debug_builds_match", "elf_sections"}, [".debug_info", ".debug_line"]), ("rust-mos-timeline", "verify-debug", {"elf_sections"}, [".debug_line"]), ("rust-mos-timeline", "verify-reproducible", {"reproducible_rebuild"}, None), ) for fixture, mode, required, sections in cases: with self.subTest(fixture=fixture, mode=mode): record = baf.build_fixture(fixture_name=fixture, mode=mode) checks = {item["check"]: item for item in record["checks"]} self.assertTrue(required <= set(checks), sorted(set(checks))) if sections is None: self.assertNotIn("elf_sections", checks) self.assertNotIn("debug_builds_match", checks) else: self.assertEqual(checks["elf_sections"]["present"], sections) if mode != "verify-debug": self.assertNotIn("debug_builds_match", checks) self.assertTrue(record["artifacts"]["prg"]["matches_expected"]) self.assertEqual( len(record["containers"]), len(set(record["containers"])), "each executed container must be recorded exactly once", )
if __name__ == "__main__": unittest.main()