Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
61 kB · 1154 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155#!/usr/bin/env python3"""Tests for tools/build_release_source.py.
These are unit tests over the harness's wiring: the prepared inventory is closedand the paths are validated before extraction, the extraction request is themanifest-derived one, the extracted tree must match the closed inventory exactly(so an unlisted ``build.rs`` can never be compiled), the work directory iscreated exclusively and never reused, the environment descriptor is written withthe pinned image's exclusive writer, the three container steps run in order withthe documented command and offline environment, the record exposes the``binary_dir`` and ``build_env`` paths the packager is given, and everylaunched-command failure — a non-zero exit, a missing or oversized binary, anunreadable or malformed descriptor, a failed copy, a timeout — is recordedhonestly with a non-zero exit.
The record is also checked *not* to be the pack gate's evidence: it is the firstbuild, so it must never be accepted as ``release_rebuild_evidence``.
No container runs and nothing is compiled here. A green result is evidence aboutthe harness, never a verified build."""
from __future__ import annotations
import contextlibimport hashlibimport ioimport jsonimport osimport statimport subprocessimport sysimport tarfileimport tempfileimport unittestfrom pathlib import Pathfrom unittest import mock
ROOT = Path(__file__).resolve().parents[2]TOOLS = ROOT / "tools"if str(TOOLS) not in sys.path: sys.path.insert(0, str(TOOLS))
import build_release_source as build # noqa: E402import package_release as packaging # noqa: E402import verify_release_local as verify # noqa: E402
SOURCE_PREFIX = "ultimate-tool-0.3.0-preview.1-linux-x86_64-gnu-source"BINARIES = {name: f"rebuilt {name}\n".encode("utf-8") for name in verify.BINARY_NAMES}RUNTIME_SOURCE_MEMBERS = { "runtime-sources/bundle.tar.gz": b"runtime source archive\n", "runtime-sources/bundle.json": b'{"record_kind": "runtime_source_bundle"}\n', "runtime-sources/SOURCE-AND-LICENSES.fragment.md": b"runtime fragment\n",}
DESCRIPTOR = { "schema": 1, "record_kind": "build_environment", "observed_at": "2026-01-01T00:00:00Z", "os": {"id": "ubuntu", "version_id": "24.04"}, "glibc": {"version": "2.39"}, "target_triple": verify.PINNED_TARGET_TRIPLE, "toolchain": {"rustc": "rustc 1.93.0 (254b59607 2025-01-06)", "cargo": "cargo 1.93.0 (083ac5135)"}, "base_image": {"reference": verify.PINNED_BASE_REFERENCE, "digest": verify.PINNED_BASE_DIGEST, "pinned": True},}
def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
class DockerInvocationForbidden(AssertionError): """A mocked test asked the real Docker CLI for something and was refused."""
DOCKER_GUARD_NOTE = ( "this test stubs the Docker seams, so no docker subprocess may start; the guard " "refused the invocation instead of asking a daemon")
def looks_like_docker(argv) -> bool: """Whether an argument vector names the Docker CLI.""" if isinstance(argv, bytes): argv = argv.decode("utf-8", "replace") if isinstance(argv, str): argv = argv.split() try: executable = Path(str(argv[0])) except (IndexError, KeyError, TypeError): return False return executable.name.startswith("docker")
@contextlib.contextmanagerdef forbid_docker_subprocesses(): """Refuse every Docker entry point for the life of one test.
The harness resolves the CLI through ``verify.docker_executable`` and then spawns it, so both seams refuse: a mocked test either stubs the seam it needs or fails loudly. Nothing is installed on ``PATH`` and no variable is exported, so the refusal covers one test only and a test elsewhere in the suite can still reach the real daemon seams. """ real_run, real_popen = subprocess.run, subprocess.Popen
def refuse(what: str) -> None: raise DockerInvocationForbidden(f"{DOCKER_GUARD_NOTE}: {what}")
def guarded_run(*args, **kwargs): argv = args[0] if args else kwargs.get("args") if looks_like_docker(argv): refuse(f"subprocess.run({argv!r})") return real_run(*args, **kwargs)
def guarded_popen(*args, **kwargs): argv = args[0] if args else kwargs.get("args") if looks_like_docker(argv): refuse(f"subprocess.Popen({argv!r})") return real_popen(*args, **kwargs)
def guarded_executable() -> str: refuse("docker_executable() was asked for the CLI") return "docker" # pragma: no cover - refuse() always raises
with mock.patch.object(verify, "docker_executable", side_effect=guarded_executable), \ mock.patch.object(verify.subprocess, "run", side_effect=guarded_run), \ mock.patch.object(verify.subprocess, "Popen", side_effect=guarded_popen): yield
def add_member(handle: tarfile.TarFile, name: str, content: bytes, mode: int = 0o644) -> None: info = tarfile.TarInfo(name) info.size = len(content) info.mode = mode handle.addfile(info, io.BytesIO(content))
def write_prepared_archive(directory: Path, *, vendored: bool = True, runtime: bool = False, listed: dict | None = None, unlisted: dict | None = None) -> tuple[Path, dict]: """A prepared-style source archive and the manifest that describes it.
``listed`` entries are added to the archive **and** to the manifest's file list; ``unlisted`` entries are written into the archive only, which is how an extra ``build.rs`` or a reserved descriptor would arrive. """ files = { "Cargo.toml": b'[workspace]\nmembers = ["crates/x"]\n', "crates/x/Cargo.toml": b'[package]\nname = "x"\nversion = "0.3.0-preview.1"\n', "Cargo.lock": b"# lock\n", } files.update(listed or {}) vendor_files = {"serde/src/lib.rs": b"// serde\n"} if vendored else {} generated = {"SOURCE-PROVENANCE.json": b'{"producer": "tools/package_release.py"}\n'} if vendored: generated["vendor-sources.toml"] = b'[source.crates-io]\nreplace-with = "vendored-sources"\n' runtime_members = RUNTIME_SOURCE_MEMBERS if runtime else {} archive = directory / f"{SOURCE_PREFIX}.tar.gz" with tarfile.open(archive, "w:gz") as handle: for name, content in sorted({**files, **vendor_files, **runtime_members}.items()): if name in vendor_files: path = f"{SOURCE_PREFIX}/vendor/{name}" else: path = f"{SOURCE_PREFIX}/{name}" add_member(handle, path, content) for name, content in sorted(generated.items()): add_member(handle, f"{SOURCE_PREFIX}/{name}", content) for name, content in sorted((unlisted or {}).items()): add_member(handle, f"{SOURCE_PREFIX}/{name}", content) file_list = [{"path": n, "sha256": sha256(c), "size": len(c)} for n, c in sorted(files.items())] vendor_list = [{"path": n, "sha256": sha256(c), "size": len(c)} for n, c in sorted(vendor_files.items())] manifest = { "schema": 1, "record_kind": "release_source_manifest_resolved", "archive": {"name": archive.name, "prefix": SOURCE_PREFIX, "sha256": sha256(archive.read_bytes())}, "files": file_list, "aggregate_sha256": verify.aggregate_digest(file_list), "vendor": {"files": vendor_list, "aggregate_sha256": verify.aggregate_digest(vendor_list)}, } if runtime: manifest["runtime_bundle"] = { "present": True, "source_members": [ {"path": name, "sha256": sha256(content), "size": len(content), "role": "runtime"} for name, content in sorted(runtime_members.items()) ], } (directory / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") return archive, manifest
def fake_docker_run(*, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None): """Stand in for the pinned image: write the artifacts a real container would.""" argv = [str(item) for item in argv] root = Path(mount[0]) if mount else Path("/nonexistent") if argv and argv[0].endswith("ultimate-tool-build-env"): (root / build.DESCRIPTOR_HOST_NAME).write_text(json.dumps(DESCRIPTOR), encoding="utf-8") elif argv == list(verify.BUILD_COMMAND): target = root / "target" / "release" target.mkdir(parents=True, exist_ok=True) for name, content in BINARIES.items(): (target / name).write_bytes(content) return verify.CommandRecord(argv=argv, exit_code=0, duration_ms=1, stdout="ok\n")
def docker_outcome(*, build_exit: int = 0, build_error: str | None = None, descriptor_exit: int = 0, descriptor_bytes: bytes | None = None, descriptor_symlink_to=None, vendor_exit: int = 0, produce: bool = True, binary_bytes: dict | None = None, post_build=None): """A docker_run stand-in whose named step behaves as the test needs.
``descriptor_symlink_to`` makes the descriptor step leave a symlink instead of a file, and ``post_build(root, target)`` runs after the binaries are written so a test can plant a symlink inside the build output tree. """ def runner(*, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None): argv = [str(item) for item in argv] root = Path(mount[0]) if mount else Path("/nonexistent") if argv and argv[0].endswith("ultimate-tool-build-env"): descriptor = root / build.DESCRIPTOR_HOST_NAME if descriptor_symlink_to is not None: descriptor.symlink_to(descriptor_symlink_to) else: payload = descriptor_bytes if descriptor_bytes is not None else json.dumps(DESCRIPTOR).encode("utf-8") descriptor.write_bytes(payload) return verify.CommandRecord(argv=argv, exit_code=descriptor_exit, duration_ms=1, stderr="descriptor failed\n") if argv == list(verify.BUILD_COMMAND): if build_error: raise verify.HarnessError(build_error) if produce: target = root / "target" / "release" target.mkdir(parents=True, exist_ok=True) for name, content in (binary_bytes or BINARIES).items(): (target / name).write_bytes(content) if post_build is not None: post_build(root, target) return verify.CommandRecord(argv=argv, exit_code=build_exit, duration_ms=1, stderr="compile error\n") return verify.CommandRecord(argv=argv, exit_code=vendor_exit, duration_ms=1, stdout="copied vendor config\n", stderr="vendor failed\n") return runner
class HarnessCase(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) self.addCleanup(self._tmp.cleanup) # Every test in this file drives the harness with the containers stubbed, # so the daemon seams a test forgets are stubbed here and a real Docker # invocation is refused rather than quietly performed. self.container_removals: list[str] = [] self.image_identities: list[str] = [] guard = forbid_docker_subprocesses() guard.__enter__() self.addCleanup(guard.__exit__, None, None, None) for patcher in ( mock.patch.object(verify, "docker_image_identity", side_effect=self.stub_image_identity), mock.patch.object(verify, "remove_containers_with_label", side_effect=self.stub_container_removal), ): patcher.start() self.addCleanup(patcher.stop)
def stub_image_identity(self, tag: str) -> dict: """The local image identity the record carries, without a daemon.""" self.image_identities.append(tag) return { "tag": tag, "id": "sha256:stubbed-image-identity", "provenance": "stubbed by these harness tests; no daemon was asked", }
def stub_container_removal(self, label: str) -> list[str]: """Container cleanup by label, without ``docker ps`` or ``docker rm``.""" self.container_removals.append(label) return []
def synthetic_checkout(self) -> Path: """A checkout root that cannot contain this test's scratch directory.
The production policy forbids the checkout, its parent and the Docker socket, and it is checked as an ancestor: a checkout at a top-level path such as ``/work`` forbids ``/`` itself. A CI or work checkout may live there, so inheriting the module's real root would make the guard refuse every scratch directory on the volume and fail tests that never touch the checkout. Mocked tests therefore name a checkout that is never created, whose parent holds nothing else, and that is nowhere near the temporary directory the work directory is created in. """ return self.root / "unmounted" / "checkout"
def args(self, *, source_dir: Path, work_dir: Path, build_jobs: int = 2, timeout: float = 1800.0, image: str = verify.RELEASE_ENV_IMAGE, checkout_root: Path | None = None): parsed = build.parse_arguments( [ "--source-dir", str(source_dir), "--work-dir", str(work_dir), "--image", image, "--build-jobs", str(build_jobs), "--timeout", str(timeout), "--checkout-root", str(checkout_root or self.synthetic_checkout()), ] ) parsed.checkout_root = parsed.checkout_root.resolve() return parsed
def argv(self, *, source_dir: Path, work_dir: Path, checkout_root: Path | None = None) -> list[str]: """The arguments ``build.main`` needs, always naming the checkout root.
Nothing here relies on the parser's default checkout root, which is the module's own repository and would forbid this test's scratch directory when the repository sits directly under a filesystem root. """ return [ "--source-dir", str(source_dir), "--work-dir", str(work_dir), "--checkout-root", str(checkout_root or self.synthetic_checkout()), ]
def prepared(self, **kwargs) -> tuple[Path, Path, dict]: dist = self.root / "dist" dist.mkdir(exist_ok=True) _, manifest = write_prepared_archive(dist, **kwargs) return dist, dist / f"{SOURCE_PREFIX}.tar.gz", manifest
def run_mocked(self, args, *, side_effect=fake_docker_run): """Drive the real run_build with every docker entry point mocked.""" journal = verify.Journal() calls: list[list[str]] = []
def counting(**kwargs): calls.append([str(item) for item in kwargs.get("argv") or []]) return side_effect(**kwargs)
with mock.patch.object(verify, "docker_available", return_value=(True, "test")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_image_identity", return_value={"tag": args.image, "id": "sha256:deadbeef"}), \ mock.patch.object(verify, "docker_run", side_effect=counting): try: return build.run_build(args, journal, verify.Cleanup()), journal, calls except Exception as error: # noqa: BLE001 - the caller decides the expectation return error, journal, calls
class WorkDirectoryTests(HarnessCase): def test_creates_a_new_directory_and_refuses_to_reuse_it(self) -> None: work = self.root / "out" self.assertEqual(build.prepare_work_directory(work, []), work) self.assertTrue(work.is_dir()) self.assertEqual(stat.S_IMODE(work.stat().st_mode), 0o700) with self.assertRaises(verify.HarnessError): build.prepare_work_directory(work, [])
def test_a_relative_path_is_normalised_to_an_absolute_one(self) -> None: previous = os.getcwd() os.chdir(self.root) self.addCleanup(os.chdir, previous) work = build.prepare_work_directory(Path("relout"), []) self.assertTrue(work.is_absolute()) self.assertTrue(work.is_dir()) self.assertEqual(work.name, "relout")
def test_refuses_a_symlinked_work_directory(self) -> None: real = self.root / "real" real.mkdir() link = self.root / "link" link.symlink_to(real) with self.assertRaises(verify.HarnessError) as caught: build.prepare_work_directory(link, []) self.assertIn("symlink", str(caught.exception))
def test_refuses_a_symlinked_ancestor(self) -> None: real = self.root / "real" real.mkdir() link = self.root / "alias" link.symlink_to(real) with self.assertRaises(verify.HarnessError) as caught: build.prepare_work_directory(link / "out", []) self.assertIn("symlink", str(caught.exception))
def test_refuses_a_work_directory_inside_a_forbidden_path(self) -> None: checkout = self.root / "checkout" checkout.mkdir() forbidden = [checkout.resolve(), checkout.resolve().parent] with self.assertRaises(verify.HarnessError) as caught: build.prepare_work_directory(checkout / "build", forbidden) self.assertIn("forbidden path", str(caught.exception)) with self.assertRaises(verify.HarnessError): build.prepare_work_directory(checkout.parent / "sibling", forbidden)
def test_refuses_a_missing_parent(self) -> None: with self.assertRaises(verify.HarnessError): build.prepare_work_directory(self.root / "absent" / "out", [])
class PreparedSourceTests(HarnessCase): def test_accepts_a_prepared_directory_and_closes_the_inventory(self) -> None: dist, archive, manifest = self.prepared() prepared = build.open_prepared_source(dist) self.assertEqual(prepared["archive_name"], archive.name) self.assertEqual(prepared["archive_sha256"], sha256(archive.read_bytes())) self.assertTrue(prepared["packager_check"]["checked"]) self.assertTrue(prepared["vendor"]["files"])
def test_refuses_a_missing_manifest_or_archive(self) -> None: with self.assertRaises(verify.HarnessError): build.open_prepared_source(self.root / "nope") empty = self.root / "empty" empty.mkdir() with self.assertRaises(verify.HarnessError): build.open_prepared_source(empty)
def test_refuses_a_symlinked_source_directory(self) -> None: dist, archive, manifest = self.prepared() link = self.root / "linked" link.symlink_to(dist) with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(link) self.assertIn("symlink", str(caught.exception))
def test_refuses_a_broken_aggregate_or_archive_hash(self) -> None: dist, archive, manifest = self.prepared() manifest["aggregate_sha256"] = "0" * 64 (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist) self.assertIn("aggregate", str(caught.exception))
dist2, archive2, manifest2 = self.prepared() manifest2["archive"]["sha256"] = "1" * 64 (dist2 / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest2), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist2) self.assertIn("does not match the hash", str(caught.exception))
def test_refuses_a_missing_or_malformed_archive_hash(self) -> None: for recorded in (None, "not-a-hash", "A" * 64, "0" * 63): dist, archive, manifest = self.prepared() if recorded is None: manifest["archive"].pop("sha256") else: manifest["archive"]["sha256"] = recorded (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist) self.assertIn("SHA-256", str(caught.exception))
def test_refuses_a_non_canonical_archive_name(self) -> None: for name in ("../escape.tar.gz", "/etc/passwd", "sub/dir.tar.gz", ".", "..", ""): dist, archive, manifest = self.prepared() manifest["archive"]["name"] = name (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaises(verify.HarnessError): build.open_prepared_source(dist)
def test_refuses_a_symlinked_archive(self) -> None: dist, archive, manifest = self.prepared() real = archive.rename(self.root / "real.tar.gz") archive.symlink_to(real) with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist) self.assertIn("not a regular file", str(caught.exception))
def test_refuses_a_manifest_listing_no_files(self) -> None: dist, archive, manifest = self.prepared() manifest["files"], manifest["aggregate_sha256"] = [], verify.aggregate_digest([]) (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist) self.assertIn("lists no files", str(caught.exception))
def test_a_changed_member_is_caught_by_the_packager_before_extraction(self) -> None: dist, archive, manifest = self.prepared() with tarfile.open(archive, "w:gz") as handle: add_member(handle, f"{SOURCE_PREFIX}/Cargo.toml", b'[workspace]\nmembers = ["crates/y"]\n') add_member(handle, f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", b'{"producer": "x"}\n') add_member(handle, f"{SOURCE_PREFIX}/vendor-sources.toml", b'[source.crates-io]\n') manifest["archive"]["sha256"] = sha256(archive.read_bytes()) (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") with self.assertRaises(verify.HarnessError) as caught: build.open_prepared_source(dist) self.assertIn("does not match its manifest", str(caught.exception))
def test_nothing_is_extracted_or_created_when_the_inventory_is_not_closed(self) -> None: dist, archive, manifest = self.prepared() manifest["aggregate_sha256"] = "f" * 64 (dist / build.SOURCE_MANIFEST_NAME).write_text(json.dumps(manifest), encoding="utf-8") work = self.root / "out" args = self.args(source_dir=dist, work_dir=work) with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ): with self.assertRaises(verify.HarnessError): build.run_build(args, verify.Journal(), verify.Cleanup()) self.assertFalse(work.exists())
class ClosedInventoryTests(HarnessCase): def test_the_inventory_includes_files_vendor_aliases_generated_and_runtime_members(self) -> None: dist, archive, manifest = self.prepared(runtime=True) inventory = build.closed_inventory(manifest) self.assertEqual(inventory["prefix"], SOURCE_PREFIX) self.assertIn(f"{SOURCE_PREFIX}/Cargo.lock", inventory["files"]) self.assertIn(f"{SOURCE_PREFIX}/vendor/serde/src/lib.rs", inventory["files"]) self.assertIn(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", inventory["required"]) self.assertIn(f"{SOURCE_PREFIX}/vendor-sources.toml", inventory["required"]) for name in RUNTIME_SOURCE_MEMBERS: self.assertIn(f"{SOURCE_PREFIX}/{name}", inventory["files"]) self.assertIn(f"{SOURCE_PREFIX}/{name}", inventory["runtime_members"])
def test_an_unlisted_member_is_rejected_before_any_container_starts(self) -> None: dist, archive, manifest = self.prepared(unlisted={"crates/x/build.rs": b"fn main() {}\n"}) work = self.root / "out" args = self.args(source_dir=dist, work_dir=work) result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn("build.rs", str(result)) self.assertEqual(calls, [])
def test_an_unlisted_cargo_build_script_never_compiles(self) -> None: # The archive hash matches the manifest (it is the archive's own digest), # so only the closed member inventory can catch this. dist, archive, manifest = self.prepared(unlisted={"build.rs": b"fn main() { compile_everything(); }\n"}) self.assertEqual(sha256(archive.read_bytes()), manifest["archive"]["sha256"]) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn("build.rs", str(result)) self.assertEqual(calls, []) self.assertFalse((self.root / "out" / "extracted" / SOURCE_PREFIX / "target").exists())
def test_an_unlisted_directory_is_rejected(self) -> None: dist, archive, manifest = self.prepared(unlisted={"extra-dir/note.txt": b"x\n"}) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertEqual(calls, [])
def test_a_listed_runtime_member_is_accepted(self) -> None: dist, archive, manifest = self.prepared(runtime=True) args = self.args(source_dir=dist, work_dir=self.root / "out") result, journal, calls = self.run_mocked(args) document, code, failures = result self.assertEqual(failures, []) self.assertEqual(code, build.EXIT_OK) self.assertEqual(document["verdict"], "built") self.assertEqual(document["closure"]["unknown_files"], []) self.assertEqual(document["closure"]["unknown_directories"], []) self.assertEqual(document["closure"]["symlinks"], []) self.assertEqual(len(calls), 3) source_root = args.work_dir / "extracted" / SOURCE_PREFIX for name in RUNTIME_SOURCE_MEMBERS: self.assertTrue((source_root / name).is_file()) self.assertIn(f"{SOURCE_PREFIX}/runtime-sources/bundle.tar.gz", document["source_manifest"]["runtime_members"])
def test_a_listed_extra_file_is_accepted_when_it_is_in_the_manifest(self) -> None: dist, archive, manifest = self.prepared(listed={"docs/NOTES.md": b"notes\n"}) args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked(args)[0] self.assertEqual(failures, []) self.assertEqual(document["verdict"], "built")
def test_the_reserved_root_descriptor_is_rejected_even_when_listed(self) -> None: dist, archive, manifest = self.prepared(listed={build.DESCRIPTOR_HOST_NAME: b"{}\n"}) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn("reserved descriptor", str(result)) self.assertEqual(calls, [])
def test_the_reserved_root_descriptor_is_rejected_when_unlisted(self) -> None: dist, archive, manifest = self.prepared(unlisted={build.DESCRIPTOR_HOST_NAME: b"{}\n"}) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn(build.DESCRIPTOR_HOST_NAME, str(result)) self.assertEqual(calls, [])
def test_a_shipped_cargo_config_is_rejected_before_any_container(self) -> None: dist, archive, manifest = self.prepared(listed={".cargo/config.toml": b"[source]\n"}) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn(".cargo/config.toml", str(result)) self.assertEqual(calls, [])
class ExtractionRequestTests(HarnessCase): def test_extraction_is_requested_with_the_manifest_inventory(self) -> None: dist, archive, manifest = self.prepared() seen: dict = {} real = verify.extract_verified_archive
def spy(archive_path, destination, **kwargs): seen.update(kwargs) return real(archive_path, destination, **kwargs)
args = self.args(source_dir=dist, work_dir=self.root / "out") with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_image_identity", return_value={"tag": args.image, "id": "i"}), \ mock.patch.object(verify, "docker_run", side_effect=fake_docker_run), mock.patch.object( verify, "extract_verified_archive", side_effect=spy ): build.run_build(args, verify.Journal(), verify.Cleanup())
inventory = build.closed_inventory(manifest) self.assertEqual(seen["expected"], inventory["expected"]) self.assertEqual(seen["required"], inventory["required"]) self.assertIn(f"{SOURCE_PREFIX}/Cargo.lock", seen["expected"]) self.assertIn(f"{SOURCE_PREFIX}/SOURCE-PROVENANCE.json", seen["required"])
def test_an_archive_above_the_member_bound_is_refused(self) -> None: dist, archive, manifest = self.prepared() with mock.patch.object(verify, "MAX_MEMBERS", 2): with self.assertRaises(verify.HarnessError): verify.extract_verified_archive( archive, self.root / "extracted", expected=build.closed_inventory(manifest)["expected"] )
def test_an_unsafe_member_is_refused_before_anything_is_written(self) -> None: dist = self.root / "dist" dist.mkdir() archive = dist / "bad.tar.gz" with tarfile.open(archive, "w:gz") as handle: add_member(handle, f"{SOURCE_PREFIX}/../../etc/passwd", b"x") destination = self.root / "extracted" with self.assertRaises(verify.HarnessError): verify.extract_verified_archive(archive, destination, expected={}) self.assertEqual(list(destination.iterdir()), []) self.assertFalse((self.root / "etc").exists())
class BuildRunTests(HarnessCase): def test_the_three_steps_run_in_order_and_the_record_binds_the_outputs(self) -> None: dist, archive, manifest = self.prepared() work = self.root / "out" args = self.args(source_dir=dist, work_dir=work) calls, environments, timeouts, mounts_seen = [], [], [], []
def recording(*, image, argv, journal, mount=None, mount_purpose=None, workdir=None, user=None, environment=None, network="none", timeout=None, note=None): calls.append([str(item) for item in argv]) environments.append(dict(environment or {})) timeouts.append(timeout) if mount: mounts_seen.append((str(mount[0]), mount[1], mount[2])) return fake_docker_run( image=image, argv=argv, journal=journal, mount=mount, mount_purpose=mount_purpose, workdir=workdir, user=user, environment=environment, network=network, timeout=timeout, note=note, )
document, code, failures = self.run_mocked(args, side_effect=recording)[0]
self.assertEqual(failures, []) self.assertEqual(code, build.EXIT_OK) self.assertEqual(len(calls), 3) self.assertIn(".cargo/config.toml", " ".join(calls[0])) self.assertEqual(calls[1], build.DESCRIPTOR_ARGV) self.assertNotIn("--replace", calls[1]) self.assertEqual(calls[2], list(verify.BUILD_COMMAND)) self.assertNotIn("-j", calls[2]) self.assertEqual([env.get("CARGO_NET_OFFLINE") for env in environments if "CARGO_NET_OFFLINE" in env], ["true"]) self.assertEqual([env.get("CARGO_BUILD_JOBS") for env in environments if "CARGO_BUILD_JOBS" in env], ["2"]) self.assertEqual(set(timeouts), {args.timeout}) self.assertEqual([step["step"] for step in document["commands"]], ["vendor-config", "build-environment", "build"]) self.assertEqual([step["exit_code"] for step in document["commands"]], [0, 0, 0])
source_root = work / "extracted" / SOURCE_PREFIX self.assertEqual(document["schema"], 1) self.assertEqual(document["record_kind"], "release_source_build") self.assertEqual(document["verdict"], "built") self.assertFalse(document["pack_gate_evidence"]) self.assertEqual(document["build_argv"], list(verify.BUILD_COMMAND)) self.assertEqual(document["binary_dir"], str(source_root / "target" / "release")) self.assertEqual(document["build_env"], str(source_root / build.DESCRIPTOR_HOST_NAME)) self.assertTrue(Path(document["build_env"]).is_file()) self.assertTrue((Path(document["binary_dir"]) / build.BUILD_ENV_NAME).is_file()) self.assertEqual(sorted(document["verified_outputs"]), sorted(verify.BINARY_NAMES)) for item in document["outputs"]: self.assertEqual(item["sha256"], sha256(BINARIES[item["name"]])) self.assertEqual(item["size"], len(BINARIES[item["name"]])) self.assertEqual(document["source_archive"]["name"], archive.name) self.assertEqual(document["source_archive"]["sha256"], sha256(archive.read_bytes())) self.assertEqual(document["source_archive"]["prefix"], SOURCE_PREFIX) self.assertEqual(document["environment"]["build_jobs"], 2) self.assertEqual(document["closure"]["unknown_files"], []) self.assertEqual(document["closure"]["unknown_directories"], []) self.assertEqual(document["closure"]["symlinks"], []) self.assertEqual(set(mounts_seen), {(str(source_root), "/work", "rw")}) self.assertEqual(len(mounts_seen), len(calls))
def test_the_record_is_not_the_pack_gates_evidence(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, _ = self.run_mocked(args)[0] self.assertEqual(code, build.EXIT_OK) self.assertEqual(document["record_kind"], "release_source_build") problems = packaging.validate_rebuild_document( document, expected_archive={"name": archive.name, "sha256": sha256(archive.read_bytes())}, expected_aggregate=manifest["aggregate_sha256"], expected_binaries={name: {"sha256": sha256(c), "size": len(c)} for name, c in BINARIES.items()}, expected_environment={ "base_image_digest": verify.PINNED_BASE_DIGEST, "rust_version": verify.PINNED_RUST_VERSION, "target_triple": verify.PINNED_TARGET_TRIPLE, }, ) self.assertTrue(any("release_rebuild_evidence" in problem for problem in problems))
def test_refuses_a_nonzero_build_and_records_it_honestly(self) -> None: dist, archive, manifest = self.prepared() work = self.root / "out" args = self.args(source_dir=dist, work_dir=work) document, code, failures = self.run_mocked(args, side_effect=docker_outcome(build_exit=101))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertEqual(document["verified_outputs"], "none") self.assertEqual(document["outputs"], []) self.assertEqual(document["failure"]["build_exit_code"], 101) self.assertEqual(document["failure"]["step_exit_codes"]["build"], 101) self.assertEqual([step["exit_code"] for step in document["commands"]][-1], 101) self.assertIn("compile error", document["commands"][-1]["stderr_tail"]) self.assertTrue(any("exited 101" in item for item in failures)) self.assertTrue((work / "extracted").is_dir())
def test_records_missing_outputs_without_claiming_success(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked( args, side_effect=docker_outcome(binary_bytes={"ultimate-tool": BINARIES["ultimate-tool"]}) )[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertEqual(document["verified_outputs"], "none") self.assertEqual(document["failure"]["missing_outputs"], ["ultimate-tool-mcp"]) self.assertTrue(any("ultimate-tool-mcp" in item for item in failures))
def test_an_oversize_binary_output_is_recorded(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") with mock.patch.object(build, "MAX_BINARY_BYTES", 4): document, code, failures = self.run_mocked(args)[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verified_outputs"], "none") self.assertTrue(any("above the" in item and "byte bound" in item for item in failures))
def test_a_timed_out_build_is_recorded_not_swallowed(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked( args, side_effect=docker_outcome(build_error="command timed out after 1800s: cargo build") )[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") build_step = document["commands"][-1] self.assertEqual(build_step["step"], "build") self.assertEqual(build_step["exit_code"], -1) self.assertIn("timed out", build_step["error"]) self.assertTrue(any("could not run" in item for item in failures)) self.assertTrue(args.work_dir.is_dir())
def test_a_failed_descriptor_step_is_recorded_and_the_build_is_not_started(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked(args, side_effect=docker_outcome(descriptor_exit=3))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertEqual([step["step"] for step in document["commands"]], ["vendor-config", "build-environment"]) self.assertIsNone(document["environment"]) self.assertTrue(any("descriptor" in item for item in failures))
def test_an_invalid_json_descriptor_is_recorded(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked(args, side_effect=docker_outcome(descriptor_bytes=b"not json\n"))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertEqual(document["verified_outputs"], "none") self.assertIsNone(document["environment"]) self.assertTrue(any("not valid JSON" in item for item in failures))
def test_a_malformed_descriptor_is_recorded(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") malformed = dict(DESCRIPTOR, base_image="ubuntu:24.04", toolchain=["rustc"], target_triple=None) payload = json.dumps(malformed).encode("utf-8") document, code, failures = self.run_mocked(args, side_effect=docker_outcome(descriptor_bytes=payload))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual([step["step"] for step in document["commands"]], ["vendor-config", "build-environment"]) self.assertTrue(any("base_image is not an object" in item for item in failures)) self.assertTrue(any("toolchain is not an object" in item for item in failures))
def test_a_wrong_observed_environment_is_recorded_not_accepted(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") wrong = dict(DESCRIPTOR, base_image={"reference": verify.PINNED_BASE_REFERENCE, "digest": "sha256:" + "0" * 64, "pinned": True}) document, code, failures = self.run_mocked( args, side_effect=docker_outcome(descriptor_bytes=json.dumps(wrong).encode("utf-8")) )[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual([step["step"] for step in document["commands"]], ["vendor-config", "build-environment"]) self.assertTrue(any("base image digest" in item for item in failures))
def test_a_failed_vendor_config_step_is_recorded_and_stops_the_build(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") document, code, failures = self.run_mocked(args, side_effect=docker_outcome(vendor_exit=1))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual([step["step"] for step in document["commands"]], ["vendor-config"]) self.assertTrue(any("vendor config" in item for item in failures))
def test_a_dangling_build_env_symlink_beside_the_binaries_is_refused(self) -> None: # candidate.exists() is false for a dangling symlink and shutil.copyfile # would follow it, writing a host path outside the extracted tree. dist, archive, manifest = self.prepared() outside = self.root / "outside" / build.BUILD_ENV_NAME args = self.args(source_dir=dist, work_dir=self.root / "out")
def plant(root: Path, target: Path) -> None: (target / build.BUILD_ENV_NAME).symlink_to(outside)
document, code, failures = self.run_mocked(args, side_effect=docker_outcome(post_build=plant))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertIsNone(document["build_env_beside_binaries"]) self.assertTrue(any("refusing to overwrite" in item for item in failures)) self.assertFalse(outside.exists()) self.assertFalse(outside.parent.exists()) link = Path(document["binary_dir"]) / build.BUILD_ENV_NAME self.assertTrue(link.is_symlink()) self.assertEqual(os.readlink(link), str(outside))
def test_a_symlinked_output_directory_is_refused_before_any_host_read(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") outside = self.root / "outside" outside.mkdir() for name, content in BINARIES.items(): (outside / name).write_bytes(content)
def swap(root: Path, target: Path) -> None: for child in target.iterdir(): child.unlink() target.rmdir() target.symlink_to(outside, target_is_directory=True)
document, code, failures = self.run_mocked(args, side_effect=docker_outcome(post_build=swap))[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertEqual(document["verified_outputs"], "none") self.assertEqual(document["outputs"], []) self.assertTrue(any("symlink" in item for item in failures)) self.assertFalse((outside / build.BUILD_ENV_NAME).exists())
def test_a_symlinked_descriptor_leaf_is_refused_not_followed(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") outside = self.root / "outside-descriptor.json" document, code, failures = self.run_mocked( args, side_effect=docker_outcome(descriptor_symlink_to=outside) )[0] self.assertEqual(code, build.EXIT_FAILED) self.assertIsNone(document["environment"]) self.assertIsNone(document["build_env"]) self.assertTrue(any("could not be read" in item for item in failures)) self.assertFalse(outside.exists())
def test_a_copy_failure_is_recorded(self) -> None: dist, archive, manifest = self.prepared() args = self.args(source_dir=dist, work_dir=self.root / "out") with mock.patch.object(build, "write_new_exclusive", side_effect=verify.HarnessError("no space left on device")): document, code, failures = self.run_mocked(args)[0] self.assertEqual(code, build.EXIT_FAILED) self.assertEqual(document["verdict"], "failed") self.assertIsNone(document["build_env_beside_binaries"]) self.assertTrue(any("could not be copied" in item for item in failures)) self.assertTrue(any("no space left" in item for item in failures))
def test_a_source_without_vendored_crates_is_refused_before_compiling(self) -> None: dist, archive, manifest = self.prepared(vendored=False) args = self.args(source_dir=dist, work_dir=self.root / "out") result, _, calls = self.run_mocked(args) self.assertIsInstance(result, verify.HarnessError) self.assertIn("no vendored crate sources", str(result)) self.assertEqual(calls, []) self.assertFalse(args.work_dir.exists())
class IsolationTests(HarnessCase): def test_the_isolation_guard_rejects_a_mount_outside_the_extracted_tree(self) -> None: problems = verify.isolation_problems( checkout_root=ROOT, mounts=[{"source": str(ROOT), "target": "/work"}], forbidden_paths=[ROOT.resolve()], expected_source=ROOT.resolve(), ) self.assertTrue(problems)
def test_a_work_directory_under_the_forbidden_checkout_is_refused_end_to_end(self) -> None: checkout = self.root / "checkout" checkout.mkdir() dist = checkout / "dist" dist.mkdir() write_prepared_archive(dist) args = self.args(source_dir=dist, work_dir=checkout / "out", checkout_root=checkout) with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ): with self.assertRaises(verify.HarnessError) as caught: build.run_build(args, verify.Journal(), verify.Cleanup()) self.assertIn("forbidden path", str(caught.exception))
def test_the_cli_still_defaults_its_checkout_root_to_this_repository(self) -> None: """The production default is unchanged; only the tests name a root.""" parsed = build.parse_arguments( ["--source-dir", str(self.root / "dist"), "--work-dir", str(self.root / "out")] ) self.assertEqual(parsed.checkout_root.resolve(), ROOT.resolve())
def test_the_default_harness_checkout_is_synthetic_and_leaves_the_scratch_usable(self) -> None: """No mocked test may inherit the module's real checkout root.
A real checkout sitting directly under a filesystem root forbids ``/`` itself through the parent rule, so inheriting it would refuse a scratch directory anywhere on the volume. """ args = self.args(source_dir=self.root / "dist", work_dir=self.root / "out") self.assertNotEqual(args.checkout_root, ROOT.resolve()) self.assertNotIn(ROOT.resolve(), args.checkout_root.parents) work = (self.root / "out").resolve() forbidden = build.forbidden_paths(args) for anchor in forbidden: self.assertFalse(work == anchor or anchor in work.parents, anchor) self.assertEqual( verify.isolation_problems( checkout_root=args.checkout_root, mounts=[{"source": str(work), "target": "/work"}], forbidden_paths=forbidden, expected_source=work, ), [], )
def test_a_checkout_whose_parent_is_the_filesystem_root_still_forbids_every_scratch(self) -> None: """The CI layout, pinned so the mistake cannot be papered over.
A checkout at ``/work`` makes ``/`` forbidden, so the guard refuses any scratch directory on that volume. That refusal is the production policy, not the bug the tests work around. """ simulated = Path("/work") forbidden = verify.default_forbidden(simulated) self.assertIn(Path("/").resolve(), [path.resolve() for path in forbidden]) work = (self.root / "out").resolve() problems = verify.isolation_problems( checkout_root=simulated, mounts=[{"source": str(work), "target": "/work"}], forbidden_paths=forbidden, expected_source=work, ) self.assertTrue(any("forbidden path" in item for item in problems), problems)
class RecordBoundTests(HarnessCase): def test_command_tails_are_dropped_when_the_record_is_too_large(self) -> None: document = {"harness": {"commands": [{"argv": ["x"], "stdout_tail": "y" * 4000, "stderr_tail": "z" * 4000}]}} with mock.patch.object(build, "MAX_RECORD_BYTES", 256): build.bound_record(document) command = document["harness"]["commands"][0] self.assertEqual(command["stdout_tail"], "") self.assertEqual(command["stderr_tail"], "") self.assertTrue(any("inspectable" in note for note in document["harness"]["notes"]))
def test_a_small_record_is_left_alone(self) -> None: document = {"harness": {"commands": [{"stdout_tail": "y", "stderr_tail": "z"}]}} build.bound_record(document) self.assertEqual(document["harness"]["commands"][0]["stdout_tail"], "y")
class MainTests(HarnessCase): def test_usage_errors_return_the_usage_code(self) -> None: dist, archive, manifest = self.prepared() base = self.argv(source_dir=dist, work_dir=self.root / "o") self.assertEqual(build.main(base + ["--build-jobs", "0"]), build.EXIT_USAGE) self.assertEqual(build.main(base + ["--build-jobs", "65"]), build.EXIT_USAGE) for value in ("0", "-1", "nan", "inf", "1800.5"): self.assertEqual(build.main(base + ["--timeout", value]), build.EXIT_USAGE, value)
def test_a_relative_work_directory_is_accepted(self) -> None: dist, archive, manifest = self.prepared() previous = os.getcwd() os.chdir(self.root) self.addCleanup(os.chdir, previous) with mock.patch.object(verify, "docker_available", return_value=(False, "no docker")): self.assertEqual(build.main(self.argv(source_dir=dist, work_dir=Path("rel-out"))), build.EXIT_ENVIRONMENT)
def test_missing_docker_is_environment_unavailable(self) -> None: dist, archive, manifest = self.prepared() with mock.patch.object(verify, "docker_available", return_value=(False, "no docker")): self.assertEqual(build.main(self.argv(source_dir=dist, work_dir=self.root / "out")), build.EXIT_ENVIRONMENT)
def test_a_failed_build_returns_nonzero_and_writes_the_record(self) -> None: dist, archive, manifest = self.prepared() out_dir = self.root / "out" removed: list[str] = [] with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_image_identity", return_value={"tag": verify.RELEASE_ENV_IMAGE, "id": "i"}), \ mock.patch.object(verify, "docker_run", side_effect=docker_outcome(build_exit=101)), \ mock.patch.object(verify, "remove_containers_with_label", side_effect=lambda label: removed.append(label)): code = build.main(self.argv(source_dir=dist, work_dir=out_dir)) self.assertEqual(code, build.EXIT_FAILED) record = json.loads((out_dir / build.RECORD_NAME).read_text(encoding="utf-8")) self.assertEqual(record["record_kind"], build.RECORD_KIND) self.assertEqual(record["verdict"], "failed") self.assertEqual(record["verified_outputs"], "none") self.assertEqual(record["failure"]["build_exit_code"], 101) # The record names every step with its real exit code. self.assertEqual(record["failure"]["step_exit_codes"], {"vendor-config": 0, "build-environment": 0, "build": 101}) # Cleanup ran, and only for this tool's own container label. self.assertEqual(removed, [f"{build.CONTAINER_LABEL}.{verify.INVOCATION}"])
def test_a_dangling_build_env_symlink_outside_the_work_dir_is_refused(self) -> None: """A mocked build plants a dangling symlink whose target sits outside the work dir.""" dist, archive, manifest = self.prepared() out_dir = self.root / "out" outside = self.root / "outside" / build.BUILD_ENV_NAME
def plant(root: Path, target: Path) -> None: (target / build.BUILD_ENV_NAME).symlink_to(outside)
with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_image_identity", return_value={"tag": verify.RELEASE_ENV_IMAGE, "id": "i"}), \ mock.patch.object(verify, "docker_run", side_effect=docker_outcome(post_build=plant)), \ mock.patch.object(verify, "remove_containers_with_label", return_value=[]): code = build.main(self.argv(source_dir=dist, work_dir=out_dir)) self.assertEqual(code, build.EXIT_FAILED) record = json.loads((out_dir / build.RECORD_NAME).read_text(encoding="utf-8")) self.assertEqual(record["verdict"], "failed") self.assertEqual(record["verified_outputs"], "none") self.assertIsNone(record["build_env_beside_binaries"]) self.assertTrue(record["closure"]["unknown_files"] == []) # Nothing was created outside the tool's own work directory, and the # symlink the build left behind is untouched. self.assertFalse(outside.exists()) self.assertFalse(outside.parent.exists()) link = out_dir / "extracted" / SOURCE_PREFIX / "target" / "release" / build.BUILD_ENV_NAME self.assertTrue(link.is_symlink()) self.assertEqual(os.readlink(link), str(outside))
def test_a_symlinked_build_output_ancestor_creates_nothing_outside(self) -> None: dist, archive, manifest = self.prepared() out_dir = self.root / "out" outside = self.root / "outside" outside.mkdir()
def swap(root: Path, target: Path) -> None: for child in target.iterdir(): child.unlink() target.rmdir() target.symlink_to(outside, target_is_directory=True)
with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_image_identity", return_value={"tag": verify.RELEASE_ENV_IMAGE, "id": "i"}), \ mock.patch.object(verify, "docker_run", side_effect=docker_outcome(post_build=swap)), \ mock.patch.object(verify, "remove_containers_with_label", return_value=[]): code = build.main(self.argv(source_dir=dist, work_dir=out_dir)) self.assertEqual(code, build.EXIT_FAILED) record = json.loads((out_dir / build.RECORD_NAME).read_text(encoding="utf-8")) self.assertEqual(record["verified_outputs"], "none") self.assertEqual(list(outside.iterdir()), [])
def test_no_record_is_written_for_a_preflight_refusal(self) -> None: dist, archive, manifest = self.prepared(unlisted={"build.rs": b"fn main() {}\n"}) out_dir = self.root / "out" with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ): self.assertEqual(build.main(self.argv(source_dir=dist, work_dir=out_dir)), build.EXIT_FAILED) self.assertFalse((out_dir / build.RECORD_NAME).exists())
def test_the_record_is_never_reused_or_overwritten(self) -> None: dist, archive, manifest = self.prepared() out_dir = self.root / "out" argv = self.argv(source_dir=dist, work_dir=out_dir)
def invoke() -> int: """One fully stubbed invocation: availability, image and containers.""" with mock.patch.object(verify, "docker_available", return_value=(True, "t")), mock.patch.object( verify, "docker_image_present", return_value=True ), mock.patch.object(verify, "docker_run", side_effect=docker_outcome(build_exit=1)): return build.main(argv)
self.assertEqual(invoke(), build.EXIT_FAILED) record = out_dir / build.RECORD_NAME before = record.read_bytes() # The second invocation is stubbed exactly like the first: proving the # record is protected must not need the daemon the first one was denied. # The image identity and the label cleanup come from the harness stubs. self.assertEqual(invoke(), build.EXIT_FAILED) self.assertEqual(record.read_bytes(), before)
def test_a_build_failure_raised_before_the_record_is_written_still_fails(self) -> None: dist, archive, manifest = self.prepared() with mock.patch.object(build, "run_build", side_effect=verify.HarnessError("boomed")), \ mock.patch.object(verify, "docker_available", return_value=(True, "t")): self.assertEqual(build.main(self.argv(source_dir=dist, work_dir=self.root / "out")), build.EXIT_FAILED)
class DockerSeamGuardTests(HarnessCase): """The mocked harness cannot start a container, even when a seam is missed."""
def test_a_real_docker_subprocess_is_refused_not_run(self) -> None: """A missed stub fails the test instead of contacting a daemon.
The refusal is per test: an unrelated command still runs here, so nothing global — ``PATH`` above all — was rewritten to achieve it. """ with self.assertRaises(DockerInvocationForbidden): verify.docker_executable() with self.assertRaises(DockerInvocationForbidden): subprocess.run(["docker", "image", "inspect", verify.RELEASE_ENV_IMAGE], check=False) with self.assertRaises(DockerInvocationForbidden): subprocess.Popen(["docker", "ps", "--all"]) self.assertEqual(subprocess.run(["/bin/sh", "-c", "exit 0"], check=False).returncode, 0)
def test_the_image_identity_and_the_label_cleanup_are_stubbed(self) -> None: """Both seams the tool reaches outside a container are stubbed.""" identity = verify.docker_image_identity(verify.RELEASE_ENV_IMAGE) self.assertEqual(identity["tag"], verify.RELEASE_ENV_IMAGE) self.assertIn("stubbed", identity["id"]) label = verify.container_label() self.assertEqual(verify.remove_containers_with_label(label), []) self.assertEqual(self.image_identities, [verify.RELEASE_ENV_IMAGE]) self.assertEqual(self.container_removals, [label])
def test_an_unstubbed_preflight_seam_fails_instead_of_reaching_the_daemon(self) -> None: """``docker_available`` is deliberately left unmocked in this test.""" dist, _, _ = self.prepared() with self.assertRaises(DockerInvocationForbidden) as caught: build.main(self.argv(source_dir=dist, work_dir=self.root / "out")) self.assertIn("no docker subprocess may start", str(caught.exception))
if __name__ == "__main__": unittest.main()