Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
41 kB · 790 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791"""Vendor provenance: the inventory resolution and its independent re-check.
The vendored crate tree is the one inventory entry the packager can resolve byrunning a tool rather than by collecting evidence by hand, so its status needs amechanical contract rather than a human's word. These tests pin that contract:
* ``conditional_on_vendor`` becomes ``collected`` only for an observed ``cargo vendor`` execution whose recorded command and file aggregate hold up and whose proof is the whole vendored file index, not one license file;* an operator-supplied ``--vendor-dir`` tree is never relabelled, so its entry stays pending;* every resolved entry still means "mechanically collected", not "licenses approved", and nothing here makes a package release-ready.
Two layers are used. Pure unit tests drive the helpers with fabricated recordsso each acceptance and each refusal is isolated. The end-to-end tests drive thereal packager and the real checker, with a stand-in ``cargo`` child on ``PATH``that writes a vendored tree: no network, no emulator and no hardware are used,and the real ``cargo vendor`` run stays the parent's gate."""
from __future__ import annotations
import hashlibimport jsonimport osimport sysimport tarfileimport tempfileimport unittestimport unittest.mockfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "tools"))sys.path.insert(0, str(Path(__file__).resolve().parent))
import check_release # noqa: E402import package_release as packaging # noqa: E402from test_release_package import PackageTestCase, SyntheticProject # noqa: E402
CONDITIONAL_COMPONENT = "Rust crates from Cargo.lock"# The component name the synthetic repository fixture uses for its vendored crates.SYNTHETIC_CRATE_COMPONENT = "Rust crate dependencies"
# A stand-in for `cargo vendor`: it writes two crate directories with their own# license files and prints the source-replacement snippet the real command# prints, so the packager's generated path runs end to end without a network.FAKE_CARGO_SCRIPT = '''#!/usr/bin/env python3"""Stand-in for `cargo vendor`, used by the vendor-provenance tests."""import pathlibimport sys
CRATES = ( ("serde-1.0.229", "1.0.229"), ("itoa-1.0.15", "1.0.15"),)SNIPPET = """[source.crates-io]replace-with = "vendored-sources"
[source.vendored-sources]directory = "%s""""
def main(argv): if argv[:1] != ["vendor"]: raise SystemExit("this stand-in cargo only implements `vendor`") target = pathlib.Path(argv[-1]) for name, version in CRATES: directory = target / name (directory / "src").mkdir(parents=True, exist_ok=True) (directory / "Cargo.toml").write_text( '[package]\\nname = "%s"\\nversion = "%s"\\nlicense = "MIT OR Apache-2.0"\\n' % (name, version), encoding="utf-8", ) (directory / "LICENSE-MIT").write_text("Permission is hereby granted.\\n", encoding="utf-8") (directory / "LICENSE-APACHE").write_text("Licensed under the Apache License.\\n", encoding="utf-8") (directory / "src/lib.rs").write_text("pub fn placeholder() {}\\n", encoding="utf-8") # Real cargo echoes the vendor directory exactly as it was passed on the # command line, so this prints the host output path. print(SNIPPET % target.as_posix(), end="") return 0
if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))'''
def vendored_files(*paths: str) -> list[dict]: """Fabricated vendored file entries; the hash is derived from the path.""" entries = [] for relative in paths: entries.append( { "path": relative, "size": len(relative), "sha256": hashlib.sha256(relative.encode("utf-8")).hexdigest(), } ) return sorted(entries, key=lambda item: item["path"])
STUB_FILES = vendored_files( "serde-1.0.229/Cargo.toml", "serde-1.0.229/LICENSE-MIT", "serde-1.0.229/src/lib.rs", "itoa-1.0.15/Cargo.toml", "itoa-1.0.15/COPYING", "itoa-1.0.15/src/lib.rs",)
CARGO_VENDOR_COMMAND = [ "cargo", "vendor", "--locked", "--versioned-dirs", "--manifest-path", "/work/project/Cargo.toml", "/work/project/vendor",]
def generated_record(files: list[dict], **overrides) -> dict: record = { "status": "generated_cargo_vendor", "command": list(CARGO_VENDOR_COMMAND), "directory": "vendor", "file_count": len(files), "total_bytes": sum(item["size"] for item in files), "aggregate_sha256": packaging.aggregate_digest(files), "files": [dict(item) for item in files], } record.update(overrides) return record
def supplied_record(files: list[dict]) -> dict: """What ``--vendor-dir`` records: bytes on disk, no observed generation.""" return { "status": "provided", "command": None, "directory": "vendor", "file_count": len(files), "total_bytes": sum(item["size"] for item in files), "aggregate_sha256": packaging.aggregate_digest(files), "files": [dict(item) for item in files], "notes": "operator-supplied vendor directory; not observed cargo vendor output", }
def inventory_entry(component: str, status: str, **overrides) -> dict: entry = { "component": component, "scope": "linked into both host binaries", "license": "per crate", "license_source": "vendor/<crate-name>-<version>/ license files", "status": status, } entry.update(overrides) return entry
def static_inventory() -> list[dict]: return [ inventory_entry("ultimate-tool first-party code", "collected", collected_paths=["LICENSE"]), inventory_entry(CONDITIONAL_COMPONENT, "conditional_on_vendor"), inventory_entry("Rust standard library", "pending_external"), ]
def without_component(inventory: list[dict], component: str) -> list[dict]: return [entry for entry in inventory if entry.get("component") != component]
def report_for(manifest: dict, source_manifest: dict) -> dict: report = check_release.Report() check_release.check_source_completeness(manifest, source_manifest, report) checks = {item["check"]: item for item in report.checks} return checks["source_completeness"]
def crate_entry(inventory: list[dict]) -> dict: """The synthetic fixture's vendored-crate inventory entry.""" return next(item for item in inventory if item["component"] == SYNTHETIC_CRATE_COMPONENT)
class FakeCargo: """Puts a stand-in ``cargo`` at the front of ``PATH`` for one block."""
def __init__(self, directory: Path) -> None: self.directory = directory directory.mkdir(parents=True, exist_ok=True) script = directory / "cargo" script.write_text(FAKE_CARGO_SCRIPT, encoding="utf-8") script.chmod(0o755)
def __enter__(self) -> "FakeCargo": patched = self.directory.as_posix() + os.pathsep + os.environ.get("PATH", "") self._patch = unittest.mock.patch.dict(os.environ, {"PATH": patched}) self._patch.start() return self
def __exit__(self, *exc_info) -> None: self._patch.stop()
class VendorConfigPortabilityTests(unittest.TestCase): """The archived cargo configuration must name the extracted vendor directory."""
def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory(prefix="ut-vendor-config-") self.addCleanup(self._tmp.cleanup) self.vendor_dir = Path(self._tmp.name) / "release-out" / "vendor" self.vendor_dir.mkdir(parents=True)
def observed(self, directory: str, registries: tuple[str, ...] = ("crates-io",)) -> str: """A cargo vendor snippet in the shape real cargo prints.""" lines = [] for name in registries: lines.append(f'[source."{name}"]' if ":" in name else f"[source.{name}]") lines.append('replace-with = "vendored-sources"') lines.append("") lines.append("[source.vendored-sources]") lines.append(f'directory = "{directory}"') return "\n".join(lines) + "\n"
def test_absolute_observed_directory_is_rewritten(self) -> None: rendered, provenance = packaging.normalize_generated_vendor_config( self.observed(self.vendor_dir.as_posix()), self.vendor_dir ) self.assertNotIn(self.vendor_dir.as_posix(), rendered) self.assertNotIn("/", packaging.parse_vendor_config(rendered)["directory"]) self.assertEqual(packaging.parse_vendor_config(rendered)["directory"], "vendor") self.assertEqual(provenance["replacements"], {"crates-io": "vendored-sources"}) self.assertEqual(provenance["vendored_source"], "vendored-sources") self.assertEqual(provenance["directory_field"], "vendored-sources.directory") self.assertEqual(provenance["archive_directory"], "vendor") self.assertTrue(provenance["directory_rewritten"]) self.assertTrue(provenance["observed_directory_absolute"]) # The provenance record must not carry the host path itself. self.assertNotIn(self.vendor_dir.as_posix(), json.dumps(provenance))
def test_relative_observed_directory_is_accepted_and_reported(self) -> None: relative = os.path.relpath(self.vendor_dir, Path.cwd()) rendered, provenance = packaging.normalize_generated_vendor_config( self.observed(relative), self.vendor_dir ) self.assertEqual(packaging.parse_vendor_config(rendered)["directory"], "vendor") self.assertFalse(provenance["observed_directory_absolute"]) # The observed text was not the portable directory, so it was rewritten. self.assertTrue(provenance["directory_rewritten"])
def test_an_already_portable_directory_is_not_rewritten(self) -> None: rendered, provenance = packaging.normalize_generated_vendor_config( self.observed("vendor"), Path("vendor") ) self.assertEqual(packaging.parse_vendor_config(rendered)["directory"], "vendor") self.assertFalse(provenance["directory_rewritten"]) self.assertFalse(provenance["observed_directory_absolute"])
def test_every_observed_replacement_is_retained(self) -> None: rendered, provenance = packaging.normalize_generated_vendor_config( self.observed(self.vendor_dir.as_posix(), ("crates-io", "git+https://example.invalid/crate")), self.vendor_dir, ) self.assertEqual( provenance["replacements"], {"crates-io": "vendored-sources", "git+https://example.invalid/crate": "vendored-sources"}, ) parsed = packaging.parse_vendor_config(rendered) self.assertEqual(sorted(parsed["replacements"]), ["crates-io", "git+https://example.invalid/crate"]) self.assertIn('[source."git+https://example.invalid/crate"]', rendered) self.assertEqual(parsed["directory"], "vendor")
def test_refusals_are_explicit(self) -> None: cases = { "empty output": "", "not TOML": "this is not toml = = =", "unsupported key": '[source.crates-io]\nreplace-with = "vendored-sources"\n' "[source.vendored-sources]\ndirectory = \"vendor\"\nother = \"x\"\n", "no replacement": '[source.vendored-sources]\ndirectory = "vendor"\n', "replacement to another source": '[source.crates-io]\nreplace-with = "other"\n' '[source.other]\nreplace-with = "vendored-sources"\n' '[source.vendored-sources]\ndirectory = "vendor"\n', "two vendored directories": '[source.crates-io]\nreplace-with = "vendored-sources"\n' '[source.vendored-sources]\ndirectory = "vendor"\n' '[source.vendored-2]\ndirectory = "vendor-2"\n', "both keys on one source": '[source.crates-io]\nreplace-with = "vendored-sources"\ndirectory = "x"\n' '[source.vendored-sources]\ndirectory = "vendor"\n', "not a source table": '[release]\nname = "x"\n', } for label, snippet in cases.items(): with self.subTest(label=label): with self.assertRaises(packaging.CheckFailure): packaging.normalize_generated_vendor_config(snippet, self.vendor_dir)
def test_a_directory_that_is_not_ours_is_refused(self) -> None: other = Path(self._tmp.name) / "elsewhere" / "vendor" with self.assertRaises(packaging.CheckFailure) as caught: packaging.normalize_generated_vendor_config(self.observed(other.as_posix()), self.vendor_dir) self.assertIn("is not the vendor directory this packager produced", str(caught.exception))
def test_the_archive_gate_refuses_a_host_path(self) -> None: with self.assertRaises(packaging.CheckFailure): packaging.assert_portable_vendor_config(self.observed("/work/target/candidate/vendor")) with self.assertRaises(packaging.CheckFailure): packaging.assert_portable_vendor_config(self.observed("target/vendor")) self.assertEqual( packaging.assert_portable_vendor_config(self.observed("vendor"))["directory"], "vendor" ) # The canned fallback used for an operator-supplied tree passes the gate. self.assertEqual( packaging.assert_portable_vendor_config( "[source.crates-io]\n" 'replace-with = "vendored-sources"\n\n' "[source.vendored-sources]\n" 'directory = "vendor"\n' )["replacements"], {"crates-io": "vendored-sources"}, )
class VendoredTreeHelpersTests(unittest.TestCase): def test_license_markers_name_text_but_not_source(self) -> None: for relative in ("serde-1.0.229/LICENSE-MIT", "crate/COPYING", "crate/LICENSE.txt", "crate/NOTICE"): with self.subTest(relative=relative): self.assertTrue(packaging.is_license_file(relative)) for relative in ("crate/src/license.rs", "crate/Cargo.toml", "crate/LICENSE.sha256"): with self.subTest(relative=relative): self.assertFalse(packaging.is_license_file(relative))
def test_crate_directories_group_members_and_ignore_root_files(self) -> None: grouped = packaging.vendored_crate_directories( [{"path": "serde-1.0.229/src/lib.rs"}, {"path": "itoa-1.0.15/Cargo.toml"}, {"path": "stray.txt"}] ) self.assertEqual(sorted(grouped), ["itoa-1.0.15", "serde-1.0.229"]) self.assertEqual(grouped["serde-1.0.229"], ["serde-1.0.229/src/lib.rs"])
class VendorGenerationVerifiedTests(unittest.TestCase): def test_observed_execution_with_matching_aggregate_is_verified(self) -> None: self.assertTrue(packaging.vendor_generation_verified(generated_record(STUB_FILES)))
def test_a_supplied_tree_is_never_verified(self) -> None: self.assertFalse(packaging.vendor_generation_verified(supplied_record(STUB_FILES)))
def test_refusals(self) -> None: cases = { "empty tree": generated_record([]), "missing command": generated_record(STUB_FILES, command=None), "command that never ran vendor": generated_record(STUB_FILES, command=["cargo", "build", "--locked"]), "aggregate does not match": generated_record(STUB_FILES, aggregate_sha256="0" * 64), "count disagrees with the list": generated_record(STUB_FILES, file_count=len(STUB_FILES) + 1), "no aggregate": generated_record(STUB_FILES, aggregate_sha256=None), } for label, record in cases.items(): with self.subTest(label=label): self.assertFalse(packaging.vendor_generation_verified(record))
class ResolveVendorInventoryTests(unittest.TestCase): def test_generated_tree_resolves_to_the_whole_vendored_index(self) -> None: resolved = packaging.resolve_vendor_inventory(static_inventory(), generated_record(STUB_FILES)) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) self.assertEqual(entry["status"], "collected") self.assertEqual( entry["collected_paths"], sorted(f"vendor/{item['path']}" for item in STUB_FILES), ) self.assertEqual(entry["collected_license_paths"], ["vendor/itoa-1.0.15/COPYING", "vendor/serde-1.0.229/LICENSE-MIT"]) self.assertEqual(entry["collected_crates"], ["itoa-1.0.15", "serde-1.0.229"]) self.assertEqual(entry["crates_without_license_files"], []) self.assertEqual(entry["vendor_provenance"]["command"], CARGO_VENDOR_COMMAND) self.assertEqual(entry["vendor_provenance"]["aggregate_sha256"], packaging.aggregate_digest(STUB_FILES)) self.assertIn("not a human license approval", entry["note"])
def test_proof_is_the_source_tree_not_the_first_license_file(self) -> None: resolved = packaging.resolve_vendor_inventory(static_inventory(), generated_record(STUB_FILES)) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) self.assertGreater(len(entry["collected_paths"]), len(entry["collected_license_paths"])) for relative in ("vendor/serde-1.0.229/src/lib.rs", "vendor/itoa-1.0.15/Cargo.toml"): self.assertIn(relative, entry["collected_paths"])
def test_a_crate_without_a_named_license_file_is_reported(self) -> None: files = vendored_files("bare-1.0.0/Cargo.toml", "bare-1.0.0/src/lib.rs") resolved = packaging.resolve_vendor_inventory(static_inventory(), generated_record(files)) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) self.assertEqual(entry["status"], "collected") self.assertEqual(entry["collected_license_paths"], []) self.assertEqual(entry["crates_without_license_files"], ["bare-1.0.0"])
def test_supplied_tree_keeps_the_entry_pending(self) -> None: resolved = packaging.resolve_vendor_inventory(static_inventory(), supplied_record(STUB_FILES)) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) self.assertEqual(entry["status"], "conditional_on_vendor") self.assertNotIn("collected_paths", entry)
def test_every_failed_generation_keeps_the_entry_pending(self) -> None: cases = { "empty tree": generated_record([]), "no command": generated_record(STUB_FILES, command=None), "not a vendor command": generated_record(STUB_FILES, command=["cargo", "build"]), "unmatched aggregate": generated_record(STUB_FILES, aggregate_sha256="1" * 64), } for label, record in cases.items(): with self.subTest(label=label): resolved = packaging.resolve_vendor_inventory(static_inventory(), record) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) self.assertEqual(entry["status"], "conditional_on_vendor")
def test_the_static_inventory_is_never_edited(self) -> None: base = static_inventory() snapshot = json.dumps(base, sort_keys=True) resolved = packaging.resolve_vendor_inventory(base, generated_record(STUB_FILES)) self.assertEqual(json.dumps(base, sort_keys=True), snapshot) self.assertIsNot(resolved, base) conditional = next(item for item in base if item["component"] == CONDITIONAL_COMPONENT) self.assertEqual(conditional["status"], "conditional_on_vendor") self.assertNotIn("collected_paths", conditional)
def test_unrelated_entries_pass_through_unchanged(self) -> None: base = static_inventory() resolved = packaging.resolve_vendor_inventory(base, generated_record(STUB_FILES)) by_component = {item["component"]: item for item in resolved} self.assertEqual(by_component["ultimate-tool first-party code"], base[0]) self.assertEqual(by_component["Rust standard library"], base[2]) self.assertEqual(len(resolved), len(base))
class VendorStatusCheckTests(unittest.TestCase): """The checker re-derives the resolution; it never takes the record's word."""
def checker_manifest(self, status: str) -> dict: return {"inventory": static_inventory(), "source_completeness": {"status": status}}
def resolved(self, record: dict) -> list[dict]: return packaging.resolve_vendor_inventory(static_inventory(), record)
def test_generated_resolution_is_accepted_and_reported_pending(self) -> None: record = generated_record(STUB_FILES) source_manifest = { "inventory": self.resolved(record), "vendor": record, } manifest = self.checker_manifest("incomplete_pending_third_party_notices") check = report_for(manifest, source_manifest) self.assertEqual(check["state"], "pending") self.assertEqual(check["severity"], check_release.POLICY)
def test_a_supplied_tree_relabelled_as_collected_is_refused(self) -> None: record = supplied_record(STUB_FILES) relabelled = self.resolved(record) entry = next(item for item in relabelled if item["component"] == CONDITIONAL_COMPONENT) entry["status"] = "collected" entry["collected_paths"] = [f"vendor/{item['path']}" for item in record["files"]] source_manifest = {"inventory": relabelled, "vendor": record} problems = check_release.vendor_resolution_problems(static_inventory(), relabelled, record) self.assertTrue(any("only an observed cargo vendor run resolves it" in item for item in problems), problems) check = report_for(self.checker_manifest("complete_pending_clean_rebuild_verification"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("only an observed cargo vendor run resolves it", check["detail"])
def test_a_record_whose_aggregate_no_longer_matches_is_refused(self) -> None: record = generated_record(STUB_FILES) record["aggregate_sha256"] = "0" * 64 source_manifest = {"inventory": self.resolved(generated_record(STUB_FILES)), "vendor": record} check = report_for(self.checker_manifest("incomplete_pending_third_party_notices"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("aggregate", check["detail"])
def test_a_partial_index_is_refused(self) -> None: record = generated_record(STUB_FILES) partial = self.resolved(record) entry = next(item for item in partial if item["component"] == CONDITIONAL_COMPONENT) entry["collected_paths"] = ["vendor/serde-1.0.229/LICENSE-MIT"] source_manifest = {"inventory": partial, "vendor": record} check = report_for(self.checker_manifest("incomplete_pending_third_party_notices"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("not the whole vendored file index", check["detail"])
def test_a_collected_path_outside_the_vendored_record_is_refused(self) -> None: record = generated_record(STUB_FILES) inventory = self.resolved(record) entry = next(item for item in inventory if item["component"] == CONDITIONAL_COMPONENT) entry["collected_paths"] = sorted(entry["collected_paths"] + ["vendor/moved-2.0.0/src/lib.rs"]) source_manifest = {"inventory": inventory, "vendor": record} check = report_for(self.checker_manifest("incomplete_pending_third_party_notices"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("which the vendored record does not contain", check["detail"])
def test_a_claim_above_the_vendor_state_is_refused(self) -> None: """A supplied tree keeps the entry pending, so completeness must stay incomplete.""" record = supplied_record(STUB_FILES) source_manifest = {"inventory": self.resolved(record), "vendor": record} check = report_for(self.checker_manifest("complete_pending_clean_rebuild_verification"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("incomplete_pending_third_party_notices", check["detail"])
def test_no_vendor_tree_requires_the_missing_sources_state(self) -> None: source_manifest = {"inventory": static_inventory(), "vendor": {"status": "not_included", "files": []}} check = report_for(self.checker_manifest("incomplete_pending_third_party_notices"), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) self.assertIn("incomplete_missing_vendored_crate_sources", check["detail"])
class InventoryReconciliationTests(unittest.TestCase): """A rewritten resolved inventory must not make an obligation disappear."""
def checker_manifest(self, status: str) -> dict: return {"inventory": static_inventory(), "source_completeness": {"status": status}}
def problem_detail(self, resolved: list[dict], record: dict, status: str) -> str: source_manifest = {"inventory": resolved, "vendor": record} check = report_for(self.checker_manifest(status), source_manifest) self.assertEqual(check["state"], check_release.NOT_VERIFIED) return check["detail"]
def test_a_dropped_vendor_entry_is_refused(self) -> None: """Deleting only the vendor condition, keeping every unrelated entry.""" record = supplied_record(STUB_FILES) resolved = without_component(static_inventory(), CONDITIONAL_COMPONENT) detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("drops the static entry", detail) self.assertIn(CONDITIONAL_COMPONENT, detail)
def test_a_renamed_vendor_entry_is_refused(self) -> None: record = supplied_record(STUB_FILES) resolved = without_component(static_inventory(), CONDITIONAL_COMPONENT) resolved.append(inventory_entry("Vendored crates", "conditional_on_vendor")) detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("drops the static entry", detail) self.assertIn("adds 'Vendored crates'", detail)
def test_a_duplicated_entry_is_refused(self) -> None: record = generated_record(STUB_FILES) resolved = packaging.resolve_vendor_inventory(static_inventory(), record) resolved.append(dict(resolved[2])) detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("2 times", detail)
def test_an_unsupported_status_is_refused(self) -> None: record = generated_record(STUB_FILES) resolved = packaging.resolve_vendor_inventory(static_inventory(), record) entry = next(item for item in resolved if item["component"] == CONDITIONAL_COMPONENT) entry["status"] = "approved_by_maintainer" detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("unsupported status", detail)
def test_a_status_flip_without_a_bundle_or_observation_is_refused(self) -> None: record = generated_record(STUB_FILES) resolved = static_inventory() stdlib = next(item for item in resolved if item["component"] == "Rust standard library") stdlib["status"] = "collected" stdlib["collected_paths"] = ["rust-std/lib.rs"] detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("without a resolved runtime-bundle record", detail)
def test_a_collected_entry_cannot_be_weakened(self) -> None: record = generated_record(STUB_FILES) resolved = packaging.resolve_vendor_inventory(static_inventory(), record) first = next(item for item in resolved if item["component"] == "ultimate-tool first-party code") first["status"] = "pending_external" detail = self.problem_detail(resolved, record, "complete_pending_clean_rebuild_verification") self.assertIn("which the static status cannot become", detail)
def test_an_explicit_runtime_bundle_resolution_is_accepted(self) -> None: """The vendored entry check must not refuse the runtime bundle's own resolution.""" record = supplied_record(STUB_FILES) resolved = static_inventory() stdlib = next(item for item in resolved if item["component"] == "Rust standard library") stdlib.update( { "status": "collected", "collected_paths": ["runtime-sources/rust-std.tar.gz"], "bundled_by": "tools/runtime_source_bundle.py", "runtime_source_coverage": {"reviewed": True, "complete": True, "problems": []}, } ) problems = check_release.inventory_membership_problems(static_inventory(), resolved) problems += check_release.inventory_transition_problems(static_inventory(), resolved) self.assertEqual(problems, []) # The vendored rule still demands the vendored tree be accounted for. self.assertEqual( check_release.vendored_tree_accounting_problems(static_inventory(), resolved, record), [] )
def test_a_bundle_added_component_is_accepted_only_with_its_record(self) -> None: record = supplied_record(STUB_FILES) added = inventory_entry( "MOS sysroot", "collected", collected_paths=["runtime-sources/sysroot.tar.gz"], bundled_by="tools/runtime_source_bundle.py", runtime_source_coverage={"reviewed": True, "complete": True, "problems": []}, ) resolved = static_inventory() + [added] self.assertEqual(check_release.inventory_membership_problems(static_inventory(), resolved), []) stripped = dict(added) stripped.pop("bundled_by") stripped.pop("runtime_source_coverage") problems = check_release.inventory_membership_problems(static_inventory(), static_inventory() + [stripped]) self.assertTrue(any("no runtime bundle produced" in item for item in problems), problems)
def test_the_vendored_tree_must_be_accounted_for_from_the_bytes(self) -> None: """Deleting the entry from both inventories cannot make a shipped tree vanish.""" record = supplied_record(STUB_FILES) stripped = without_component(static_inventory(), CONDITIONAL_COMPONENT) problems = check_release.vendored_tree_accounting_problems(stripped, stripped, record) self.assertEqual(len(problems), 1) self.assertIn("no inventory entry accounts for", problems[0]) # An entry that declares the obligation satisfies the rule, and so does an # operator's own collected entry naming a real shipped member. self.assertEqual(check_release.vendored_tree_accounting_problems(static_inventory(), stripped, record), []) operator = stripped + [ inventory_entry( CONDITIONAL_COMPONENT, "collected", collected_paths=["vendor/serde-1.0.229/LICENSE-MIT"], ) ] self.assertEqual(check_release.vendored_tree_accounting_problems(stripped, operator, record), []) # A collected path outside the record is not an account of the tree. wrong = stripped + [ inventory_entry(CONDITIONAL_COMPONENT, "collected", collected_paths=["vendor/absent-1.0/LICENSE"]) ] self.assertEqual(len(check_release.vendored_tree_accounting_problems(stripped, wrong, record)), 1) generated = packaging.resolve_vendor_inventory(static_inventory(), generated_record(STUB_FILES)) self.assertEqual(check_release.vendored_tree_accounting_problems(stripped, generated, record), [])
class GeneratedVendorEndToEndTests(PackageTestCase): """The real packager, driven with a stand-in ``cargo`` child."""
def generated_source_manifest(self, project: SyntheticProject) -> dict: with FakeCargo(project.tmp_path("fake-cargo")): project.run_packager("--prepare-source", "--development", "synthetic") return json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8"))
def test_the_generated_tree_resolves_the_entry_and_the_checker_accepts_it(self) -> None: project = self.project() manifest = self.generated_source_manifest(project) vendor = manifest["vendor"] self.assertEqual(vendor["status"], "generated_cargo_vendor") command = vendor["command"] self.assertEqual(command[:4], ["cargo", "vendor", "--locked", "--versioned-dirs"]) self.assertEqual(command[4], "--manifest-path") self.assertTrue(command[5].endswith("Cargo.toml")) self.assertEqual(Path(command[6]).name, "vendor") self.assertEqual(vendor["file_count"], len(vendor["files"])) self.assertEqual(vendor["aggregate_sha256"], packaging.aggregate_digest(vendor["files"]))
entry = crate_entry(manifest["inventory"]) self.assertEqual(entry["status"], "collected") self.assertEqual( entry["collected_paths"], sorted(f"vendor/{item['path']}" for item in vendor["files"]), ) with tarfile.open(project.output / manifest["archive"]["name"], "r:gz") as tar: shipped = tar.extractfile(f"{manifest['archive']['prefix']}/vendor-sources.toml").read().decode("utf-8") self.assertNotIn(str(project.output), shipped) self.assertIn('directory = "vendor"', shipped) self.assertTrue(any(path.endswith("LICENSE-MIT") for path in entry["collected_license_paths"])) self.assertEqual(entry["vendor_provenance"]["command"], command) self.assertIn("not a human license approval", entry["note"]) # The archived configuration is portable: cargo printed the host output # path, and the shipped file names the directory the archive extracts to. self.assertNotIn(str(project.output), vendor["config_snippet"]) self.assertIn('directory = "vendor"', vendor["config_snippet"]) self.assertIn('replace-with = "vendored-sources"', vendor["config_snippet"]) self.assertEqual(vendor["config_provenance"]["archive_directory"], "vendor") self.assertTrue(vendor["config_provenance"]["directory_rewritten"]) self.assertTrue(vendor["config_provenance"]["observed_directory_absolute"]) self.assertEqual(vendor["config_provenance"]["replacements"], {"crates-io": "vendored-sources"}) self.assertIn("cargo vendor", vendor["config_snippet_source"]) # Mechanically collected still means not release-ready: the third-party # runtime notices stay pending and no license approval is claimed. self.assertEqual(manifest["source_completeness"]["status"], "incomplete_pending_third_party_notices") self.assertNotIn("release_ready", manifest["source_completeness"])
project.write_binaries() project.write_build_env() project.run_packager( "--pack", "--binary-dir", str(project.binary_dir), "--development", "synthetic", ) result = project.run_checker("--json") checks = {item["check"]: item for item in json.loads(result.stdout)["checks"]} self.assertEqual(checks["source_completeness"]["state"], "pending") self.assertEqual(checks["source_archive_contents"]["state"], "verified") self.assertEqual(checks["vendor_contents"]["state"], "verified") release = json.loads((project.output / "release-manifest.json").read_text(encoding="utf-8")) self.assertFalse(release["release_ready"]) # The static inventory in the release manifest keeps the pending entry. self.assertEqual(crate_entry(release["inventory"])["status"], "conditional_on_vendor")
def test_a_supplied_tree_with_the_entry_deleted_cannot_claim_completeness(self) -> None: """DELETE only the vendor condition, keep every unrelated entry, claim complete.""" project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") project.write_binaries() project.write_build_env() project.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", ) source_path = project.output / "source-manifest.json" source_manifest = json.loads(source_path.read_text(encoding="utf-8")) source_manifest["inventory"] = without_component(source_manifest["inventory"], SYNTHETIC_CRATE_COMPONENT) source_path.write_text(json.dumps(source_manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") manifest_path = project.output / "release-manifest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["inventory"] = without_component(manifest["inventory"], SYNTHETIC_CRATE_COMPONENT) manifest["source_completeness"]["status"] = "complete_pending_clean_rebuild_verification" manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") result = project.run_checker("--json", expect=1) checks = {item["check"]: item for item in json.loads(result.stdout)["checks"]} self.assertEqual(checks["source_completeness"]["state"], check_release.NOT_VERIFIED) self.assertIn("no inventory entry accounts for", checks["source_completeness"]["detail"])
def test_a_supplied_tree_is_never_resolved(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") manifest = json.loads((project.output / "source-manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["vendor"]["status"], "provided") self.assertIsNone(manifest["vendor"]["command"]) self.assertIn("not observed cargo vendor output", manifest["vendor"]["notes"]) entry = crate_entry(manifest["inventory"]) self.assertEqual(entry["status"], "conditional_on_vendor") self.assertNotIn("collected_paths", entry) self.assertEqual(manifest["source_completeness"]["status"], "incomplete_pending_third_party_notices") self.assertTrue( any("conditional_on_vendor" in item for item in manifest["source_completeness"]["pending_inventory"]) )
project.write_binaries() project.write_build_env() project.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", ) result = project.run_checker("--json") checks = {item["check"]: item for item in json.loads(result.stdout)["checks"]} self.assertEqual(checks["source_completeness"]["state"], "pending")
def test_a_supplied_tree_rewritten_as_generated_is_refused(self) -> None: project = self.project() vendor = project.write_vendor(project.root.parent / "vendor") project.run_packager("--prepare-source", "--vendor-dir", str(vendor), "--development", "synthetic") project.write_binaries() project.write_build_env() project.run_packager( "--pack", "--vendor-dir", str(vendor), "--binary-dir", str(project.binary_dir), "--development", "synthetic", ) path = project.output / "source-manifest.json" manifest = json.loads(path.read_text(encoding="utf-8")) entry = crate_entry(manifest["inventory"]) entry["status"] = "collected" entry["collected_paths"] = [f"vendor/{item['path']}" for item in manifest["vendor"]["files"]] path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") result = project.run_checker("--json", expect=1) checks = {item["check"]: item for item in json.loads(result.stdout)["checks"]} self.assertEqual(checks["source_completeness"]["state"], check_release.NOT_VERIFIED) self.assertIn("only an observed cargo vendor run resolves it", checks["source_completeness"]["detail"])
if __name__ == "__main__": unittest.main()