Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
56 kB · 1186 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187"""Tests for tools/collect_runtime_sources.py.
The collector touches two things that must never fail open: the filesystem pathit writes into and the bytes it records. These tests pin the safety propertiesdirectly -- traversal and symlink refusal, the size and count bounds, checksummismatch handling, the recorded artifact shape, and the read-only containerextraction contract -- without requiring a container runtime or a network."""
from __future__ import annotations
import ioimport jsonimport osimport reimport sysimport tarfileimport tempfileimport timeimport unittestfrom 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: E402
EMPTY_SHA256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"COMMIT_RE = re.compile(r"^[0-9a-f]{40}$")
class FakeResponse: """A minimal stand-in for an http.client.HTTPResponse."""
def __init__(self, body: bytes, headers: dict | None = None) -> None: self._body = body self.headers = headers if headers is not None else {} self._position = 0
def read(self, amount: int = -1) -> bytes: if amount is None or amount < 0: chunk = self._body[self._position:] self._position = len(self._body) return chunk chunk = self._body[self._position:self._position + amount] self._position += len(chunk) return chunk
def __enter__(self) -> "FakeResponse": return self
def __exit__(self, *_exc) -> bool: return False
def opener_for(body: bytes, headers: dict | None = None): def _open(_request, _timeout): return FakeResponse(body, headers) return _open
def minimal_spec(**overrides) -> dict: spec = { "schema": 1, "record_kind": crs.INPUT_RECORD_KIND, "images": [], "fetches": [], "obligations": [], } spec.update(overrides) return spec
class CollectorTestCase(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.root = Path(self._tmp.name) self.output = self.root / "out"
def make_collector(self, **kwargs) -> crs.Collector: options = dict( output_root=self.output, max_file_bytes=1024, max_total_bytes=4096, max_files=8, http_timeout=1.0, ) options.update(kwargs) return crs.Collector(**options)
# --------------------------------------------------------------------------# malformed input# --------------------------------------------------------------------------
class MalformedInputTests(CollectorTestCase): def test_rejects_missing_record_kind(self) -> None: collector = self.make_collector() with self.assertRaisesRegex(crs.CollectError, "record_kind"): collector.run({"schema": 1})
def test_rejects_wrong_schema(self) -> None: collector = self.make_collector() with self.assertRaisesRegex(crs.CollectError, "schema"): collector.run(minimal_spec(schema=2))
def test_rejects_non_object_action(self) -> None: collector = self.make_collector() with self.assertRaisesRegex(crs.CollectError, "must be an object"): collector.run(minimal_spec(images=["not-an-object"]))
def test_rejects_image_without_reference(self) -> None: collector = self.make_collector() with self.assertRaisesRegex(crs.CollectError, "reference"): collector.run(minimal_spec(images=[{"id": "x"}]))
def test_rejects_relative_container_path(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:0", image_labels_fn=lambda _r: {}, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "extract": [{"container_path": "etc/passwd", "output_name": "x"}], }]) with self.assertRaisesRegex(crs.CollectError, "absolute"): collector.run(spec)
def test_rejects_probe_with_empty_argv(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:0", image_labels_fn=lambda _r: {}, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "probe": [{"output_name": "p.txt", "argv": []}], }]) with self.assertRaisesRegex(crs.CollectError, "argv"): collector.run(spec)
def test_rejects_image_id_drift(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:aaaaaaaa", image_labels_fn=lambda _r: {}, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "expected_image_id": "sha256:bbbbbbbb", }]) with self.assertRaisesRegex(crs.CollectError, "expected"): collector.run(spec)
def test_rejects_label_drift(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {"org.opencontainers.image.revision": "deadbeef"}, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "expect_labels": {"org.opencontainers.image.revision": "cafebabe"}, }]) with self.assertRaisesRegex(crs.CollectError, "label"): collector.run(spec)
def test_main_returns_usage_error_for_malformed_input(self) -> None: bad = self.root / "bad.json" bad.write_text("{not json", encoding="utf-8") code = crs.main(["--input", str(bad), "--output", str(self.output), "--allow-non-target-output", "--dry-run"]) self.assertEqual(code, 2)
# --------------------------------------------------------------------------# checksum parsing and retrieval shape# --------------------------------------------------------------------------
class ChecksumTests(CollectorTestCase): def test_parse_rejects_malformed_digests(self) -> None: for bad in ["", "abc", "Z" * 64, "A" * 64, "a" * 63, "a" * 65, None, 7]: with self.assertRaises(crs.CollectError, msg=f"{bad!r} should be rejected"): crs.parse_sha256(bad, "digest")
def test_parse_accepts_lowercase_hex(self) -> None: self.assertEqual(crs.parse_sha256(EMPTY_SHA256, "digest"), EMPTY_SHA256)
def test_artifact_records_carry_path_hash_and_size(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: b"payload") report = collector.run(minimal_spec(fetches=[{ "id": "group", "files": [{"url": "https://example.invalid/x", "output_name": "ns/file.txt"}], }])) record = report["fetches"][0]["files"][0] self.assertEqual(set(record) >= {"output_name", "sha256", "size_bytes"}, True) self.assertEqual(record["output_name"], "ns/file.txt") self.assertEqual(record["size_bytes"], len(b"payload")) self.assertEqual(record["sha256"], crs.sha256_hex(b"payload")) # the recorded hash must match the bytes actually on disk self.assertEqual(crs.sha256_file(self.output / "ns/file.txt"), record["sha256"])
def test_matching_expected_checksum_is_accepted(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: b"payload") report = collector.run(minimal_spec(fetches=[{ "id": "group", "files": [{ "url": "https://example.invalid/x", "output_name": "ns/file.txt", "sha256": crs.sha256_hex(b"payload"), }], }])) self.assertEqual(report["totals"]["files"], 1)
def test_mismatched_expected_checksum_fails_closed(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: b"payload") with self.assertRaisesRegex(crs.CollectError, "checksum mismatch"): collector.run(minimal_spec(fetches=[{ "id": "group", "files": [{ "url": "https://example.invalid/x", "output_name": "ns/file.txt", "sha256": EMPTY_SHA256, }], }])) self.assertFalse((self.output / "ns/file.txt").exists())
def test_image_extract_checksum_is_verified(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=lambda _r, _p: b"crt0", ) good = minimal_spec(images=[{ "id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/crt0.o", "output_name": "x/crt0.o", "sha256": crs.sha256_hex(b"crt0")}], }]) self.assertEqual(collector.run(good)["totals"]["files"], 1)
collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=lambda _r, _p: b"crt0", ) bad = minimal_spec(images=[{ "id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/crt0.o", "output_name": "y/crt0.o", "sha256": EMPTY_SHA256}], }]) with self.assertRaisesRegex(crs.CollectError, "checksum mismatch"): collector.run(bad)
# --------------------------------------------------------------------------# fetch bounds# --------------------------------------------------------------------------
class FetchTests(CollectorTestCase): def test_rejects_non_http_schemes(self) -> None: for url in ["file:///etc/passwd", "ftp://example.invalid/x", "gopher://x/y", "data:text/plain,hi"]: with self.assertRaises(crs.CollectError, msg=url): crs.fetch_bytes(url, max_bytes=1024, timeout=1.0)
def test_rejects_url_without_host(self) -> None: with self.assertRaisesRegex(crs.CollectError, "host"): crs.fetch_bytes("http:///nothing", max_bytes=1024, timeout=1.0)
def test_enforces_stream_bound_without_content_length(self) -> None: opener = opener_for(b"x" * 4096) with self.assertRaisesRegex(crs.CollectError, "max_file_bytes"): crs.fetch_bytes("https://example.invalid/big", max_bytes=1024, timeout=1.0, opener=opener)
def test_enforces_declared_content_length(self) -> None: opener = opener_for(b"x" * 10, headers={"Content-Length": "999999"}) with self.assertRaisesRegex(crs.CollectError, "max_file_bytes"): crs.fetch_bytes("https://example.invalid/big", max_bytes=1024, timeout=1.0, opener=opener)
def test_rejects_malformed_content_length(self) -> None: opener = opener_for(b"x", headers={"Content-Length": "not-a-number"}) with self.assertRaisesRegex(crs.CollectError, "malformed Content-Length"): crs.fetch_bytes("https://example.invalid/x", max_bytes=1024, timeout=1.0, opener=opener)
def test_accepts_body_within_bound(self) -> None: opener = opener_for(b"hello", headers={"Content-Length": "5"}) self.assertEqual( crs.fetch_bytes("https://example.invalid/x", max_bytes=1024, timeout=1.0, opener=opener), b"hello", )
def test_transport_error_is_reported_plainly(self) -> None: import urllib.error
def boom(_request, _timeout): raise urllib.error.URLError("name or service not known")
with self.assertRaisesRegex(crs.CollectError, "cannot fetch"): crs.fetch_bytes("https://example.invalid/x", max_bytes=1024, timeout=1.0, opener=boom)
# --------------------------------------------------------------------------# path safety# --------------------------------------------------------------------------
class PathSafetyTests(CollectorTestCase): def test_rejects_traversal_and_absolute_names(self) -> None: for name in ["../escape", "a/../../b", "/etc/passwd", "", ".", "a/./b", "a//b", "x\x00y"]: with self.assertRaises(crs.CollectError, msg=f"{name!r} should be rejected"): crs.relative_output_name(name)
def test_accepts_nested_relative_name(self) -> None: self.assertEqual(crs.relative_output_name("ns/sub/file.txt"), "ns/sub/file.txt")
def test_refuses_symlinked_destination(self) -> None: self.output.mkdir(parents=True) target = self.root / "target.txt" target.write_text("original", encoding="utf-8") (self.output / "link.txt").symlink_to(target) with self.assertRaisesRegex(crs.CollectError, "symlink"): crs.write_artifact(self.output, "link.txt", b"new", force=True, max_file_bytes=1024) self.assertEqual(target.read_text(encoding="utf-8"), "original")
def test_refuses_symlinked_parent_directory(self) -> None: self.output.mkdir(parents=True) outside = self.root / "outside" outside.mkdir() (self.output / "sub").symlink_to(outside) with self.assertRaisesRegex(crs.CollectError, "symlink"): crs.write_artifact(self.output, "sub/file.txt", b"new", force=True, max_file_bytes=1024) self.assertFalse((outside / "file.txt").exists())
def test_refuses_to_overwrite_without_force(self) -> None: self.output.mkdir(parents=True) crs.write_artifact(self.output, "a.txt", b"first", force=False, max_file_bytes=1024) with self.assertRaisesRegex(crs.CollectError, "refusing to overwrite"): crs.write_artifact(self.output, "a.txt", b"second", force=False, max_file_bytes=1024) self.assertEqual((self.output / "a.txt").read_bytes(), b"first")
def test_force_overwrites(self) -> None: self.output.mkdir(parents=True) crs.write_artifact(self.output, "a.txt", b"first", force=False, max_file_bytes=1024) crs.write_artifact(self.output, "a.txt", b"second", force=True, max_file_bytes=1024) self.assertEqual((self.output / "a.txt").read_bytes(), b"second")
def test_output_root_outside_target_is_refused(self) -> None: with self.assertRaisesRegex(crs.CollectError, "target"): crs.check_output_root(self.root / "not-target", allow_non_target=False)
def test_output_root_override_is_honoured(self) -> None: resolved = crs.check_output_root(self.root / "anywhere", allow_non_target=True) self.assertEqual(resolved, Path(os.path.abspath(self.root / "anywhere")))
def test_dry_run_writes_nothing(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: b"payload", dry_run=True) report = collector.run(minimal_spec(fetches=[{ "id": "group", "files": [{"url": "https://example.invalid/x", "output_name": "ns/file.txt"}], }])) self.assertFalse(self.output.exists()) self.assertFalse(report["fetches"][0]["files"][0]["written"]) self.assertEqual(report["fetches"][0]["files"][0]["sha256"], crs.sha256_hex(b"payload"))
# --------------------------------------------------------------------------# limits# --------------------------------------------------------------------------
class LimitTests(CollectorTestCase): def test_single_artifact_over_file_bound_fails(self) -> None: self.output.mkdir(parents=True) with self.assertRaisesRegex(crs.CollectError, "max_file_bytes"): crs.write_artifact(self.output, "big.bin", b"x" * 2048, force=False, max_file_bytes=1024) self.assertFalse((self.output / "big.bin").exists())
def test_total_byte_bound_fails(self) -> None: collector = self.make_collector(max_total_bytes=10, fetch_fn=lambda url, **_: b"x" * 8) spec = minimal_spec(fetches=[{ "id": "group", "files": [ {"url": "https://example.invalid/a", "output_name": "ns/a"}, {"url": "https://example.invalid/b", "output_name": "ns/b"}, ], }]) with self.assertRaisesRegex(crs.CollectError, "max_total_bytes"): collector.run(spec)
def test_file_count_bound_fails(self) -> None: collector = self.make_collector(max_files=1, fetch_fn=lambda url, **_: b"x") spec = minimal_spec(fetches=[{ "id": "group", "files": [ {"url": "https://example.invalid/a", "output_name": "ns/a"}, {"url": "https://example.invalid/b", "output_name": "ns/b"}, ], }]) with self.assertRaisesRegex(crs.CollectError, "max_files"): collector.run(spec)
def test_resolve_limits_prefers_cli_then_input_then_default(self) -> None: parser = crs.build_parser() args = parser.parse_args([]) self.assertEqual(crs.resolve_limits(args, {})["max_file_bytes"], crs.BUILTIN_LIMITS["max_file_bytes"]) self.assertEqual(crs.resolve_limits(args, {"limits": {"max_file_bytes": 5}})["max_file_bytes"], 5) args = parser.parse_args(["--max-file-bytes", "7"]) self.assertEqual(crs.resolve_limits(args, {"limits": {"max_file_bytes": 5}})["max_file_bytes"], 7)
def test_resolve_limits_rejects_non_positive(self) -> None: parser = crs.build_parser() args = parser.parse_args([]) with self.assertRaisesRegex(crs.CollectError, "positive"): crs.resolve_limits(args, {"limits": {"max_file_bytes": 0}})
# --------------------------------------------------------------------------# read-only container extraction# --------------------------------------------------------------------------
class ImageExtractionTests(CollectorTestCase): def test_symlink_inside_image_is_recorded_absent_not_written(self) -> None: def reader(_reference, container_path): raise crs.CollectError(f"{container_path} in example:1 is a symlink")
collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=reader, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/link.o", "output_name": "x/link.o"}], }]) report = collector.run(spec) entry = report["images"][0]["files"][0] self.assertEqual(entry["status"], crs.STATUS_ABSENT) self.assertNotIn("sha256", entry) self.assertFalse((self.output / "x/link.o").exists())
def test_missing_image_file_is_recorded_absent(self) -> None: def reader(_reference, container_path): raise crs.CollectError(f"{container_path} in example:1 does not exist")
collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=reader, ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/missing.o", "output_name": "x/missing.o"}], }]) self.assertEqual(collector.run(spec)["images"][0]["files"][0]["status"], crs.STATUS_ABSENT)
def test_probe_output_and_argv_are_recorded(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_probe_fn=lambda _r, _argv: b"rustc 1.94.0-dev\n", ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "probe": [{"output_name": "probe/rustc.txt", "argv": ["/bin/bash", "-c", "rustc -vV"]}], }]) record = collector.run(spec)["images"][0]["probes"][0] self.assertEqual(record["argv"], ["/bin/bash", "-c", "rustc -vV"]) self.assertEqual(record["sha256"], crs.sha256_hex(b"rustc 1.94.0-dev\n"))
def test_empty_output_records_the_empty_digest(self) -> None: # An empty probe result (for example "no license files in this image") # is evidence, not an error, and must carry the empty-string digest. collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_probe_fn=lambda _r, _argv: b"", ) spec = minimal_spec(images=[{ "id": "i", "reference": "example:1", "probe": [{"output_name": "probe/license-files.txt", "argv": ["/bin/sh", "-c", "true"]}], }]) record = collector.run(spec)["images"][0]["probes"][0] self.assertEqual(record["sha256"], EMPTY_SHA256) self.assertEqual(record["size_bytes"], 0)
def test_container_commands_are_read_only_and_offline(self) -> None: seen: list = []
def fake_run(argv, **_kwargs): seen.append(list(argv)) return mock.Mock(returncode=0, stdout=b"data", stderr=b"")
with mock.patch.object(crs, "_run", side_effect=fake_run): self.assertEqual(crs.docker_read_file("example:1", "/lib/crt0.o", max_bytes=1024), b"data") crs.docker_probe("example:1", ["/bin/bash", "-c", "rustc -vV"])
for argv in seen: self.assertEqual(argv[:2], ["docker", "run"]) self.assertIn("--rm", argv) self.assertIn("--network", argv) self.assertIn("none", argv) self.assertNotIn("--privileged", argv) self.assertNotIn("-v", argv)
# --------------------------------------------------------------------------# provenance / pins# --------------------------------------------------------------------------
class ProvenanceTests(CollectorTestCase): def test_obligations_report_collected_paths(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: b"license text") spec = minimal_spec( fetches=[{ "id": "group", "files": [{"url": "https://example.invalid/L", "output_name": "ns/LICENSE"}], }], obligations=[{"id": "o", "component": "component", "license": "MIT", "required_sources": [], "required_notices": ["ns/LICENSE"]}], ) report = collector.run(spec) self.assertEqual(report["obligations"][0]["status"], crs.STATUS_COLLECTED) self.assertEqual(report["obligations"][0]["required_notices"], ["ns/LICENSE"]) self.assertEqual(report["obligations"][0]["missing_notices"], [])
def test_obligations_incomplete_when_nothing_collected(self) -> None: collector = self.make_collector() spec = minimal_spec(obligations=[{"id": "o", "component": "component", "required_sources": ["ns/src/x.c"]}]) report = collector.run(spec) self.assertEqual(report["obligations"][0]["status"], crs.STATUS_INCOMPLETE) self.assertEqual(report["obligations"][0]["missing_sources"], ["ns/src/x.c"])
def test_resolve_github_commit_returns_full_sha(self) -> None: payload = json.dumps({ "sha": "4aaa40e160fcabfcc5b0c30740a518084f00ab5d", "commit": {"committer": {"date": "2026-03-18T08:03:50Z"}}, }).encode() result = crs.resolve_github_commit( "https://github.com/koalabuttz/rust-mos", "4aaa40e16", opener=opener_for(payload, {"Content-Length": str(len(payload))}), ) self.assertEqual(result["resolved_commit"], "4aaa40e160fcabfcc5b0c30740a518084f00ab5d") self.assertEqual(result["repository"], "https://github.com/koalabuttz/rust-mos")
def test_resolve_github_commit_rejects_non_repository_url(self) -> None: with self.assertRaisesRegex(crs.CollectError, "slug"): crs.resolve_github_commit("https://example.invalid/x", "abc", opener=opener_for(b"{}"))
# --------------------------------------------------------------------------# the local input spec, when present# --------------------------------------------------------------------------
class LocalInputSpecTests(unittest.TestCase): spec_path = ROOT / "target/release-runtime-sources/runtime-sources-input.json"
def load(self) -> dict: if not self.spec_path.exists(): self.skipTest("local collection input is not present") return json.loads(self.spec_path.read_text(encoding="utf-8"))
def test_spec_is_well_formed(self) -> None: spec = self.load() self.assertEqual(spec["record_kind"], crs.INPUT_RECORD_KIND) self.assertEqual(spec["schema"], crs.SCHEMA) self.assertTrue(spec["pins"], "at least one lineage pin is required")
def test_every_pin_has_an_exact_commit(self) -> None: for pin in self.load()["pins"]: self.assertRegex(pin["commit"], COMMIT_RE, msg=f"{pin.get('component')} pin") self.assertTrue(pin["repository"].startswith("https://github.com/")) self.assertIn("resolved_from", pin)
def test_every_fetch_url_is_https_and_commit_addressed(self) -> None: for group in self.load()["fetches"]: self.assertRegex(group["commit"], COMMIT_RE) for entry in group["files"]: self.assertTrue(entry["url"].startswith("https://raw.githubusercontent.com/")) self.assertIn(group["commit"], entry["url"]) if "sha256" in entry: self.assertRegex(entry["sha256"], crs.SHA256_RE)
def test_expected_sha256_values_are_well_formed(self) -> None: spec = self.load() for image in spec["images"]: for entry in image.get("extract", []): if "sha256" in entry: self.assertRegex(entry["sha256"], crs.SHA256_RE)
def tar_bytes(members: dict, *, prefix: str = "", links: dict | None = None) -> bytes: """Build a tar in memory for archive tests."""
buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w") as tar: for name, data in members.items(): info = tarfile.TarInfo(prefix + name) info.size = len(data) info.mtime = 0 info.mode = 0o644 info.type = tarfile.REGTYPE tar.addfile(info, io.BytesIO(data)) for name, target in (links or {}).items(): info = tarfile.TarInfo(prefix + name) info.type = tarfile.SYMTYPE info.linkname = target info.mtime = 0 tar.addfile(info) return buffer.getvalue()
def read_archive(blob: bytes) -> dict: with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tar: out = {} for info in tar: if info.isfile(): out[info.name] = tar.extractfile(info).read() return out
# --------------------------------------------------------------------------# archive member handling# --------------------------------------------------------------------------
class ArchiveMemberTests(unittest.TestCase): def test_safe_member_name_normalises_and_rejects(self) -> None: self.assertEqual(crs.safe_member_name("./a/b.txt"), "a/b.txt") self.assertEqual(crs.safe_member_name("a//b"), "a/b") for bad in ["/etc/passwd", "../escape", "a/../../b", "", ".", "x\x00y"]: self.assertIsNone(crs.safe_member_name(bad))
def test_extract_skips_symlinks_and_directories(self) -> None: blob = tar_bytes({"a.txt": b"one", "d/b.txt": b"two"}, links={"link": "/etc/passwd"}) members, skipped = crs.extract_archive_members( blob, strip_prefix=None, include=None, max_members=10, max_bytes=1024 ) self.assertEqual(set(members), {"a.txt", "d/b.txt"}) self.assertIn("link", skipped)
def test_extract_applies_strip_prefix_and_include(self) -> None: blob = tar_bytes( {"root-abc/mos-platform/crt0.S": b"crt", "root-abc/README.md": b"readme"}, prefix="", ) members, _ = crs.extract_archive_members( blob, strip_prefix="root-abc/", include=["mos-platform/"], max_members=10, max_bytes=1024, ) self.assertEqual(members, {"mos-platform/crt0.S": b"crt"})
def test_extract_enforces_member_and_byte_bounds(self) -> None: blob = tar_bytes({"a": b"1", "b": b"2"}) with self.assertRaisesRegex(crs.CollectError, "max_source_files"): crs.extract_archive_members(blob, strip_prefix=None, include=None, max_members=1, max_bytes=1024) with self.assertRaisesRegex(crs.CollectError, "max_source_bytes"): crs.extract_archive_members(blob, strip_prefix=None, include=None, max_members=10, max_bytes=1)
def test_extract_rejects_a_malformed_archive(self) -> None: with self.assertRaisesRegex(crs.CollectError, "cannot read the downloaded archive"): crs.extract_archive_members(b"not a tar", strip_prefix=None, include=None, max_members=10, max_bytes=1024)
def test_deterministic_archive_is_reproducible(self) -> None: members = {"b.txt": b"two", "a.txt": b"one"} first = crs.deterministic_tar_gz(members) second = crs.deterministic_tar_gz({"a.txt": b"one", "b.txt": b"two"}) self.assertEqual(first, second) self.assertEqual(crs.sha256_hex(first), crs.sha256_hex(second)) self.assertEqual(read_archive(first), members)
def test_tree_digest_is_order_independent(self) -> None: self.assertEqual( crs.tree_digest({"a": b"1", "b": b"2"}), crs.tree_digest({"b": b"2", "a": b"1"}), ) self.assertNotEqual(crs.tree_digest({"a": b"1"}), crs.tree_digest({"a": b"2"}))
def test_notice_path_detection(self) -> None: for name in ["LICENSE", "a/COPYING", "x/NOTICE.txt", "y/Copyright", "AUTHORS"]: self.assertTrue(crs.is_notice_path(name), name) self.assertFalse(crs.is_notice_path("src/main.rs"))
# --------------------------------------------------------------------------# source archive assembly# --------------------------------------------------------------------------
class SourceArchiveTests(CollectorTestCase): def make_image_collector(self, tar_fn, **kwargs) -> crs.Collector: return self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_tar_fn=tar_fn, **kwargs, )
def base_images(self) -> list: return [{"id": "tool", "reference": "example:1"}]
def test_image_tree_members_are_archived(self) -> None: blob = tar_bytes({"library/core/lib.rs": b"core", "library/Cargo.lock": b"lock"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec( images=self.base_images(), source_archive={"name": "src.tar.gz", "inputs": [{ "id": "forklib", "kind": "image_tree", "image": "tool", "container_path": "/usr/local/rust-mos/lib/rustlib/src/rust/library", "arcname": "rust-mos/library", "commit": "4aaa40e160fcabfcc5b0c30740a518084f00ab5d", }]}, ) archive = collector.run(spec)["source_archive"] self.assertEqual(archive["source_members"], 2) self.assertGreater(archive["members"], archive["source_members"]) # manifest is added self.assertTrue(archive["written"]) payload = (self.output / "src.tar.gz").read_bytes() self.assertEqual(crs.sha256_file(self.output / "src.tar.gz"), archive["sha256"]) members = read_archive(payload) self.assertEqual(members["rust-mos/library/library/core/lib.rs"], b"core") self.assertIn(crs.ARCHIVE_MANIFEST_NAME, members) manifest = members[crs.ARCHIVE_MANIFEST_NAME].decode() self.assertIn("rust-mos/library/library/core/lib.rs", manifest)
def test_url_archive_checksum_is_verified(self) -> None: blob = tar_bytes({"root-abc/LICENSE": b"lic", "root-abc/src/x.c": b"src"}) good = minimal_spec(source_archive={"name": "s.tar.gz", "inputs": [{ "id": "sdk", "kind": "url_archive", "url": "https://example.invalid/sdk.tar.gz", "sha256": crs.sha256_hex(blob), "strip_prefix": "root-abc/", "include": ["src/"], "arcname": "sdk", }]}) collector = self.make_collector(fetch_fn=lambda url, **_: blob) archive = collector.run(good)["source_archive"] self.assertEqual(archive["inputs"][0]["members"], 1)
bad = minimal_spec(source_archive={"name": "s.tar.gz", "inputs": [{ "id": "sdk", "kind": "url_archive", "url": "https://example.invalid/sdk.tar.gz", "sha256": EMPTY_SHA256, }]}) collector = self.make_collector(fetch_fn=lambda url, **_: blob) with self.assertRaisesRegex(crs.CollectError, "checksum mismatch"): collector.run(bad)
def test_notices_are_embedded(self) -> None: blob = tar_bytes({"src/x.c": b"src"})
def fetch(url, **_kwargs): return blob if url.endswith("/t") else b"license text"
collector = self.make_collector(fetch_fn=fetch) spec = minimal_spec( fetches=[{"id": "g", "files": [ {"url": "https://example.invalid/L", "output_name": "fork/LICENSE-MIT"}, {"url": "https://example.invalid/R", "output_name": "fork/README.md"}, ]}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "sdk", "kind": "url_archive", "url": "https://example.invalid/t", "arcname": "sdk", }]}, ) archive = collector.run(spec)["source_archive"] self.assertEqual(archive["notices_included"], ["notices/fork/LICENSE-MIT"]) self.assertIn("notices/fork/LICENSE-MIT", read_archive((self.output / "s.tar.gz").read_bytes()))
def test_unknown_image_and_unknown_kind_are_rejected(self) -> None: collector = self.make_image_collector(lambda _r, _p: tar_bytes({"a": b"1"})) with self.assertRaisesRegex(crs.CollectError, "unknown image"): collector.run(minimal_spec( images=self.base_images(), source_archive={"name": "s.tar.gz", "inputs": [ {"id": "x", "kind": "image_tree", "image": "nope", "container_path": "/a"}]}, )) collector = self.make_collector(fetch_fn=lambda url, **_: b"") with self.assertRaisesRegex(crs.CollectError, "unknown kind"): collector.run(minimal_spec(source_archive={"name": "s.tar.gz", "inputs": [ {"id": "x", "kind": "carrier-pigeon"}]}))
def test_empty_archive_is_rejected(self) -> None: collector = self.make_collector(fetch_fn=lambda url, **_: tar_bytes({})) with self.assertRaisesRegex(crs.CollectError, "no members"): collector.run(minimal_spec(source_archive={"name": "s.tar.gz", "inputs": [ {"id": "x", "kind": "url_archive", "url": "https://example.invalid/e"}]}))
def test_archive_size_bound_is_enforced(self) -> None: blob = tar_bytes({"src/x": os.urandom(8192)}) collector = self.make_collector(fetch_fn=lambda url, **_: blob) spec = minimal_spec(source_archive={ "name": "s.tar.gz", "max_archive_bytes": 1024, "inputs": [{"id": "sdk", "kind": "url_archive", "url": "https://example.invalid/t", "arcname": "sdk"}], }) with self.assertRaisesRegex(crs.CollectError, "max_archive_bytes"): collector.run(spec) self.assertFalse((self.output / "s.tar.gz").exists())
def test_dry_run_does_not_write_the_archive(self) -> None: blob = tar_bytes({"library/core/lib.rs": b"core"}) collector = self.make_image_collector(lambda _r, _p: blob, dry_run=True) spec = minimal_spec( images=self.base_images(), source_archive={"name": "src.tar.gz", "inputs": [{ "id": "forklib", "kind": "image_tree", "image": "tool", "container_path": "/x", "arcname": "rust-mos/library"}]}, ) archive = collector.run(spec)["source_archive"] self.assertFalse(archive["written"]) self.assertFalse((self.output / "src.tar.gz").exists())
def test_duplicate_members_are_counted_not_silently_overwritten(self) -> None: blob = tar_bytes({"library/Cargo.lock": b"one"}) other = tar_bytes({"library/Cargo.lock": b"two"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec( images=self.base_images(), source_archive={"name": "s.tar.gz", "inputs": [ {"id": "a", "kind": "image_tree", "image": "tool", "container_path": "/a", "arcname": "same"}, {"id": "b", "kind": "image_tree", "image": "tool", "container_path": "/b", "arcname": "same"}, ]}, ) collector.image_tar_fn = lambda reference, path: blob if path == "/a" else other archive = collector.run(spec)["source_archive"] self.assertEqual(archive["inputs"][0]["members"], 1) self.assertEqual(archive["inputs"][1]["members"], 0) self.assertEqual(archive["inputs"][1]["collisions_skipped"], 1) self.assertEqual(read_archive((self.output / "s.tar.gz").read_bytes())["same/library/Cargo.lock"], b"one")
class HardeningTests(CollectorTestCase): """Regression tests for the review findings: fail closed, bounded, portable."""
def make_image_collector(self, tar_fn, **kwargs) -> crs.Collector: return self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_tar_fn=tar_fn, **kwargs, )
# -- fetch contract ----------------------------------------------------
def test_fetch_rejects_embedded_credentials(self) -> None: with self.assertRaisesRegex(crs.CollectError, "credentials"): crs.fetch_bytes("https://user:pw@example.invalid/x", max_bytes=1024, timeout=1.0)
def test_redirects_are_refused(self) -> None: class Request: full_url = "https://example.invalid/a"
handler = crs.BlockRedirects() with self.assertRaisesRegex(crs.CollectError, "redirect"): handler.redirect_request(Request(), None, 302, "Found", {}, "https://elsewhere.invalid/b")
def test_default_opener_blocks_redirects(self) -> None: opener = crs.HTTP_OPENER self.assertTrue(any(isinstance(h, crs.BlockRedirects) for h in opener.handlers))
# -- budget reserved before the write ----------------------------------
def test_rejected_artifact_never_reaches_the_disk(self) -> None: collector = self.make_collector(max_total_bytes=10, fetch_fn=lambda url, **_: b"x" * 8) spec = minimal_spec(fetches=[{"id": "g", "files": [ {"url": "https://example.invalid/a", "output_name": "ns/a"}, {"url": "https://example.invalid/b", "output_name": "ns/b"}, ]}]) with self.assertRaisesRegex(crs.CollectError, "max_total_bytes"): collector.run(spec) self.assertTrue((self.output / "ns/a").exists()) self.assertFalse((self.output / "ns/b").exists())
# -- required declared sources ----------------------------------------
def test_required_source_missing_is_a_hard_failure(self) -> None: blob = tar_bytes({"core/Cargo.toml": b"x"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "rust-mos/library", "required_paths": ["Cargo.lock"]}]}) with self.assertRaisesRegex(crs.CollectError, "required source"): collector.run(spec) self.assertFalse((self.output / "s.tar.gz").exists())
def test_required_source_that_is_a_symlink_is_a_hard_failure(self) -> None: # The member exists in the tar but only as a symlink, so extraction skips # it; a declared required source must still fail rather than pass quietly. blob = tar_bytes({}, links={"Cargo.lock": "/etc/passwd"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "required_paths": ["Cargo.lock"]}]}) with self.assertRaisesRegex(crs.CollectError, "required source"): collector.run(spec)
# -- explicit obligations ----------------------------------------------
def obligation_collector(self) -> crs.Collector: blob = tar_bytes({"src/x.c": b"src"})
def fetch(url, **_kwargs): return blob if url.endswith("/t") else b"LIC"
return self.make_collector(fetch_fn=fetch)
def test_obligation_needs_both_source_and_notice(self) -> None: collector = self.obligation_collector() spec = minimal_spec( fetches=[{"id": "g", "files": [ {"url": "https://example.invalid/L", "output_name": "n/LICENSE"}]}], source_archive={"name": "s.tar.gz", "inputs": [ {"id": "i", "kind": "url_archive", "url": "https://example.invalid/t", "arcname": "ns"}]}, obligations=[ {"id": "ok", "component": "ok", "required_sources": ["ns/src/x.c"], "required_notices": ["n/LICENSE"]}, {"id": "miss-src", "component": "miss-src", "required_sources": ["ns/src/y.c"], "required_notices": ["n/LICENSE"]}, {"id": "miss-notice", "component": "miss-notice", "required_sources": ["ns/src/x.c"], "required_notices": ["n/NOTICE"]}, {"id": "notice-only", "component": "notice-only", "required_sources": [], "required_notices": ["n/LICENSE"]}, {"id": "undeclared", "component": "undeclared"}, ], ) statuses = {o["component"]: o for o in collector.run(spec)["obligations"]} self.assertEqual(statuses["ok"]["status"], crs.STATUS_COLLECTED) self.assertEqual(statuses["miss-src"]["status"], crs.STATUS_INCOMPLETE) self.assertEqual(statuses["miss-src"]["missing_sources"], ["ns/src/y.c"]) self.assertEqual(statuses["miss-notice"]["status"], crs.STATUS_INCOMPLETE) self.assertEqual(statuses["miss-notice"]["missing_notices"], ["n/NOTICE"]) self.assertEqual(statuses["notice-only"]["status"], crs.STATUS_COLLECTED) self.assertEqual(statuses["undeclared"]["status"], crs.STATUS_DEFAULT)
def test_a_bare_license_never_credits_a_source_obligation(self) -> None: # The old namespace-prefix rule credited "collected" from any file in the # same namespace. A notice alone must not satisfy a source obligation. collector = self.make_collector(fetch_fn=lambda url, **_: b"license text") spec = minimal_spec( fetches=[{"id": "g", "files": [ {"url": "https://example.invalid/L", "output_name": "rust-mos/LICENSE-MIT"}]}], obligations=[{"id": "o", "component": "runtime", "required_sources": ["rust-mos/library/core/Cargo.toml"], "required_notices": ["rust-mos/LICENSE-MIT"]}], ) entry = collector.run(spec)["obligations"][0] self.assertEqual(entry["status"], crs.STATUS_INCOMPLETE) self.assertEqual(entry["missing_sources"], ["rust-mos/library/core/Cargo.toml"])
def test_obligation_reports_dependencies(self) -> None: collector = self.make_collector() spec = minimal_spec(obligations=[{ "id": "o", "component": "runtime", "depends_on": ["rust-lang/compiler-builtins"], "required_sources": []}]) entry = collector.run(spec)["obligations"][0] self.assertEqual(entry["depends_on"], ["rust-lang/compiler-builtins"])
def test_dependency_records_land_in_the_archive_inputs(self) -> None: blob = tar_bytes({"Cargo.toml": b"x"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "required_paths": ["Cargo.toml"], "dependencies": [{"crate": "compiler_builtins", "locked_version": "0.1.160"}]}]}) record = collector.run(spec)["source_archive"]["inputs"][0] self.assertEqual(record["required_paths"], ["Cargo.toml"]) self.assertEqual(record["required_paths_verified"], 1) self.assertEqual(record["dependencies"][0]["locked_version"], "0.1.160")
# -- portable report ---------------------------------------------------
def test_portable_path_is_repo_relative(self) -> None: self.assertEqual(crs.portable_path(crs.REPOSITORY_ROOT / "target/x"), "target/x") private_example = "/" + "home/someone/secret/report.json" self.assertNotIn("/home/", crs.portable_path(private_example)) self.assertEqual(crs.portable_path(private_example), "report.json")
def test_report_output_root_has_no_absolute_home_path(self) -> None: collector = self.make_collector() report = collector.run(minimal_spec()) self.assertNotIn("/home/", report["output_root"])
# -- the report is written through the safe writer ---------------------
def write_spec(self, name: str = "in.json") -> Path: path = self.root / name path.write_text(json.dumps(minimal_spec()), encoding="utf-8") return path
def test_main_writes_the_report_inside_the_approved_root(self) -> None: spec_path = self.write_spec() target = self.root / "target" / "rs" code = crs.main(["--input", str(spec_path), "--output", str(target / "collected"), "--report", str(target / "runtime-sources-report.json"), "--json"]) self.assertEqual(code, 0) self.assertTrue((target / "runtime-sources-report.json").exists())
def test_main_refuses_a_report_outside_the_approved_root(self) -> None: spec_path = self.write_spec() target = self.root / "target" / "rs" code = crs.main(["--input", str(spec_path), "--output", str(target / "collected"), "--report", str(self.root / "outside.json"), "--json"]) self.assertEqual(code, 1) self.assertFalse((self.root / "outside.json").exists())
def test_main_refuses_a_report_through_a_symlink(self) -> None: spec_path = self.write_spec() target = self.root / "target" / "rs" target.mkdir(parents=True) real = self.root / "real.json" real.write_text("{}", encoding="utf-8") (target / "runtime-sources-report.json").symlink_to(real) code = crs.main(["--input", str(spec_path), "--output", str(target / "collected"), "--report", str(target / "runtime-sources-report.json"), "--json"]) self.assertEqual(code, 1) self.assertEqual(real.read_text(encoding="utf-8"), "{}")
class FinalHardeningTests(CollectorTestCase): """Findings from the final review: availability, tree strictness, pipes, report force."""
def make_image_collector(self, tar_fn, **kwargs) -> crs.Collector: return self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_tar_fn=tar_fn, **kwargs, )
def write_spec(self, name: str = "in.json") -> Path: path = self.root / name path.write_text(json.dumps(minimal_spec()), encoding="utf-8") return path
# -- only produced artifacts count ------------------------------------
def test_absent_extract_does_not_satisfy_an_obligation(self) -> None: def reader(_reference, container_path): raise crs.CollectError(f"{container_path} in example:1 does not exist")
collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=reader, ) spec = minimal_spec( images=[{"id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/missing.o", "output_name": "ns/missing.o"}]}], obligations=[{"id": "o", "component": "c", "required_sources": ["ns/missing.o"]}], ) report = collector.run(spec) entry = report["images"][0]["files"][0] self.assertEqual(entry["status"], crs.STATUS_ABSENT) self.assertFalse(entry["available"]) self.assertEqual(report["obligations"][0]["status"], crs.STATUS_INCOMPLETE) self.assertEqual(report["obligations"][0]["missing_sources"], ["ns/missing.o"])
def test_collected_extract_is_marked_available(self) -> None: collector = self.make_collector( image_id_fn=lambda _r: "sha256:a", image_labels_fn=lambda _r: {}, image_read_fn=lambda _r, _p: b"crt0", ) spec = minimal_spec( images=[{"id": "i", "reference": "example:1", "extract": [{"container_path": "/lib/crt0.o", "output_name": "ns/crt0.o"}]}], obligations=[{"id": "o", "component": "c", "required_sources": ["ns/crt0.o"]}], ) report = collector.run(spec) self.assertTrue(report["images"][0]["files"][0]["available"]) self.assertEqual(report["obligations"][0]["status"], crs.STATUS_COLLECTED)
# -- a manifest-only or partial tree cannot pass -----------------------
def test_a_manifest_only_tree_cannot_satisfy_a_source_requirement(self) -> None: # Exactly the "fake Cargo only" case: manifests present, no implementation. blob = tar_bytes({"Cargo.toml": b"[package]", "Cargo.lock": b"lock"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "required_paths": ["core/src/lib.rs"]}]}) with self.assertRaisesRegex(crs.CollectError, "required source"): collector.run(spec) self.assertFalse((self.output / "s.tar.gz").exists())
def test_partial_tree_below_min_members_is_rejected(self) -> None: blob = tar_bytes({"core/src/lib.rs": b"core"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "required_paths": ["core/src/lib.rs"], "min_members": 100}]}) with self.assertRaisesRegex(crs.CollectError, "min_members"): collector.run(spec) self.assertFalse((self.output / "s.tar.gz").exists())
def test_min_members_must_be_positive(self) -> None: blob = tar_bytes({"core/src/lib.rs": b"core"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "min_members": 0}]}) with self.assertRaisesRegex(crs.CollectError, "min_members"): collector.run(spec)
def test_per_input_tree_digest_is_recorded(self) -> None: blob = tar_bytes({"core/src/lib.rs": b"core"}) collector = self.make_image_collector(lambda _r, _p: blob) spec = minimal_spec(images=[{"id": "tool", "reference": "example:1"}], source_archive={"name": "s.tar.gz", "inputs": [{ "id": "lib", "kind": "image_tree", "image": "tool", "container_path": "/l", "arcname": "lib", "required_paths": ["core/src/lib.rs"]}]}) entry = collector.run(spec)["source_archive"]["inputs"][0] self.assertRegex(entry["tree_digest"], crs.SHA256_RE) self.assertEqual(entry["members"], 1)
# -- the pipe reader must not stall on a partial write -----------------
def test_run_does_not_stall_on_a_partial_write(self) -> None: # A child that writes one byte and then sleeps: a buffered read(65536) # would block for the whole sleep; the deadline must win instead. script = "import sys,time; sys.stdout.write('x'); sys.stdout.flush(); time.sleep(30)" start = time.monotonic() with self.assertRaisesRegex(crs.CollectError, "deadline"): crs._run(["python3", "-c", script], max_stdout_bytes=4096, timeout=0.5) self.assertLess(time.monotonic() - start, 15.0)
def test_run_captures_stdout_and_stderr_and_reaps(self) -> None: script = "import sys; sys.stdout.write('out'); sys.stderr.write('err')" result = crs._run(["python3", "-c", script], max_stdout_bytes=4096, timeout=30.0) self.assertEqual(result.returncode, 0) self.assertEqual(result.stdout, b"out") self.assertEqual(result.stderr, b"err")
def test_run_kills_a_child_that_overruns_the_stdout_bound(self) -> None: script = "import sys; sys.stdout.write('x'*100000); sys.stdout.flush()" with self.assertRaisesRegex(crs.CollectError, "exceeded the bound"): crs._run(["python3", "-c", script], max_stdout_bytes=128, timeout=30.0)
# -- the report honours --force ---------------------------------------
def test_existing_report_is_not_overwritten_without_force(self) -> None: spec_path = self.write_spec() target = self.root / "target" / "rs" target.mkdir(parents=True) report = target / "runtime-sources-report.json" report.write_text("SENTINEL", encoding="utf-8") code = crs.main(["--input", str(spec_path), "--output", str(target / "collected"), "--report", str(report)]) self.assertEqual(code, 1) self.assertEqual(report.read_text(encoding="utf-8"), "SENTINEL")
def test_existing_report_is_overwritten_with_force(self) -> None: spec_path = self.write_spec() target = self.root / "target" / "rs" target.mkdir(parents=True) report = target / "runtime-sources-report.json" report.write_text("SENTINEL", encoding="utf-8") code = crs.main(["--input", str(spec_path), "--output", str(target / "collected"), "--report", str(report), "--force"]) self.assertEqual(code, 0) self.assertNotEqual(report.read_text(encoding="utf-8"), "SENTINEL") self.assertEqual(json.loads(report.read_text(encoding="utf-8"))["record_kind"], crs.REPORT_RECORD_KIND)
if __name__ == "__main__": unittest.main()