"""Tests for tools/runtime_source_bundle.py. The bundler sits between the runtime source collector and the release packager, so the properties that must never fail open are: the archive is read only through the shared bounded reader, coverage is required and bound to the collector's *exact tree evidence* (never inferred from an archive being present), a supplied coverage mapping can only strengthen the trusted minimum, every collector obligation is accounted for, notices are byte-bound to their fetched artifacts, and an existing file or a hardlink to an unrelated file is never overwritten. The tests build small synthetic archives with the collector's own ``deterministic_tar_gz`` so the format matches the real producer, and reuse ``tools/package_release.py``'s ``Limits``, ``scan_bytes`` and bounded tar reader. Forgeries are coherent, not raw byte tampering: the internal member manifest is regenerated so a self-consistency check alone would pass. """ from __future__ import annotations import contextlib import gzip import io import json import os import sys import tarfile import tempfile import unittest from dataclasses import replace from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[2] TOOLS = ROOT / "tools" sys.path.insert(0, str(TOOLS)) import collect_runtime_sources as crs # noqa: E402 import package_release as pr # noqa: E402 import runtime_source_bundle as rsb # noqa: E402 PIN = "a" * 40 NOTICE_MEMBER = "notices/pkg/LICENSE" NOTICE_BYTES = b"Apache-2.0 WITH LLVM-exception\n" ARTIFACT = "pkg/spec.json" ARTIFACT_BYTES = b'{"target": "c64"}' def base_members(**extra) -> dict: members = { "pkg/src/lib.rs": b"pub fn f() {}\n", "pkg/src/other.rs": b"// other\n", NOTICE_MEMBER: NOTICE_BYTES, } members.update(extra) return members def write_archive(path: Path, members: dict, *, manifest_text: str | None = None) -> bytes: payload = dict(members) if manifest_text is None: manifest_text = "".join(f"{crs.sha256_hex(payload[n])} {n}\n" for n in sorted(payload)) payload[crs.ARCHIVE_MANIFEST_NAME] = manifest_text.encode("utf-8") blob = crs.deterministic_tar_gz(payload) path.write_bytes(blob) return blob def write_raw_archive(path: Path, entries) -> bytes: buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w", format=tarfile.GNU_FORMAT) as tar: for name, kind, value in entries: info = tarfile.TarInfo(name) info.mtime = 0 info.uid = 0 info.gid = 0 if kind == "file": info.size = len(value) tar.addfile(info, io.BytesIO(value)) elif kind == "sym": info.type = tarfile.SYMTYPE info.linkname = value tar.addfile(info) elif kind == "link": info.type = tarfile.LNKTYPE info.linkname = value tar.addfile(info) blob = gzip.compress(buffer.getvalue(), mtime=0) path.write_bytes(blob) return blob def selection_hashes(members: dict, prefix: str) -> dict: return {name: crs.sha256_hex(payload) for name, payload in members.items() if name.startswith(prefix)} def make_input(input_id: str, prefix: str, required_paths, members: dict, *, min_members=None, dependencies=()): selected = selection_hashes(members, prefix) return { "id": input_id, "kind": "url_archive", "component": f"{input_id} component", "repository": "https://example.invalid/r", "commit": PIN, "license": "Apache-2.0 WITH LLVM-exception", "members": len(selected), "min_members": min_members if min_members is not None else len(selected), "tree_digest": rsb.tree_digest(selected), "required_paths": list(required_paths), "required_paths_verified": len(required_paths), "dependencies": [dict(dep) for dep in dependencies], "submodules": [], } def make_report(archive_path: Path, *, inputs, obligations=None) -> dict: blob = archive_path.read_bytes() members = {} with tarfile.open(fileobj=io.BytesIO(blob), mode="r:*") as tar: for member in tar: if member.isreg() and member.name != crs.ARCHIVE_MANIFEST_NAME: members[member.name] = tar.extractfile(member).read() return { "schema": 1, "record_kind": "release_runtime_sources", "generated_by": "tools/collect_runtime_sources.py", "output_root": "target/release-runtime-sources/collected", "pins": [{"component": "pkg pin", "repository": "https://example.invalid/r", "commit": PIN}], "obligations": obligations if obligations is not None else [ { "component": "pkg runtime", "license": "Apache-2.0 WITH LLVM-exception", "status": "collected", "required_notices": ["pkg/LICENSE"], "missing_notices": [], } ], "images": [ { "id": "pkg-image", "files": [ { "output_name": ARTIFACT, "sha256": crs.sha256_hex(ARTIFACT_BYTES), "size_bytes": len(ARTIFACT_BYTES), "status": "collected", "written": True, } ], "probes": [], } ], "fetches": [ { "id": "pkg-notice-fetch", "repository": "https://example.invalid/r", "commit": PIN, "role": "notice", "files": [ { "output_name": "pkg/LICENSE", "sha256": crs.sha256_hex(NOTICE_BYTES), "size_bytes": len(NOTICE_BYTES), "status": "collected", "written": True, "url": "https://example.invalid/r/LICENSE", } ], } ], "limits": {}, "source_archive": { "name": archive_path.name, "sha256": crs.sha256_hex(blob), "size_bytes": len(blob), "members": len(members) + 1, "source_members": len(members), "tree_digest": rsb.tree_digest({name: crs.sha256_hex(payload) for name, payload in members.items()}), "notices_included": [NOTICE_MEMBER], "written": True, "inputs": list(inputs), }, } def make_coverage( *, reviewed=True, prefix="pkg/", source_input="pkg-input", paths=("src/lib.rs", "src/other.rs"), notices=(NOTICE_MEMBER,), artifacts=(ARTIFACT,), deps=({"name": "compiler_builtins", "requires_source": True, "disposition": "covered", "locked_version": "0.1.160"},), license="Apache-2.0 WITH LLVM-exception", commit=PIN, min_members=2, ): return { "schema": 1, "record_kind": rsb.COVERAGE_RECORD_KIND, "reviewed": reviewed, "maintainer": "test", "components": [ { "key": "pkg", "component": "pkg runtime", "obligation_match": "pkg", "repository": "https://example.invalid/r", "commit": commit, "scope": "linked", "license": license, "license_accepts": [license], "source_input": source_input, "archive_prefix": prefix, "min_members": min_members, "required_paths": list(paths), "required_source_prefixes": [prefix], "required_notices": list(notices), "required_collected_artifacts": list(artifacts), "dependencies": [dict(dep) for dep in deps], } ], } class BundleTestBase(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.collected = self.root / "collected" self.collected.mkdir() self.archive = self.collected / rsb.ARCHIVE_NAME def build(self, members=None, *, deps=(("compiler_builtins", "0.1.160"),), inputs=None): members = members or base_members() write_archive(self.archive, members) if inputs is None: inputs = [ make_input( "pkg-input", "pkg/", ["src/lib.rs", "src/other.rs"], members, dependencies=[ {"crate": crate, "locked_version": version, "provenance": "test"} for crate, version in deps ], ) ] report = make_report(self.archive, inputs=inputs) artifact = self.collected / ARTIFACT artifact.parent.mkdir(parents=True, exist_ok=True) artifact.write_bytes(ARTIFACT_BYTES) return members, report class AuditTests(BundleTestBase): def test_legit_archive_audits_and_binds(self) -> None: members, report = self.build() audit = rsb.audit_archive(self.archive) self.assertEqual(audit["sha256"], report["source_archive"]["sha256"]) self.assertEqual(audit["tree_digest"], report["source_archive"]["tree_digest"]) self.assertEqual(audit["notice_members"], [NOTICE_MEMBER]) self.assertTrue(audit["internal_manifest_verified"]) rsb.bind_report(report, audit) def test_symlink_member_fails_closed(self) -> None: write_raw_archive( self.archive, [(rsb.INTERNAL_MANIFEST, "file", b""), (NOTICE_MEMBER, "sym", "/etc/passwd")], ) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive) def test_hardlink_member_fails_closed(self) -> None: write_raw_archive( self.archive, [ (rsb.INTERNAL_MANIFEST, "file", b""), ("pkg/src/lib.rs", "file", b"x"), ("pkg/alias.rs", "link", "pkg/src/lib.rs"), ], ) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive) def test_traversal_member_fails_closed(self) -> None: write_raw_archive( self.archive, [(rsb.INTERNAL_MANIFEST, "file", b""), ("../escape.txt", "file", b"x")], ) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive) def test_coherent_forgery_of_the_archive_is_caught(self) -> None: # Drop a member and regenerate the internal manifest so the archive is # self-consistent; the pinned tree digest still exposes the change. members, report = self.build() del members["pkg/src/other.rs"] write_archive(self.archive, members) audit = rsb.audit_archive(self.archive) self.assertTrue(audit["internal_manifest_verified"]) with self.assertRaises(rsb.BundleError): rsb.bind_report(report, audit) def test_missing_internal_manifest_fails_closed(self) -> None: members, _ = self.build() kept = {name: payload for name, payload in members.items()} self.archive.write_bytes(crs.deterministic_tar_gz(kept)) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive) def test_large_archive_fails_on_shared_archive_bound(self) -> None: self.build(members=base_members(**{"pkg/src/big.bin": b"x" * 4096})) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive, limits=replace(pr.DEFAULT_LIMITS, max_archive_bytes=64)) def test_large_archive_fails_on_shared_member_bound(self) -> None: self.build(members=base_members(**{"pkg/src/big.bin": b"x" * 4096})) with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive, limits=replace(pr.DEFAULT_LIMITS, max_member_bytes=1024)) def test_archive_fails_on_shared_member_count_bound(self) -> None: self.build() with self.assertRaises(rsb.BundleError): rsb.audit_archive(self.archive, limits=replace(pr.DEFAULT_LIMITS, max_members=1)) class ReaderPathTests(BundleTestBase): def test_extract_members_reads_exact_paths(self) -> None: self.build() found = rsb.extract_members(self.archive, [NOTICE_MEMBER]) self.assertEqual(list(found), [NOTICE_MEMBER]) self.assertEqual(found[NOTICE_MEMBER], NOTICE_BYTES) def test_extract_members_rejects_a_missing_path(self) -> None: self.build() with self.assertRaises(rsb.BundleError): rsb.extract_members(self.archive, ["notices/pkg/NOT-HERE"]) def test_archive_filename_rejects_a_path(self) -> None: with self.assertRaises(rsb.BundleError): rsb.archive_filename({"source_archive": {"name": "../evil.tar.gz"}}) def test_display_path_uses_basename_outside_the_repo(self) -> None: self.assertEqual(rsb.display_path(Path("/tmp/somewhere/x.json"), ROOT), "x.json") class ReportBindingTests(BundleTestBase): def test_future_schema_is_rejected(self) -> None: _, report = self.build() path = self.root / "report.json" path.write_text(json.dumps({**report, "schema": 99})) with self.assertRaises(rsb.BundleError): rsb.load_report(path) def test_wrong_record_kind_is_rejected(self) -> None: _, report = self.build() path = self.root / "report.json" path.write_text(json.dumps({**report, "record_kind": "other"})) with self.assertRaises(rsb.BundleError): rsb.load_report(path) def test_tampered_digest_fails_closed(self) -> None: _, report = self.build() audit = rsb.audit_archive(self.archive) report["source_archive"]["sha256"] = "0" * 64 with self.assertRaises(rsb.BundleError): rsb.bind_report(report, audit) def test_unwritten_report_is_refused(self) -> None: _, report = self.build() audit = rsb.audit_archive(self.archive) report["source_archive"]["written"] = False with self.assertRaises(rsb.BundleError): rsb.bind_report(report, audit) class CoverageTests(BundleTestBase): def evaluate(self, coverage, report): audit = rsb.audit_archive(self.archive) return rsb.evaluate_coverage(coverage, audit, report, self.collected) def test_satisfied_coverage_completes(self) -> None: _, report = self.build() state = self.evaluate(make_coverage(), report) self.assertTrue(state["complete"], state["problems"]) self.assertEqual(state["components"][0]["status"], "satisfied") def test_unreviewed_coverage_does_not_complete(self) -> None: _, report = self.build() state = self.evaluate(make_coverage(reviewed=False), report) self.assertFalse(state["complete"]) def test_component_tree_digest_binding(self) -> None: _, report = self.build() report["source_archive"]["inputs"][0]["tree_digest"] = "0" * 64 state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("pinned tree digest" in p for p in state["components"][0]["missing_source"])) def test_missing_required_source_member_is_a_gap(self) -> None: members = base_members() del members["pkg/src/other.rs"] _, report = self.build(members=members) state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("src/other.rs" in p for p in state["components"][0]["missing_source"])) def test_partial_forgery_dropping_a_required_path_from_the_report_is_a_gap(self) -> None: members, report = self.build() # The report drops a required path and lowers its verified count; the # trusted minimum still requires it. report["source_archive"]["inputs"][0]["required_paths"] = ["src/lib.rs"] report["source_archive"]["inputs"][0]["required_paths_verified"] = 1 state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("does not require 'src/other.rs'" in p for p in state["components"][0]["missing_source"])) def test_below_min_members_is_a_gap(self) -> None: members, report = self.build() report["source_archive"]["inputs"][0]["min_members"] = 99 state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) def test_uncovered_dependency_is_a_gap(self) -> None: _, report = self.build() coverage = make_coverage(deps=({"name": "compiler_builtins", "requires_source": True, "disposition": "unresolved"},)) state = self.evaluate(coverage, report) self.assertFalse(state["complete"]) def test_missing_notice_is_a_gap(self) -> None: _, report = self.build() state = self.evaluate(make_coverage(notices=(NOTICE_MEMBER, "notices/pkg/COPYING")), report) self.assertFalse(state["complete"]) self.assertIn("notices/pkg/COPYING", state["components"][0]["missing_notices"]) def test_external_artifact_hash_mismatch_is_a_gap(self) -> None: _, report = self.build() (self.collected / ARTIFACT).write_bytes(b'{"target": "c64", "tampered": true}') state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("does not match the report's recorded" in p for p in state["components"][0]["missing_artifacts"])) def test_empty_artifact_is_a_gap(self) -> None: _, report = self.build() (self.collected / ARTIFACT).write_bytes(b"") state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) def test_unknown_commit_is_a_problem(self) -> None: _, report = self.build() state = self.evaluate(make_coverage(commit="b" * 40), report) self.assertFalse(state["complete"]) def test_dependency_wrong_version_is_a_gap(self) -> None: _, report = self.build() report["source_archive"]["inputs"][0]["dependencies"] = [ {"crate": "compiler_builtins", "locked_version": "9.9.9", "provenance": "x"} ] state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("is not the required" in p for p in state["components"][0]["missing_source"])) def test_dependency_missing_version_is_a_gap(self) -> None: _, report = self.build() report["source_archive"]["inputs"][0]["dependencies"] = [{"crate": "compiler_builtins", "provenance": "x"}] state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("without a locked version" in p for p in state["components"][0]["missing_source"])) def test_dependency_conflicting_duplicates_is_a_gap(self) -> None: _, report = self.build() report["source_archive"]["inputs"][0]["dependencies"] = [ {"crate": "compiler_builtins", "locked_version": "0.1.160", "provenance": "x"}, {"crate": "compiler_builtins", "locked_version": "0.1.150", "provenance": "y"}, ] state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("conflicting locked versions" in p for p in state["components"][0]["missing_source"])) def test_report_min_members_below_the_floor_is_a_gap(self) -> None: _, report = self.build() report["source_archive"]["inputs"][0]["min_members"] = 1 state = self.evaluate(make_coverage(min_members=2), report) self.assertFalse(state["complete"]) self.assertTrue(any("below the trusted floor" in p for p in state["components"][0]["missing_source"])) def test_artifact_size_missing_is_a_gap(self) -> None: _, report = self.build() del report["images"][0]["files"][0]["size_bytes"] state = self.evaluate(make_coverage(), report) self.assertFalse(state["complete"]) self.assertTrue(any("records no size" in p for p in state["components"][0]["missing_artifacts"])) class TrustedMinimumTests(unittest.TestCase): def test_a_minimal_reviewed_mapping_cannot_weaken_the_minimum(self) -> None: minimal = { "schema": 1, "record_kind": rsb.COVERAGE_RECORD_KIND, "reviewed": True, "components": [ { "key": "rust-mos-runtime-library", "component": "x", "scope": "linked", "license": "Apache-2.0 OR MIT", "source_input": "rust-mos-library", "archive_prefix": "rust-mos/library/", "required_paths": [], "required_notices": [], "required_collected_artifacts": [], "dependencies": [], } ], } with self.assertRaises(rsb.BundleError) as caught: rsb.validate_coverage_against_minimum(minimal) message = str(caught.exception) self.assertIn("drops required paths", message) self.assertIn("omits a required component", message) def test_the_trusted_minimum_strengthens_itself(self) -> None: rsb.validate_coverage_against_minimum(rsb.REQUIRED_COVERAGE) def test_a_schema_mismatch_is_rejected(self) -> None: with self.assertRaises(rsb.BundleError): rsb.validate_coverage_against_minimum({"schema": 2, "record_kind": rsb.COVERAGE_RECORD_KIND, "components": []}) def test_lowering_the_min_members_floor_is_rejected(self) -> None: coverage = json.loads(json.dumps(rsb.REQUIRED_COVERAGE)) for component in coverage["components"]: component["min_members"] = 0 with self.assertRaises(rsb.BundleError) as caught: rsb.validate_coverage_against_minimum(coverage) self.assertIn("below the trusted floor", str(caught.exception)) class NoticeMappingTests(BundleTestBase): def test_notices_are_bound_to_their_fetched_artifacts(self) -> None: _, report = self.build() audit = rsb.audit_archive(self.archive) mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(problems, []) self.assertEqual(mapping[0]["member"], NOTICE_MEMBER) self.assertEqual(mapping[0]["sha256"], crs.sha256_hex(NOTICE_BYTES)) def test_notice_hash_forgery_is_caught(self) -> None: _, report = self.build() report["fetches"][0]["files"][0]["sha256"] = "0" * 64 audit = rsb.audit_archive(self.archive) mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertTrue(problems) self.assertEqual(mapping, []) def test_notice_without_a_fetch_record_is_refused(self) -> None: _, report = self.build() report["fetches"] = [] audit = rsb.audit_archive(self.archive) _, problems = rsb.evaluate_notice_mapping(audit, report) self.assertTrue(problems) def test_notice_size_mismatch_is_a_gap(self) -> None: _, report = self.build() report["fetches"][0]["files"][0]["size_bytes"] = len(NOTICE_BYTES) + 1 audit = rsb.audit_archive(self.archive) mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(mapping, []) self.assertTrue(any("bytes but the fetched artifact records" in p for p in problems)) def test_notice_size_missing_is_a_gap(self) -> None: _, report = self.build() del report["fetches"][0]["files"][0]["size_bytes"] audit = rsb.audit_archive(self.archive) mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(mapping, []) self.assertTrue(any("records no size" in p for p in problems)) def test_zero_size_notice_is_a_gap(self) -> None: report = { "fetches": [ {"files": [{"output_name": "pkg/LICENSE", "sha256": crs.sha256_hex(b""), "size_bytes": 0}]} ] } audit = { "notice_members": [NOTICE_MEMBER], "member_sizes": {NOTICE_MEMBER: 0}, "member_hashes": {NOTICE_MEMBER: crs.sha256_hex(b"")}, } mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(mapping, []) self.assertTrue(any("empty" in p for p in problems)) def test_huge_notice_is_a_gap_in_the_completeness_path(self) -> None: big = b"x" * 64 _, report = self.build(members=base_members(**{NOTICE_MEMBER: big})) report["fetches"][0]["files"][0]["sha256"] = crs.sha256_hex(big) report["fetches"][0]["files"][0]["size_bytes"] = len(big) audit = rsb.audit_archive(self.archive) with mock.patch.object(rsb, "MAX_NOTICE_BYTES", 32): mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(mapping, []) self.assertTrue(any("notice bound" in p for p in problems)) def test_mapping_exports_the_observed_size(self) -> None: _, report = self.build() audit = rsb.audit_archive(self.archive) mapping, problems = rsb.evaluate_notice_mapping(audit, report) self.assertEqual(problems, []) self.assertEqual(mapping[0]["size_bytes"], len(NOTICE_BYTES)) self.assertEqual(mapping[0]["fetched_size_bytes"], len(NOTICE_BYTES)) class LicenseFindingTests(BundleTestBase): def test_mislabel_is_reported(self) -> None: _, report = self.build() report["obligations"] = [{"component": "pkg", "license": "zlib", "status": "collected"}] kinds = {f["kind"] for f in rsb.license_findings(make_coverage(), report)} self.assertIn("license_mismatch", kinds) def test_unaccounted_obligation_is_reported(self) -> None: _, report = self.build() report["obligations"].append({"component": "extra dep", "license": "zlib", "status": "collected"}) kinds = {f["kind"] for f in rsb.license_findings(make_coverage(), report)} self.assertIn("unaccounted_obligation", kinds) def test_matching_label_is_clean(self) -> None: _, report = self.build() self.assertEqual(rsb.license_findings(make_coverage(), report), []) class FactoryTests(BundleTestBase): def test_factory_returns_a_release_not_ready_record(self) -> None: _, report = self.build() record = rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=make_coverage(), created_at="2026-01-01T00:00:00Z", allow_test_seam=True, ) self.assertTrue(record["source_completeness"]["source_complete"]) self.assertFalse(record["source_completeness"]["release_ready"]) self.assertEqual(len(record["notice_mapping"]), 1) self.assertEqual(list(pr.scan_bytes(json.dumps(record, sort_keys=True).encode())), []) def test_factory_marks_an_unaccounted_obligation_incomplete(self) -> None: _, report = self.build() report["obligations"].append({"component": "extra dep", "license": "zlib", "status": "collected"}) record = rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=make_coverage(), allow_test_seam=True, ) self.assertFalse(record["source_completeness"]["source_complete"]) self.assertEqual(record["source_completeness"]["status"], "incomplete_coverage") def test_factory_marks_a_notice_forgery_incomplete(self) -> None: _, report = self.build() report["fetches"][0]["files"][0]["sha256"] = "0" * 64 record = rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=make_coverage(), allow_test_seam=True, ) self.assertFalse(record["source_completeness"]["source_complete"]) def test_factory_enforces_the_floor_on_an_injected_coverage(self) -> None: _, report = self.build() with self.assertRaises(rsb.BundleError): rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=make_coverage(), ) class StagingTests(BundleTestBase): def setUp(self) -> None: super().setUp() self.build() def record(self): _, report = self.build() return rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=make_coverage(), allow_test_seam=True, ) def stage(self, stage: Path): audit = rsb.audit_archive(self.archive) return rsb.stage_bundle( stage, archive_path=self.archive, audit=audit, record=self.record(), fragment=b"fragment\n", notice_payloads={NOTICE_MEMBER: NOTICE_BYTES}, ) def test_stage_writes_source_and_binary_trees(self) -> None: stage = self.root / "stage" result = self.stage(stage) paths = {item["path"] for item in result["files"]} self.assertIn(f"runtime-sources/{rsb.ARCHIVE_NAME}", paths) self.assertIn(f"runtime-sources/{rsb.BUNDLE_RECORD_NAME}", paths) self.assertIn("runtime-sources/NOTICES/pkg/LICENSE", paths) self.assertIn("runtime-notices/pkg/LICENSE", paths) self.assertEqual((stage / f"runtime-sources/{rsb.ARCHIVE_NAME}").read_bytes(), self.archive.read_bytes()) def test_existing_file_is_never_overwritten(self) -> None: stage = self.root / "stage" target = stage / "runtime-sources" / rsb.ARCHIVE_NAME target.parent.mkdir(parents=True) target.write_bytes(b"pre-existing") with self.assertRaises(rsb.BundleError): self.stage(stage) self.assertEqual(target.read_bytes(), b"pre-existing") def test_an_outside_hardlink_is_never_truncated(self) -> None: stage = self.root / "stage" outside = self.root / "outside.txt" outside.write_bytes(b"untouched") target = stage / "runtime-sources" / rsb.ARCHIVE_NAME target.parent.mkdir(parents=True) os.link(outside, target) with self.assertRaises(rsb.BundleError): self.stage(stage) self.assertEqual(outside.read_bytes(), b"untouched") self.assertEqual(target.read_bytes(), b"untouched") def test_force_is_removed(self) -> None: buffer = io.StringIO() with contextlib.redirect_stderr(buffer): code = rsb.main(["--force"]) self.assertEqual(code, rsb.EXIT_USAGE) class PatchTests(BundleTestBase): def manifest(self) -> dict: return { "schema": 1, "inventory": [{"component": "pkg runtime", "status": "pending_external", "collected_paths": []}], "source_paths": {"include": [{"path": "tools/collect_runtime_sources.py"}]}, } def record(self, coverage): _, report = self.build() return rsb.evaluate_bundle_from_archived_evidence( report=report, archive_path=self.archive, collected_dir=self.collected, coverage=coverage, allow_test_seam=True, ) def test_unreviewed_coverage_does_not_flip_collected(self) -> None: patch = rsb.build_inventory_patch(self.manifest(), self.record(make_coverage(reviewed=False)), report_name="report.json") self.assertFalse(patch["ready_to_apply"]) self.assertNotEqual(patch["replacement_entries"]["pkg runtime"]["status"], "collected") def test_reviewed_coverage_flips_collected_but_not_release_ready(self) -> None: patch = rsb.build_inventory_patch(self.manifest(), self.record(make_coverage()), report_name="report.json") self.assertTrue(patch["ready_to_apply"]) entry = patch["replacement_entries"]["pkg runtime"] self.assertEqual(entry["status"], "collected") self.assertFalse(entry["release_ready"]) self.assertIn(f"runtime-sources/{rsb.ARCHIVE_NAME}", entry["collected_paths"]) self.assertEqual(entry["runtime_source_coverage"]["components"][0]["license"], "Apache-2.0 WITH LLVM-exception") def test_patch_adds_the_bundler_to_the_allowlist(self) -> None: patch = rsb.build_inventory_patch(self.manifest(), self.record(make_coverage()), report_name="report.json") self.assertEqual(patch["source_paths_include_addition"]["path"], rsb.SELF_TOOL) class ConcreteInventoryTests(unittest.TestCase): """Read the real tools/release_source_manifest.json, not a fixture.""" def setUp(self) -> None: self.path = ROOT / "tools/release_source_manifest.json" if not self.path.exists(): self.skipTest("the release source manifest is not present") def test_manifest_is_well_formed_and_allowlists_the_collector(self) -> None: manifest = json.loads(self.path.read_text()) self.assertEqual(manifest["schema"], 1) include_paths = [entry.get("path") for entry in manifest["source_paths"]["include"]] self.assertIn(rsb.COLLECTOR_TOOL, include_paths) def test_every_collected_entry_names_paths(self) -> None: manifest = json.loads(self.path.read_text()) for entry in manifest["inventory"]: if entry.get("status") == "collected": self.assertTrue(entry.get("collected_paths"), entry.get("component")) def test_runtime_entries_are_present_for_the_patch(self) -> None: manifest = json.loads(self.path.read_text()) components = [entry.get("component", "") for entry in manifest["inventory"]] for needle in ( "Rust-MOS C64 toolchain image", "MOS sysroot linker scripts and CRT objects", "Rust standard library (core, std) for x86_64-unknown-linux-gnu", ): self.assertTrue(any(needle in component for component in components), needle) class RealCollectorOutputTests(unittest.TestCase): """Read the real collector output when it is present; skip otherwise.""" def setUp(self) -> None: self.report = ROOT / "target/release-runtime-sources/runtime-sources-report.json" self.collected = ROOT / "target/release-runtime-sources/collected" if not self.report.exists(): self.skipTest("the collector report is not present") self.report_data, _ = rsb.load_report(self.report) self.archive = self.collected / rsb.archive_filename(self.report_data) if not self.archive.exists(): self.skipTest("the collected archive is not present") def test_real_archive_binds_its_report(self) -> None: audit = rsb.audit_archive(self.archive) rsb.bind_report(self.report_data, audit) self.assertTrue(audit["internal_manifest_verified"]) self.assertTrue(audit["notice_members"]) def test_shipped_coverage_is_reviewed_and_legally_pending(self) -> None: coverage = rsb.validate_coverage(rsb.REQUIRED_COVERAGE) self.assertTrue(coverage["reviewed"]) self.assertEqual(coverage["legal_review"], "pending") rsb.validate_coverage_against_minimum(coverage) def test_shipped_coverage_matches_the_hardened_archive(self) -> None: audit = rsb.audit_archive(self.archive) coverage = rsb.validate_coverage(rsb.REQUIRED_COVERAGE) state = rsb.evaluate_coverage(coverage, audit, self.report_data, self.collected) gaps = [c for c in state["components"] if c["status"] != "satisfied"] self.assertEqual(gaps, [], state["problems"]) self.assertEqual(rsb.license_findings(coverage, self.report_data), []) self.assertTrue(state["complete"], state["problems"]) def test_real_bundle_never_claims_release_ready(self) -> None: record = rsb.evaluate_bundle_from_archived_evidence( report=self.report_data, archive_path=self.archive, collected_dir=self.collected, coverage=rsb.REQUIRED_COVERAGE, ) self.assertFalse(record["source_completeness"]["release_ready"]) self.assertEqual(len(record["notice_mapping"]), len(record["notices"])) if __name__ == "__main__": unittest.main()