Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
37 kB · 855 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856"""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 onlythrough the shared bounded reader, coverage is required and bound to thecollector's *exact tree evidence* (never inferred from an archive beingpresent), a supplied coverage mapping can only strengthen the trusted minimum,every collector obligation is accounted for, notices are byte-bound to theirfetched artifacts, and an existing file or a hardlink to an unrelated file isnever 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 tarreader. Forgeries are coherent, not raw byte tampering: the internal membermanifest is regenerated so a self-consistency check alone would pass."""
from __future__ import annotations
import contextlibimport gzipimport ioimport jsonimport osimport sysimport tarfileimport tempfileimport unittestfrom dataclasses import replacefrom pathlib import Pathfrom 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: E402import package_release as pr # noqa: E402import runtime_source_bundle as rsb # noqa: E402
PIN = "a" * 40NOTICE_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()