#!/usr/bin/env python3 """Tests for the deterministic first-run demo snapshot tooling. These tests are pure, standard-library ``unittest`` and run entirely inside temporary directories. They build synthetic ELF/PRG artifacts in memory, drive the real command-line entry points of ``prepare_demo_snapshot.py`` and ``materialize_demo.py``, and assert the documented schema, verification, and materialization behaviour. Because every artifact here is synthesised, a passing run proves the schema, the PRG/ELF arithmetic checks, and the packaging/materialization orchestration. It does not prove that a compiler ran, that the pinned sources produce these bytes, or that any C64 instruction executed. The one exception is :meth:`RealArtifactTests.test_pinned_artifact_matches_recorded_hash`, which is skipped unless the ignored Rust-MOS build outputs are already present. """ from __future__ import annotations import contextlib import errno import io import json import os import struct import subprocess import sys import tempfile import unittest import unittest.mock from pathlib import Path ROOT = Path(__file__).resolve().parents[2] TOOLS = ROOT / "examples" / "first-run" / "tools" if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS)) import materialize_demo as consumer # noqa: E402 import prepare_demo_snapshot as producer # noqa: E402 PT_LOAD = producer.PT_LOAD EM_MOS_6502 = producer.EM_MOS_6502 # -------------------------------------------------------------------------- # Synthetic artifact builders # -------------------------------------------------------------------------- SOURCE_BODIES = { "Cargo.toml": "[workspace]\nmembers = [\"crates/ultimate-tool-c64\"]\n", "crates/ultimate-tool-c64/Cargo.toml": "[package]\nname = \"ultimate-tool-c64\"\n", "crates/ultimate-tool-c64/src/lib.rs": "pub fn placeholder() {}\n", "fixtures/rust-mos-smoke/Cargo.toml": "[package]\nname = \"ultimate-tool-smoke\"\n", "fixtures/rust-mos-smoke/Cargo.lock": "version = 4\n", "fixtures/rust-mos-smoke/build.rs": "fn main() {}\n", "fixtures/rust-mos-smoke/link.ld": "OUTPUT_FORMAT { TRIM(ram) }\n", "fixtures/rust-mos-smoke/.cargo/config.toml": "[build]\ntarget = \"mos-c64-none\"\n", "fixtures/rust-mos-smoke/src/main.rs": "fn main() {}\n", } def sample_segments() -> list[dict]: """A resident layout plus NOLOAD mailbox regions, mirroring the fixture.""" return [ {"vaddr": 0x0801, "file": bytes([0xA0]) * 12}, {"vaddr": 0x080D, "file": bytes([0xB0]) * 64}, {"vaddr": 0x084D, "file": bytes([0xC0]) * 32}, {"vaddr": 0xC000, "file": None, "memsz": 0x100}, {"vaddr": 0xC100, "file": None, "memsz": 0x40}, {"vaddr": 0xC140, "file": None, "memsz": 0x36}, ] def build_elf32( segments: list[dict], *, entry: int | None = None, machine: int = EM_MOS_6502, etype: int = producer.ET_EXEC, ei_class: int = producer.ELFCLASS32, ei_data: int = producer.ELFDATA2LSB, phentsize: int = producer.ELF32_PHENTSIZE, ehsize: int = producer.ELF32_HEADER_SIZE, ) -> bytes: """Serialise a minimal 32-bit ELF with PT_LOAD program headers.""" count = len(segments) phoff = ehsize cursor = phoff + phentsize * count headers = b"" body = b"" first_resident = None for segment in segments: vaddr = segment["vaddr"] if segment.get("file") is None: offset, filesz, memsz = cursor, 0, segment["memsz"] else: payload = segment["file"] offset, filesz = cursor, len(payload) memsz = segment.get("memsz", filesz) cursor += filesz body += payload if first_resident is None: first_resident = vaddr headers += struct.pack( " bytes: if resident is None: resident = b"".join( segment["file"] for segment in sample_segments() if segment.get("file") ) return struct.pack(" None: for relative, body in SOURCE_BODIES.items(): target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_text(body, encoding="utf-8") class SnapshotTestCase(unittest.TestCase): """Shared scaffolding: a synthetic project, artifacts, and driven CLIs.""" def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.tmp = Path(self._tmp.name) self.project = self.tmp / "repo" write_project(self.project) self.artifacts = self.tmp / "artifacts" self.artifacts.mkdir() self.package = self.tmp / "package" self.prg = sample_prg() self.elf = build_elf32(sample_segments()) self.map = b"Memory Configuration\nbasic_header 0x0801\n" self._write_artifacts() def _write_artifacts(self) -> None: (self.artifacts / "smoke.prg").write_bytes(self.prg) (self.artifacts / "smoke.map").write_bytes(self.map) (self.artifacts / "smoke.elf").write_bytes(self.elf) # -- CLI drivers ------------------------------------------------------- def run_prepare( self, *extra: str, package: Path | None = None, revision: str | None = "0" * 40, ) -> tuple[int, str, str]: argv = [ "--project-root", str(self.project), "--package", str(package or self.package), "--prg", str(self.artifacts / "smoke.prg"), "--map", str(self.artifacts / "smoke.map"), "--elf", str(self.artifacts / "smoke.elf"), "--expect-prg-sha256", producer.sha256_bytes(self.prg), *extra, ] if revision is not None: argv += ["--source-revision", revision] out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): code = producer.main(argv) return code, out.getvalue(), err.getvalue() def run_materialize( self, dest: Path, *extra: str, package: Path | None = None ) -> tuple[int, str, str]: argv = ["--package", str(package or self.package), "--dest", str(dest), *extra] out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): code = consumer.main(argv) return code, out.getvalue(), err.getvalue() def dest(self) -> Path: return self.tmp / "checkout" def prepare_ok(self, *extra: str) -> dict: code, _out, err = self.run_prepare(*extra) self.assertEqual(code, 0, err) return producer.read_snapshot(self.package / producer.SNAPSHOT_FILENAME) def entries_by_path(self, snapshot: dict) -> dict: return {entry["path"]: entry for entry in snapshot["file_entries"]} def prg_entry_path(self) -> str: return f"{producer.OUTPUTS_DIRNAME}/build/ultimate-tool-smoke.prg" def packaged(self, entry_path: str) -> Path: return self.package / entry_path # -------------------------------------------------------------------------- # Producer: happy path, schema, idempotence # -------------------------------------------------------------------------- class PrepareTests(SnapshotTestCase): def test_writes_schema1_package(self) -> None: snapshot = self.prepare_ok() self.assertEqual(snapshot["schema"], 1) self.assertEqual(snapshot["kind"], producer.SNAPSHOT_KIND) self.assertEqual(snapshot["demo"], "first-run") self.assertEqual(snapshot["producer"]["source_revision"], "0" * 40) self.assertFalse(snapshot["producer"]["source_dirty"]) self.assertEqual(snapshot["producer"]["image"], producer.DEFAULT_IMAGE) self.assertIsNone(snapshot["producer"]["image_digest"]) self.assertEqual(snapshot["producer"]["command"], producer.DEFAULT_COMMAND) self.assertIn("make build", snapshot["producer"]["reproduction"]) entries = self.entries_by_path(snapshot) # Three outputs, declared with output=true. self.assertEqual( sum(1 for entry in entries.values() if entry["output"]), 3 ) for _name, destination in producer.OUTPUT_FILES: self.assertIn(f"{producer.OUTPUTS_DIRNAME}/{destination}", entries) # Every allowlisted source, including the SDK crate source and root manifest. for relative in producer.SOURCE_FILES: self.assertIn(f"{producer.SOURCES_DIRNAME}/{relative}", entries) self.assertIn( f"{producer.SOURCES_DIRNAME}/crates/ultimate-tool-c64/src/lib.rs", entries ) # Paths are sorted and hashes agree with the bytes on disk. paths = [entry["path"] for entry in snapshot["file_entries"]] self.assertEqual(paths, sorted(paths)) for entry in snapshot["file_entries"]: data = self.packaged(entry["path"]).read_bytes() self.assertEqual(len(data), entry["size"]) self.assertEqual(producer.sha256_bytes(data), entry["sha256"]) # The PRG entry records the accepted artifact hash. self.assertEqual( entries[self.prg_entry_path()]["sha256"], producer.sha256_bytes(self.prg), ) def test_records_verified_prg_and_elf_facts(self) -> None: snapshot = self.prepare_ok() self.assertEqual(snapshot["prg"]["load_address"], 0x0801) self.assertEqual(snapshot["prg"]["resident_bytes"], 108) self.assertEqual(snapshot["prg"]["payload_bytes"], 108) self.assertEqual(snapshot["prg"]["padding_bytes"], 0) elf = snapshot["elf"] self.assertEqual(elf["class"], 32) self.assertEqual(elf["data"], "little-endian") self.assertEqual(elf["type"], "exec") self.assertEqual(elf["machine"], EM_MOS_6502) self.assertEqual(elf["file_backed_segments"], 3) self.assertEqual(elf["noload_segments"], 3) def test_idempotent_rerun_reuses_files_and_leaves_no_stage(self) -> None: self.prepare_ok() before = { entry["path"]: self.packaged(entry["path"]).read_bytes() for entry in producer.read_snapshot(self.package / producer.SNAPSHOT_FILENAME)["file_entries"] } code, out, err = self.run_prepare("--json") self.assertEqual(code, 0, err) report = json.loads(out) self.assertEqual(report["written"], [], report) self.assertTrue(report["reused"]) after = { entry["path"]: self.packaged(entry["path"]).read_bytes() for entry in producer.read_snapshot(self.package / producer.SNAPSHOT_FILENAME)["file_entries"] } self.assertEqual(before, after) self.assertEqual(producer.find_owned_stages(self.package), []) for directory in (self.package / "outputs", self.package / "sources"): for current, _dirs, files in os.walk(directory): for name in files: self.assertFalse(name.endswith(producer.STAGE_SUFFIX)) def test_dry_run_writes_nothing(self) -> None: code, out, err = self.run_prepare("--dry-run", "--json") self.assertEqual(code, 0, err) self.assertTrue(json.loads(out)["dry_run"]) self.assertFalse(self.package.exists()) def test_rejects_accepted_hash_mismatch_by_default(self) -> None: argv = [ "--project-root", str(self.project), "--package", str(self.package), "--prg", str(self.artifacts / "smoke.prg"), "--map", str(self.artifacts / "smoke.map"), "--elf", str(self.artifacts / "smoke.elf"), "--source-revision", "0" * 40, ] out, err = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): code = producer.main(argv) self.assertEqual(code, 1) self.assertIn(producer.ACCEPTED_SMOKE_PRG_SHA256, err.getvalue()) self.assertFalse(self.package.exists()) def test_records_null_revision_when_git_is_unavailable(self) -> None: self.project = self.tmp / "not-a-repo" write_project(self.project) code, _out, err = self.run_prepare(revision=None) self.assertEqual(code, 0, err) snapshot = producer.read_snapshot(self.package / producer.SNAPSHOT_FILENAME) self.assertIsNone(snapshot["producer"]["source_revision"]) self.assertFalse(snapshot["producer"]["source_dirty"]) def test_records_explicit_image_digest(self) -> None: snapshot = self.prepare_ok("--image-digest", "sha256:" + "a" * 64) self.assertEqual(snapshot["producer"]["image_digest"], "sha256:" + "a" * 64) def test_rejects_empty_map(self) -> None: (self.artifacts / "smoke.map").write_bytes(b" \n") code, _out, err = self.run_prepare() self.assertEqual(code, 1) self.assertIn("map output is empty", err) def test_rejects_missing_source_file(self) -> None: (self.project / "fixtures/rust-mos-smoke/src/main.rs").unlink() code, _out, err = self.run_prepare() self.assertEqual(code, 1) self.assertIn("does not exist", err) def test_rejects_undeclared_file_already_in_package(self) -> None: self.prepare_ok() stale = self.package / "outputs" / "stale.bin" stale.write_bytes(b"stale") code, _out, err = self.run_prepare() self.assertEqual(code, 1) self.assertIn("undeclared", err) def test_print_schema_declares_paths(self) -> None: out = io.StringIO() with contextlib.redirect_stdout(out): code = producer.main(["--print-schema"]) self.assertEqual(code, 0) schema = json.loads(out.getvalue()) self.assertEqual(schema["schema"], 1) self.assertEqual(schema["accepted_prg_sha256"], producer.ACCEPTED_SMOKE_PRG_SHA256) self.assertIn("crates/ultimate-tool-c64/src", schema["declared_sdk_source_dir"]) for _name, destination in producer.OUTPUT_FILES: self.assertIn( f"{producer.OUTPUTS_DIRNAME}/{destination}", schema["declared_outputs"].values(), ) # -------------------------------------------------------------------------- # ELF verification # -------------------------------------------------------------------------- class ElfVerificationTests(unittest.TestCase): def test_accepts_sample(self) -> None: info = producer.verify_elf(build_elf32(sample_segments())) self.assertEqual(info["load_address"], 0x0801) self.assertEqual(info["resident_bytes"], 108) self.assertEqual(info["fingerprint"]["file_backed_segments"], 3) self.assertEqual(info["fingerprint"]["noload_segments"], 3) def test_rejects_bad_magic(self) -> None: blob = bytearray(build_elf32(sample_segments())) blob[1:4] = b"XXX" with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(bytes(blob)) self.assertIn("magic", str(caught.exception)) def test_rejects_64bit_class(self) -> None: blob = build_elf32(sample_segments(), ei_class=2) with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(blob) self.assertIn("32-bit", str(caught.exception)) def test_rejects_big_endian(self) -> None: blob = build_elf32(sample_segments(), ei_data=2) with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(blob) self.assertIn("little-endian", str(caught.exception)) def test_rejects_wrong_machine(self) -> None: blob = build_elf32(sample_segments(), machine=0x3E) with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(blob) self.assertIn("6502", str(caught.exception)) def test_rejects_truncated_header(self) -> None: with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(b"\x7fELF\x01\x01\x01") self.assertIn("shorter", str(caught.exception)) def test_rejects_no_file_backed_segment(self) -> None: segments = [{"vaddr": 0xC000, "file": None, "memsz": 0x100}] with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(build_elf32(segments)) self.assertIn("no file-backed", str(caught.exception)) def test_rejects_gapped_resident_ranges(self) -> None: segments = [ {"vaddr": 0x0801, "file": b"A" * 8}, {"vaddr": 0x0810, "file": b"B" * 8}, ] with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(build_elf32(segments)) self.assertIn("not contiguous", str(caught.exception)) def test_rejects_overlapping_resident_ranges(self) -> None: segments = [ {"vaddr": 0x0801, "file": b"A" * 16}, {"vaddr": 0x0808, "file": b"B" * 8}, ] with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(build_elf32(segments)) self.assertIn("overlaps", str(caught.exception)) def test_rejects_noload_overlapping_resident(self) -> None: segments = [ {"vaddr": 0x0801, "file": b"A" * 16}, {"vaddr": 0x0805, "file": None, "memsz": 0x10}, ] with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(build_elf32(segments)) self.assertIn("NOLOAD", str(caught.exception)) def test_rejects_entry_outside_resident(self) -> None: blob = build_elf32(sample_segments(), entry=0xD000) with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(blob) self.assertIn("entry", str(caught.exception)) def test_rejects_address_above_16_bits(self) -> None: segments = [{"vaddr": 0xFF00, "file": b"A" * 0x200}] with self.assertRaises(producer.SnapshotError) as caught: producer.verify_elf(build_elf32(segments)) self.assertIn("16-bit", str(caught.exception)) def test_rejects_program_header_table_past_end(self) -> None: blob = bytearray(build_elf32(sample_segments())) struct.pack_into(" None: self.elf = producer.verify_elf(build_elf32(sample_segments())) self.resident = b"".join( segment["file"] for segment in sample_segments() if segment.get("file") ) def test_accepts_exact_image(self) -> None: info = producer.verify_prg(sample_prg(), self.elf) self.assertEqual(info["load_address"], 0x0801) self.assertEqual(info["resident_bytes"], len(self.resident)) self.assertEqual(info["payload_bytes"], len(self.resident)) def test_accepts_declared_padding(self) -> None: blob = sample_prg(padding=b"\x00\x00") info = producer.verify_prg(blob, self.elf, padding_bytes=2) self.assertEqual(info["padding_bytes"], 2) self.assertEqual(info["payload_bytes"], len(self.resident) + 2) def test_rejects_undeclared_padding(self) -> None: with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(sample_prg(padding=b"\x00"), self.elf) self.assertIn("resident image", str(caught.exception)) def test_rejects_load_address_mismatch(self) -> None: with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(sample_prg(load_address=0x0900), self.elf) self.assertIn("load address", str(caught.exception)) def test_rejects_truncated_payload(self) -> None: with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(sample_prg()[:-1], self.elf) self.assertIn("payload", str(caught.exception)) def test_rejects_flattened_noload(self) -> None: # Concatenating the NOLOAD memory images into the PRG is the exact bug # the check exists to catch: it would add the $C000 mailbox regions. flattened = ( struct.pack(" None: blob = bytearray(sample_prg()) blob[2] ^= 0xFF with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(bytes(blob), self.elf) self.assertIn("do not match", str(caught.exception)) def test_rejects_payload_over_limit(self) -> None: big = {"vaddr": 0x0801, "file": b"Z" * (producer.MAX_PRG_PAYLOAD_BYTES + 8)} elf = producer.verify_elf(build_elf32([big])) with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(struct.pack(" None: with self.assertRaises(producer.SnapshotError) as caught: producer.verify_prg(b"\x00", self.elf) self.assertIn("too short", str(caught.exception)) # -------------------------------------------------------------------------- # Schema validation # -------------------------------------------------------------------------- class SchemaValidationTests(SnapshotTestCase): def good_snapshot(self) -> dict: return self.prepare_ok() def test_rejects_non_object(self) -> None: for document in ([], "x", 3, None): with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(document) def test_rejects_unknown_top_level_key(self) -> None: snapshot = self.good_snapshot() snapshot["extra"] = 1 with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("unknown top-level", str(caught.exception)) def test_rejects_unsupported_schema_version(self) -> None: snapshot = self.good_snapshot() snapshot["schema"] = 2 with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("unsupported snapshot schema", str(caught.exception)) def test_rejects_wrong_schema_type(self) -> None: snapshot = self.good_snapshot() snapshot["schema"] = "1" with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_missing_top_level_key(self) -> None: for key in ("schema", "kind", "demo", "producer", "file_entries"): snapshot = self.good_snapshot() del snapshot[key] with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_unknown_producer_key(self) -> None: snapshot = self.good_snapshot() snapshot["producer"]["extra"] = "x" with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("producer has unknown keys", str(caught.exception)) def test_rejects_bad_revision_and_digest(self) -> None: for key, value in (("source_revision", "XYZ"), ("image_digest", "sha256:zz")): snapshot = self.good_snapshot() snapshot["producer"][key] = value with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_non_boolean_dirty(self) -> None: snapshot = self.good_snapshot() snapshot["producer"]["source_dirty"] = "no" with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_path_traversal(self) -> None: for bad in ("../escape", "outputs/../../escape", "sources/./x", "sources/a/../b"): snapshot = self.good_snapshot() snapshot["file_entries"][0]["path"] = bad with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("unsafe component", str(caught.exception)) def test_rejects_absolute_and_backslash_paths(self) -> None: for bad in ("/etc/passwd", "C:\\x", "sources\\x"): snapshot = self.good_snapshot() snapshot["file_entries"][0]["path"] = bad with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_path_outside_layout(self) -> None: snapshot = self.good_snapshot() snapshot["file_entries"][0]["path"] = "other/file" with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("must start with", str(caught.exception)) def test_rejects_output_flag_mismatch(self) -> None: snapshot = self.good_snapshot() entry = next(item for item in snapshot["file_entries"] if item["output"]) entry["output"] = False with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("output flag", str(caught.exception)) def test_rejects_non_boolean_output_flag(self) -> None: snapshot = self.good_snapshot() snapshot["file_entries"][0]["output"] = 1 with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_duplicate_path(self) -> None: snapshot = self.good_snapshot() snapshot["file_entries"].append(dict(snapshot["file_entries"][0])) with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("duplicate", str(caught.exception)) def test_rejects_bad_sha256(self) -> None: for bad in ("zz", "A" * 64, 12345): snapshot = self.good_snapshot() snapshot["file_entries"][0]["sha256"] = bad with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_oversized_entry_and_total(self) -> None: snapshot = self.good_snapshot() snapshot["file_entries"][0]["size"] = producer.MAX_FILE_BYTES + 1 with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) snapshot = self.good_snapshot() for entry in snapshot["file_entries"]: entry["size"] = producer.MAX_FILE_BYTES snapshot["file_entries"] = [ dict(entry, path=f"sources/generated/{index}") for index, entry in enumerate( snapshot["file_entries"] * 10 ) ] with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_rejects_missing_and_unknown_entry_keys(self) -> None: snapshot = self.good_snapshot() del snapshot["file_entries"][0]["size"] with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("missing", str(caught.exception)) snapshot = self.good_snapshot() snapshot["file_entries"][0]["mode"] = "0644" with self.assertRaises(producer.SnapshotError) as caught: producer.validate_snapshot(snapshot) self.assertIn("unknown keys", str(caught.exception)) def test_rejects_too_many_entries(self) -> None: snapshot = self.good_snapshot() template = snapshot["file_entries"][0] snapshot["file_entries"] = [ dict(template, path=f"sources/gen-{index}", size=0) for index in range(producer.MAX_FILE_ENTRIES + 1) ] with self.assertRaises(producer.SnapshotError): producer.validate_snapshot(snapshot) def test_read_snapshot_rejects_malformed_json(self) -> None: (self.prepare_ok(),) target = self.package / producer.SNAPSHOT_FILENAME data = target.read_bytes() target.write_bytes(data[: len(data) // 2]) with self.assertRaises(producer.SnapshotError) as caught: producer.read_snapshot(target) self.assertIn("not valid JSON", str(caught.exception)) def test_read_snapshot_rejects_non_object_json(self) -> None: path = self.tmp / "array.json" path.write_text("[1, 2]\n", encoding="utf-8") with self.assertRaises(producer.SnapshotError): producer.read_snapshot(path) def test_read_snapshot_rejects_symlink(self) -> None: real = self.tmp / "real.json" real.write_text("{}", encoding="utf-8") link = self.tmp / "link.json" link.symlink_to(real) with self.assertRaises(producer.SnapshotError) as caught: producer.read_snapshot(link) self.assertIn("symlink", str(caught.exception)) def test_read_snapshot_rejects_oversized_file(self) -> None: path = self.tmp / "big.json" path.write_bytes(b"{" + b" " * (producer.MAX_SNAPSHOT_BYTES + 1)) with self.assertRaises(producer.SnapshotError) as caught: producer.read_snapshot(path) self.assertIn("limit", str(caught.exception)) # -------------------------------------------------------------------------- # Materializer # -------------------------------------------------------------------------- class MaterializeTests(SnapshotTestCase): def installed_prg(self, dest: Path | None = None) -> Path: return (dest or self.dest()) / "build/ultimate-tool-smoke.prg" def test_installs_declared_outputs(self) -> None: self.prepare_ok() code, _out, err = self.run_materialize(self.dest(), "--json") self.assertEqual(code, 0, err) summary = json.loads(_out) self.assertEqual(summary["selected"], 3) self.assertEqual(len(summary["installed"]), 3) self.assertEqual(summary["reused"], []) self.assertEqual(self.installed_prg().read_bytes(), self.prg) self.assertEqual( (self.dest() / "build/ultimate-tool-smoke.elf").read_bytes(), self.elf, ) self.assertEqual( (self.dest() / "build/ultimate-tool-smoke.map").read_bytes(), self.map, ) # Sources are not installed by default. self.assertFalse((self.dest() / "Cargo.toml").exists()) self.assertEqual(producer.find_owned_stages(self.installed_prg().parent), []) def test_idempotent_second_run_reuses(self) -> None: self.prepare_ok() self.run_materialize(self.dest()) before = { path.relative_to(self.dest()): path.read_bytes() for path in (self.dest()).rglob("*") if path.is_file() } code, out, err = self.run_materialize(self.dest(), "--json") self.assertEqual(code, 0, err) summary = json.loads(out) self.assertEqual(summary["installed"], []) self.assertEqual(len(summary["reused"]), 3) after = { path.relative_to(self.dest()): path.read_bytes() for path in (self.dest()).rglob("*") if path.is_file() } self.assertEqual(before, after) def test_check_writes_nothing(self) -> None: self.prepare_ok() code, out, err = self.run_materialize(self.dest(), "--check", "--json") self.assertEqual(code, 0, err) summary = json.loads(out) self.assertTrue(summary["check"]) self.assertEqual(len(summary["verified"]), 3) self.assertEqual(summary["installed"], []) self.assertFalse(self.dest().exists()) def test_check_detects_tampering(self) -> None: self.prepare_ok() target = self.packaged(self.prg_entry_path()) blob = bytearray(target.read_bytes()) blob[-1] ^= 0xFF target.write_bytes(bytes(blob)) code, _out, err = self.run_materialize(self.dest(), "--check") self.assertEqual(code, 1) self.assertIn("sha256", err) self.assertFalse(self.dest().exists()) def test_include_sources_installs_allowlisted_tree(self) -> None: self.prepare_ok() code, out, err = self.run_materialize(self.dest(), "--include-sources", "--json") self.assertEqual(code, 0, err) summary = json.loads(out) self.assertEqual(len(summary["installed"]), 12) self.assertEqual(summary["reused"], []) installed_paths = {record["path"] for record in summary["installed"]} self.assertIn("build/ultimate-tool-smoke.prg", installed_paths) self.assertIn("crates/ultimate-tool-c64/src/lib.rs", installed_paths) self.assertEqual( (self.dest() / "Cargo.toml").read_text(encoding="utf-8"), SOURCE_BODIES["Cargo.toml"], ) self.assertTrue((self.dest() / "crates/ultimate-tool-c64/src/lib.rs").is_file()) def test_detects_tampered_output(self) -> None: self.prepare_ok() target = self.packaged(self.prg_entry_path()) blob = bytearray(target.read_bytes()) blob[3] ^= 0xFF target.write_bytes(bytes(blob)) code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("sha256", err) self.assertFalse(self.installed_prg().exists()) def test_detects_truncated_output(self) -> None: self.prepare_ok() target = self.packaged(self.prg_entry_path()) target.write_bytes(target.read_bytes()[:-2]) code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("bytes", err) def test_rejects_unknown_output_file(self) -> None: self.prepare_ok() extra = self.package / "outputs/build/surprise.bin" extra.parent.mkdir(parents=True, exist_ok=True) extra.write_bytes(b"surprise") code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("undeclared", err) self.assertFalse(self.dest().exists()) def test_rejects_unknown_source_file(self) -> None: self.prepare_ok() extra = self.package / "sources/extra.txt" extra.write_bytes(b"extra") code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("undeclared", err) def test_rejects_symlinked_packaged_output(self) -> None: self.prepare_ok() target = self.packaged(self.prg_entry_path()) real = self.tmp / "real.prg" real.write_bytes(self.prg) target.unlink() target.symlink_to(real) code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("symlink", err) def test_rejects_missing_declared_output(self) -> None: self.prepare_ok() self.packaged(self.prg_entry_path()).unlink() code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("does not exist", err) def test_refuses_conflicting_destination(self) -> None: self.prepare_ok() destination = self.installed_prg() destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(b"a previous, different build") code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("refusing to overwrite", err) self.assertEqual(destination.read_bytes(), b"a previous, different build") def test_reuses_identical_destination(self) -> None: self.prepare_ok() destination = self.installed_prg() destination.parent.mkdir(parents=True, exist_ok=True) destination.write_bytes(self.prg) code, out, err = self.run_materialize(self.dest(), "--json") self.assertEqual(code, 0, err) summary = json.loads(out) reused_paths = {record["path"] for record in summary["reused"]} self.assertIn("build/ultimate-tool-smoke.prg", reused_paths) def test_rejects_symlinked_destination(self) -> None: self.prepare_ok() destination = self.installed_prg() destination.parent.mkdir(parents=True, exist_ok=True) real = self.tmp / "elsewhere.prg" real.write_bytes(b"x") destination.symlink_to(real) code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertEqual(real.read_bytes(), b"x") def test_rejects_directory_destination(self) -> None: self.prepare_ok() destination = self.installed_prg() destination.mkdir(parents=True) code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("not a regular file", err) def test_missing_snapshot_is_reported(self) -> None: code, _out, err = self.run_materialize(self.dest()) self.assertEqual(code, 1) self.assertIn("does not exist", err) def test_default_package_root_is_the_tools_parent(self) -> None: self.assertEqual(consumer.default_package_dir(), TOOLS.parent) def test_default_destination_is_the_package_dir(self) -> None: self.assertEqual( consumer.default_destination(self.package), self.package ) self.assertEqual( consumer.default_destination(TOOLS.parent), TOOLS.parent ) # -------------------------------------------------------------------------- # Staging hygiene # -------------------------------------------------------------------------- class StagingTests(SnapshotTestCase): def test_install_leaves_no_stage_on_success(self) -> None: destination = self.tmp / "out/file.bin" outcome = producer.install_payload_file( b"payload", destination, expected_sha256=producer.sha256_bytes(b"payload") ) self.assertEqual(outcome, "installed") self.assertEqual(destination.read_bytes(), b"payload") self.assertEqual(producer.find_owned_stages(destination.parent), []) def test_install_cleans_owned_stage_on_link_failure(self) -> None: destination = self.tmp / "out/file.bin" with unittest.mock.patch.object( producer.os, "link", side_effect=OSError(errno.ENOSPC, "no space") ): with self.assertRaises(producer.SnapshotError) as caught: producer.install_payload_file(b"payload", destination) self.assertIn("cannot install", str(caught.exception)) self.assertFalse(os.path.lexists(destination)) self.assertEqual(producer.find_owned_stages(destination.parent), []) def test_unsupported_atomic_publication_never_writes_final_leaf(self) -> None: for error_code in (errno.EPERM, errno.EOPNOTSUPP, errno.EXDEV, errno.EMLINK): with self.subTest(error_code=error_code): destination = self.tmp / f"unsupported-{error_code}/file.bin" destination.parent.mkdir() unrelated = destination.parent / "unrelated.bin" unrelated.write_bytes(b"preserve") with unittest.mock.patch.object(producer.os, "link", side_effect=OSError(error_code, "unsupported")): with self.assertRaises(producer.SnapshotError) as caught: producer.install_payload_file(b"payload", destination) self.assertIn("atomic publication", str(caught.exception)) self.assertFalse(os.path.lexists(destination)) self.assertEqual(unrelated.read_bytes(), b"preserve") self.assertEqual(producer.find_owned_stages(destination.parent), []) def test_install_cleans_owned_stage_when_write_fails(self) -> None: destination = self.tmp / "out/file.bin" with unittest.mock.patch.object( producer.os, "fsync", side_effect=OSError(errno.EIO, "disk error") ): with self.assertRaises(OSError): producer.install_payload_file(b"payload", destination) self.assertFalse(os.path.lexists(destination)) self.assertEqual(producer.find_owned_stages(destination.parent), []) def test_install_verifies_staged_bytes(self) -> None: destination = self.tmp / "out/file.bin" with self.assertRaises(producer.SnapshotError) as caught: producer.install_payload_file( b"payload", destination, expected_sha256="0" * 64 ) self.assertIn("do not match", str(caught.exception)) self.assertFalse(os.path.lexists(destination)) def test_partial_failure_keeps_earlier_installs_and_removes_stages(self) -> None: self.prepare_ok() dest = self.tmp / "checkout" build = dest / "build" # Entries install in sorted path order: .elf, then .map, then .prg. # Pre-create a conflicting destination for the map output so the run # fails after the ELF has already been installed. conflict = build / "ultimate-tool-smoke.map" conflict.parent.mkdir(parents=True, exist_ok=True) conflict.write_bytes(b"different") code, _out, err = self.run_materialize(dest) self.assertEqual(code, 1) self.assertIn("refusing to overwrite", err) self.assertTrue((build / "ultimate-tool-smoke.elf").is_file()) self.assertEqual(conflict.read_bytes(), b"different") self.assertFalse((build / "ultimate-tool-smoke.prg").exists()) for directory in {path.parent for path in dest.rglob("*") if path.is_file()}: self.assertEqual(producer.find_owned_stages(directory), []) def test_foreign_stage_like_file_is_not_owned(self) -> None: destination = self.tmp / "out/file.bin" destination.parent.mkdir(parents=True, exist_ok=True) foreign = destination.parent / "unrelated.stage" foreign.write_bytes(b"someone else's") producer.install_payload_file(b"payload", destination) self.assertTrue(foreign.is_file()) self.assertEqual(foreign.read_bytes(), b"someone else's") # -------------------------------------------------------------------------- # Traversal hardening: symlinked roots, ancestors, and parent swaps # -------------------------------------------------------------------------- class SymlinkHardeningTests(SnapshotTestCase): def test_producer_rejects_symlinked_source_file(self) -> None: real = self.tmp / "real-main.rs" real.write_text("fn main() {}\n", encoding="utf-8") target = self.project / "fixtures/rust-mos-smoke/src/main.rs" target.unlink() target.symlink_to(real) code, _out, err = self.run_prepare() self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertFalse(self.package.exists()) def test_rejects_symlinked_destination_root(self) -> None: self.prepare_ok() real = self.tmp / "real-root" real.mkdir() link = self.tmp / "linked-root" link.symlink_to(real, target_is_directory=True) code, _out, err = self.run_materialize(link) self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertEqual(list(real.rglob("*")), []) def test_rejects_symlinked_ancestor_in_destination(self) -> None: self.prepare_ok() dest = self.tmp / "checkout" outside = self.tmp / "outside" outside.mkdir() (dest / "build").parent.mkdir(parents=True) (dest / "build").symlink_to(outside, target_is_directory=True) code, _out, err = self.run_materialize(dest) self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertEqual(list(outside.iterdir()), []) def test_rejects_preexisting_symlinked_build_directory(self) -> None: self.prepare_ok() dest = self.tmp / "checkout" build_real = self.tmp / "build-real" build_real.mkdir() (dest).mkdir() (dest / "build").symlink_to(build_real, target_is_directory=True) code, _out, err = self.run_materialize(dest) self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertEqual(list(build_real.iterdir()), []) def test_rejects_symlinked_package_root(self) -> None: self.prepare_ok() link = self.tmp / "linked-package" link.symlink_to(self.package, target_is_directory=True) code, _out, err = self.run_materialize(self.dest(), package=link) self.assertEqual(code, 1) self.assertIn("symlink", err) self.assertFalse(self.dest().exists()) def test_rejects_destination_path_with_parent_component(self) -> None: self.prepare_ok() escape = self.tmp / "nested" / ".." / "outside" (self.tmp / "nested").mkdir() code, _out, err = self.run_materialize(escape) self.assertEqual(code, 1) self.assertIn("..", err) self.assertFalse((self.tmp / "outside").exists()) def test_install_rejects_symlinked_destination_component(self) -> None: real = self.tmp / "real" real.mkdir() (real / "file.bin").write_bytes(b"other") link = self.tmp / "link" link.symlink_to(real, target_is_directory=True) with self.assertRaises(producer.SnapshotError) as caught: producer.install_payload_file(b"payload", link / "file.bin") self.assertIn("symlink", str(caught.exception)) self.assertEqual((real / "file.bin").read_bytes(), b"other") def test_parent_swap_cannot_redirect_writes_outside_root(self) -> None: self.prepare_ok() dest = self.tmp / "checkout" build = dest / "build" build.mkdir(parents=True) outside = self.tmp / "outside" outside.mkdir() real_link = producer.os.link state = {"swapped": False} def swapping_link(src, dst, *, src_dir_fd=None, dst_dir_fd=None, **kwargs): # Deterministically replace the *path* `dest/build` with a symlink to # `outside` at the moment of publication. Descriptors already opened # by the installer still reference the original directory inode. if not state["swapped"]: state["swapped"] = True build.rename(dest / "build-real") build.symlink_to(outside, target_is_directory=True) return real_link(src, dst, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd, **kwargs) with unittest.mock.patch.object(producer.os, "link", side_effect=swapping_link): code, _out, err = self.run_materialize(dest) self.assertTrue(state["swapped"], err) # Nothing may be created through the swapped symlink. self.assertEqual(list(outside.iterdir()), []) # The run is rejected once the swapped parent is traversed again ... self.assertEqual(code, 1) self.assertIn("symlink", err) # ... and the first publication landed inside the root's real directory. anchored = dest / "build-real" / "ultimate-tool-smoke.elf" self.assertTrue(anchored.is_file()) self.assertEqual(anchored.read_bytes(), self.elf) self.assertEqual(producer.find_owned_stages(dest / "build-real"), []) # -------------------------------------------------------------------------- # FIFO rejection, exercised in subprocesses with a bounded timeout so a hang # fails the test instead of blocking the suite. # -------------------------------------------------------------------------- def _mktmp_fifo(path: Path) -> None: path.parent.mkdir(parents=True, exist_ok=True) os.mkfifo(path) def _run_cli(script: Path, args: list[str], *, timeout: int = 20) -> subprocess.CompletedProcess: return subprocess.run( [sys.executable, str(script), *args], capture_output=True, text=True, timeout=timeout, check=False, ) class FifoRejectionTests(SnapshotTestCase): MATERIALIZER = TOOLS / "materialize_demo.py" PRODUCER = TOOLS / "prepare_demo_snapshot.py" def test_snapshot_fifo_is_rejected_without_blocking(self) -> None: self.prepare_ok() target = self.package / producer.SNAPSHOT_FILENAME target.unlink() _mktmp_fifo(target) result = _run_cli(self.MATERIALIZER, ["--package", str(self.package), "--dest", str(self.dest())]) self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("FIFO", result.stderr) self.assertFalse(self.dest().exists()) def test_packaged_output_fifo_is_rejected_without_blocking(self) -> None: self.prepare_ok() target = self.packaged(self.prg_entry_path()) target.unlink() _mktmp_fifo(target) result = _run_cli(self.MATERIALIZER, ["--package", str(self.package), "--dest", str(self.dest())]) self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("FIFO", result.stderr) self.assertFalse(self.dest().exists()) def test_packaged_source_fifo_is_rejected_without_blocking(self) -> None: self.prepare_ok() target = self.package / producer.SOURCES_DIRNAME / "Cargo.toml" target.unlink() _mktmp_fifo(target) result = _run_cli(self.MATERIALIZER, ["--package", str(self.package), "--dest", str(self.dest())]) self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("FIFO", result.stderr) def test_producer_input_fifo_is_rejected_without_blocking(self) -> None: fifo = self.artifacts / "fifo.prg" _mktmp_fifo(fifo) result = _run_cli( self.PRODUCER, [ "--project-root", str(self.project), "--package", str(self.package), "--prg", str(fifo), "--map", str(self.artifacts / "smoke.map"), "--elf", str(self.artifacts / "smoke.elf"), "--source-revision", "0" * 40, "--expect-prg-sha256", "none", ], ) self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("FIFO", result.stderr) self.assertFalse(self.package.exists()) def test_source_fifo_is_rejected_by_producer_without_blocking(self) -> None: target = self.project / "fixtures/rust-mos-smoke/src/main.rs" target.unlink() _mktmp_fifo(target) result = _run_cli( self.PRODUCER, [ "--project-root", str(self.project), "--package", str(self.package), "--prg", str(self.artifacts / "smoke.prg"), "--map", str(self.artifacts / "smoke.map"), "--elf", str(self.artifacts / "smoke.elf"), "--source-revision", "0" * 40, "--expect-prg-sha256", producer.sha256_bytes(self.prg), ], ) self.assertEqual(result.returncode, 1, result.stderr) self.assertIn("FIFO", result.stderr) class RealArtifactTests(unittest.TestCase): BUILD = ROOT / "fixtures/rust-mos-smoke/build" def test_pinned_artifact_matches_recorded_hash(self) -> None: prg = self.BUILD / "ultimate-tool-smoke.prg" elf = self.BUILD / "ultimate-tool-smoke.elf" if not (prg.is_file() and elf.is_file()): # pragma: no cover - gitignored self.skipTest("pinned Rust-MOS outputs are not built") elf_info = producer.verify_elf(elf.read_bytes()) prg_info = producer.verify_prg(prg.read_bytes(), elf_info) self.assertEqual(prg_info["load_address"], 0x0801) self.assertEqual(producer.sha256_bytes(prg.read_bytes()), producer.ACCEPTED_SMOKE_PRG_SHA256) self.assertEqual(elf_info["fingerprint"]["machine"], EM_MOS_6502) self.assertEqual(elf_info["fingerprint"]["file_backed_segments"], 3) self.assertEqual(elf_info["fingerprint"]["noload_segments"], 3) self.assertEqual(prg_info["resident_bytes"], 4930) self.assertEqual(elf_info["noload"][0]["vaddr"], 0xC000) if __name__ == "__main__": unittest.main(verbosity=2)