cat / goq
clone

turns a linux box into a private steam streaming appliance: plug in, tap your key, play from another computer

fix/hero-release-note
goq / scripts / tests / test_ci_cross_build_policy.py
test_ci_cross_build_policy.py 26.8K
1from __future__ import annotations23import unittest4from pathlib import Path56from scripts.verify_ci_cross_build_policy import PolicyError, verify, verify_gate_script789REPO_DIR = Path(__file__).resolve().parents[2]10WORKFLOW = (REPO_DIR / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")11GATE_SCRIPT = (REPO_DIR / "scripts" / "verify-demo-build.sh").read_text(12 encoding="utf-8"13)141516def replace_once(source: str, old: str, new: str) -> str:17 if source.count(old) != 1:18 raise AssertionError(f"fixture source must contain exactly one {old!r}")19 return source.replace(old, new, 1)202122def job_marker(name: str) -> str:23 return f" {name}:\n"242526def add_job_field(source: str, job: str, field: str) -> str:27 """Insert a field as the first entry of the named job."""28 return replace_once(source, job_marker(job), f"{job_marker(job)} {field}\n")293031def in_job(source: str, job: str, old: str, new: str) -> str:32 """Replace the first occurrence of `old` at or after the named job."""33 start = source.index(job_marker(job))34 offset = source.index(old, start)35 return source[:offset] + new + source[offset + len(old) :]363738class WorkflowStructureTests(unittest.TestCase):39 def test_repository_workflow_and_gate_script_pass(self) -> None:40 verify(WORKFLOW)41 verify_gate_script(GATE_SCRIPT)4243 def test_rejects_extra_top_level_field(self) -> None:44 mutated = replace_once(WORKFLOW, "name: CI\n", "name: CI\nenv:\n X: '1'\n")45 with self.assertRaisesRegex(PolicyError, "top-level fields changed"):46 verify(mutated)4748 def test_rejects_failure_suppressing_workflow_shell_default(self) -> None:49 mutated = replace_once(50 WORKFLOW,51 "name: CI\n\non:",52 "name: CI\n\ndefaults:\n run:\n shell: 'bash {0} || true'\n\non:",53 )54 with self.assertRaisesRegex(PolicyError, "workflow-level defaults are forbidden"):55 verify(mutated)5657 def test_rejects_missing_pull_request_trigger(self) -> None:58 mutated = replace_once(WORKFLOW, " pull_request:\n", "")59 with self.assertRaisesRegex(PolicyError, "empty pull_request trigger"):60 verify(mutated)6162 def test_rejects_filtered_pull_request_trigger(self) -> None:63 for filter_line in (64 " branches: [main]",65 " paths: ['crates/**']",66 " types: [opened]",67 ):68 with self.subTest(filter_line=filter_line):69 mutated = replace_once(70 WORKFLOW, " pull_request:\n", f" pull_request:\n{filter_line}\n"71 )72 with self.assertRaisesRegex(PolicyError, "exactly empty and unfiltered"):73 verify(mutated)7475 def test_rejects_inline_pull_request_mapping(self) -> None:76 mutated = replace_once(WORKFLOW, " pull_request:", " pull_request: {}")77 with self.assertRaisesRegex(PolicyError, "exactly empty and unfiltered"):78 verify(mutated)7980 def test_rejects_widened_push_trigger(self) -> None:81 mutated = replace_once(82 WORKFLOW, " branches: [main]", " branches: [main, release]"83 )84 with self.assertRaisesRegex(PolicyError, "limited to main"):85 verify(mutated)8687 def test_rejects_unparseable_workflow(self) -> None:88 mutated = replace_once(WORKFLOW, "permissions:\n", "permissions:\n\t- bogus\n")89 with self.assertRaises(PolicyError):90 verify(mutated)919293class JobInventoryTests(unittest.TestCase):94 def test_rejects_job_added_outside_the_gate(self) -> None:95 mutated = WORKFLOW + (96 "\n sneaky:\n"97 " runs-on: ubuntu-24.04\n"98 " steps:\n"99 " - name: Do something unreviewed\n"100 " run: echo unexpected\n"101 )102 with self.assertRaisesRegex(PolicyError, "jobs changed from the CI policy"):103 verify(mutated)104105 def test_rejects_removed_gate_leg(self) -> None:106 start = WORKFLOW.index(job_marker("release-containment"))107 end = WORKFLOW.index(job_marker("decky-gate"))108 mutated = WORKFLOW[:start] + WORKFLOW[end:]109 with self.assertRaisesRegex(PolicyError, "jobs changed from the CI policy"):110 verify(mutated)111112 def test_rejects_summary_job_missing_a_dependency(self) -> None:113 mutated = replace_once(WORKFLOW, " - release-containment\n", "")114 with self.assertRaisesRegex(PolicyError, "must depend on every other job"):115 verify(mutated)116117 def test_rejects_duplicate_summary_dependency(self) -> None:118 mutated = replace_once(119 WORKFLOW,120 " - release-containment\n",121 " - release-containment\n - release-containment\n",122 )123 with self.assertRaisesRegex(PolicyError, "duplicate"):124 verify(mutated)125126127class SummaryJobTests(unittest.TestCase):128 def test_rejects_renamed_summary_job(self) -> None:129 mutated = replace_once(130 WORKFLOW, " name: Complete demo gate", " name: Optional demo gate"131 )132 with self.assertRaisesRegex(PolicyError, "required name"):133 verify(mutated)134135 def test_rejects_summary_job_off_the_pinned_runner(self) -> None:136 mutated = in_job(WORKFLOW, "demo-gate", "runs-on: ubuntu-24.04", "runs-on: ubuntu-latest")137 with self.assertRaisesRegex(PolicyError, "must run on ubuntu-24.04"):138 verify(mutated)139140 def test_rejects_summary_job_that_cannot_report_failures(self) -> None:141 for condition in ("if: false", "if: success()", "if: github.ref == 'x'"):142 with self.subTest(condition=condition):143 mutated = in_job(WORKFLOW, "demo-gate", "if: always()", condition)144 with self.assertRaisesRegex(PolicyError, "if: always\\(\\)"):145 verify(mutated)146147 def test_rejects_summary_job_that_never_fails(self) -> None:148 mutated = replace_once(149 WORKFLOW,150 " echo 'complete demo gate failed' >&2\n exit 1\n",151 " echo 'complete demo gate failed' >&2\n",152 )153 with self.assertRaisesRegex(PolicyError, "must fail when a leg did not pass"):154 verify(mutated)155156 def test_rejects_widened_skippable_leg_list(self) -> None:157 mutated = replace_once(158 WORKFLOW,159 "skippable='native-tests cross-build",160 "skippable='quick-checks repo-tests native-tests cross-build",161 )162 with self.assertRaisesRegex(PolicyError, "skippable legs changed"):163 verify(mutated)164165 def test_rejects_skip_tolerated_without_the_scope_decision(self) -> None:166 mutated = replace_once(167 WORKFLOW,168 " if [[ \"$SCOPE_CODE\" == 'false' ]]; then\n docs_only=true\n fi",169 " docs_only=true",170 )171 with self.assertRaisesRegex(172 PolicyError, "only tolerate a skip the docs-only scope made"173 ):174 verify(mutated)175176177class LegConditionTests(unittest.TestCase):178 def test_rejects_conditionally_disabled_always_run_leg(self) -> None:179 for leg in ("quick-checks", "repo-tests", "decky-gate", "scope"):180 with self.subTest(leg=leg):181 mutated = add_job_field(WORKFLOW, leg, "if: false")182 with self.assertRaises(PolicyError):183 verify(mutated)184185 def test_rejects_quoted_conditionally_disabled_leg(self) -> None:186 mutated = add_job_field(WORKFLOW, "quick-checks", '"if": false')187 with self.assertRaises(PolicyError):188 verify(mutated)189190 def test_rejects_skippable_leg_with_a_different_condition(self) -> None:191 for condition in (192 "if: false",193 "if: github.actor != 'nobody'",194 "if: needs.scope.outputs.code != 'true'",195 ):196 with self.subTest(condition=condition):197 mutated = in_job(198 WORKFLOW,199 "native-tests",200 "if: needs.scope.outputs.code == 'true'",201 condition,202 )203 with self.assertRaisesRegex(PolicyError, "audited docs-only scope"):204 verify(mutated)205206 def test_rejects_leg_continue_on_error(self) -> None:207 for leg in ("cross-build", "release-containment", "demo-gate"):208 with self.subTest(leg=leg):209 mutated = add_job_field(WORKFLOW, leg, "continue-on-error: true")210 with self.assertRaisesRegex(PolicyError, "must not suppress failures"):211 verify(mutated)212213 def test_rejects_leg_run_defaults_override(self) -> None:214 mutated = add_job_field(215 WORKFLOW, "cross-build", "defaults:\n run:\n shell: 'bash {0} || true'"216 )217 with self.assertRaisesRegex(PolicyError, "must not override run defaults"):218 verify(mutated)219220221class StepIntegrityTests(unittest.TestCase):222 def test_rejects_step_that_masks_failure(self) -> None:223 for token in ("|| true", "set +e", "trap 'exit 0' EXIT", "exec true"):224 with self.subTest(token=token):225 mutated = in_job(226 WORKFLOW,227 "native-tests",228 "run: ./scripts/verify-demo-build.sh --stage native",229 f"run: ./scripts/verify-demo-build.sh --stage native {token}",230 )231 with self.assertRaisesRegex(PolicyError, "forbidden token"):232 verify(mutated)233234 def test_rejects_step_poisoning_the_cargo_toolchain(self) -> None:235 mutated = in_job(236 WORKFLOW,237 "native-tests",238 " - name: Run the native gate stage\n",239 " - name: Poison Cargo environment\n"240 " run: printf 'x\\n' 'export RUSTC_WRAPPER=true' >>~/.cargo/env\n\n"241 " - name: Run the native gate stage\n",242 )243 with self.assertRaisesRegex(PolicyError, "forbidden token"):244 verify(mutated)245246 def test_rejects_conditionally_disabled_step(self) -> None:247 mutated = in_job(248 WORKFLOW,249 "cross-build",250 " - name: Install pinned cross-build tools\n",251 " - name: Install pinned cross-build tools\n if: false\n",252 )253 with self.assertRaisesRegex(PolicyError, "conditionally disabled step"):254 verify(mutated)255256 def test_rejects_step_continue_on_error(self) -> None:257 mutated = in_job(258 WORKFLOW,259 "cross-build",260 " - name: Install pinned cross-build tools\n",261 " - name: Install pinned cross-build tools\n"262 " continue-on-error: true\n",263 )264 with self.assertRaisesRegex(PolicyError, "step that suppresses failures"):265 verify(mutated)266267 def test_rejects_action_that_is_not_digest_pinned(self) -> None:268 mutated = in_job(269 WORKFLOW,270 "quick-checks",271 "uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1",272 "uses: actions/checkout@v6",273 )274 with self.assertRaisesRegex(PolicyError, "not digest-pinned"):275 verify(mutated)276277 def test_rejects_checkout_ref_override(self) -> None:278 mutated = in_job(279 WORKFLOW,280 "cross-build",281 "uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1",282 "uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1\n"283 " with:\n ref: main",284 )285 with self.assertRaisesRegex(PolicyError, "overrides the checked-out ref"):286 verify(mutated)287288 def test_rejects_swapped_checkout_digest(self) -> None:289 mutated = in_job(290 WORKFLOW,291 "quick-checks",292 "actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8",293 "actions/checkout@" + "b" * 40,294 )295 with self.assertRaisesRegex(PolicyError, "not digest-pinned to the policy"):296 verify(mutated)297298299class GateLegTests(unittest.TestCase):300 def test_rejects_leg_running_the_wrong_stage(self) -> None:301 mutated = in_job(302 WORKFLOW,303 "release-containment",304 "run: ./scripts/verify-demo-build.sh --stage containment",305 "run: ./scripts/verify-demo-build.sh --stage quick",306 )307 with self.assertRaisesRegex(PolicyError, "must run the repository gate stage"):308 verify(mutated)309310 def test_rejects_leg_that_does_not_run_the_gate(self) -> None:311 mutated = in_job(312 WORKFLOW,313 "native-tests",314 "run: ./scripts/verify-demo-build.sh --stage native",315 "run: echo bypassed",316 )317 mutated += "\n# run: ./scripts/verify-demo-build.sh --stage native\n"318 with self.assertRaisesRegex(PolicyError, "must invoke the repository gate"):319 verify(mutated)320321 def test_rejects_cross_build_leg_without_the_required_cross_build(self) -> None:322 mutated = in_job(323 WORKFLOW, "cross-build", 'GOQ_REQUIRE_LINUX_CROSS_BUILD: "1"', 'GOQ_REQUIRE_LINUX_CROSS_BUILD: "0"'324 )325 with self.assertRaisesRegex(PolicyError, "does not require the Linux cross build"):326 verify(mutated)327328 def test_rejects_cross_build_leg_without_the_gstreamer_backend(self) -> None:329 mutated = in_job(330 WORKFLOW,331 "cross-build",332 ' GOQ_VERIFY_IN_PROCESS_GSTREAMER: "1"\n',333 "",334 )335 with self.assertRaisesRegex(PolicyError, "in-process GStreamer backend"):336 verify(mutated)337338 def test_rejects_gstreamer_leg_without_its_gate(self) -> None:339 mutated = in_job(340 WORKFLOW,341 "gstreamer-gate",342 ' GOQ_VERIFY_IN_PROCESS_GSTREAMER: "1"\n',343 "",344 )345 with self.assertRaisesRegex(346 PolicyError, "does not require the in-process GStreamer gate"347 ):348 verify(mutated)349350 def test_rejects_leg_that_skips_dependency_installation(self) -> None:351 mutated = in_job(352 WORKFLOW,353 "loopback",354 " run: ./scripts/install-linux-build-deps.sh --profile full\n",355 " run: echo skipped\n",356 )357 with self.assertRaisesRegex(PolicyError, "install the Linux build dependencies"):358 verify(mutated)359360 def test_rejects_leg_off_the_pinned_runner(self) -> None:361 mutated = in_job(362 WORKFLOW, "loopback", "runs-on: ubuntu-24.04", "runs-on: ubuntu-latest"363 )364 with self.assertRaisesRegex(PolicyError, "must run on ubuntu-24.04"):365 verify(mutated)366367368class ScopeJobTests(unittest.TestCase):369 def test_rejects_missing_fail_closed_branch(self) -> None:370 for branch in (371 " echo 'scope=full (event is not a pull request)'\n",372 " echo 'scope=full (could not enumerate the pull request diff)'\n",373 " echo 'scope=full (empty diff listing)'\n",374 ):375 with self.subTest(branch=branch):376 mutated = replace_once(WORKFLOW, branch, "")377 with self.assertRaisesRegex(PolicyError, "fail-closed branch"):378 verify(mutated)379380 def test_rejects_widened_documentation_allowlist(self) -> None:381 for arm in (382 "docs/* | website/* | *.md | LICENSE | crates/*) ;;",383 "docs/* | website/* | *.md | LICENSE | src-tauri/*) ;;",384 "*) ;;",385 ):386 with self.subTest(arm=arm):387 mutated = replace_once(388 WORKFLOW, "docs/* | website/* | *.md | LICENSE) ;;", arm389 )390 with self.assertRaisesRegex(PolicyError, "allowlist changed"):391 verify(mutated)392393 def test_rejects_extra_allowlist_arm(self) -> None:394 mutated = replace_once(395 WORKFLOW,396 " docs/* | website/* | *.md | LICENSE) ;;\n",397 " docs/* | website/* | *.md | LICENSE) ;;\n"398 " crates/*) ;;\n",399 )400 with self.assertRaisesRegex(PolicyError, "allowlist changed"):401 verify(mutated)402403 def test_rejects_scope_job_that_cannot_be_trusted(self) -> None:404 mutated = add_job_field(WORKFLOW, "scope", "if: false")405 with self.assertRaisesRegex(PolicyError, "must carry no if condition"):406 verify(mutated)407408 def test_rejects_extra_scope_output(self) -> None:409 mutated = replace_once(410 WORKFLOW,411 " code: ${{ steps.classify.outputs.code }}\n",412 " code: ${{ steps.classify.outputs.code }}\n"413 " other: ${{ steps.classify.outputs.other }}\n",414 )415 with self.assertRaisesRegex(PolicyError, "exactly the code output"):416 verify(mutated)417418419class GateScriptTests(unittest.TestCase):420 def test_rejects_changed_executable_prefix(self) -> None:421 for mutation in ("set -eu\n", "set -euo pipefail\nexit 0\n"):422 with self.subTest(mutation=mutation):423 mutated = replace_once(GATE_SCRIPT, "set -euo pipefail\n", mutation)424 with self.assertRaisesRegex(PolicyError, "executable prefix changed"):425 verify_gate_script(mutated)426427 def test_rejects_failure_masked_cross_build_helper(self) -> None:428 mutated = replace_once(429 GATE_SCRIPT,430 " ./scripts/run-linux-cross-build-gate.sh\n",431 " ./scripts/run-linux-cross-build-gate.sh || true\n",432 )433 with self.assertRaisesRegex(PolicyError, "forbidden token"):434 verify_gate_script(mutated)435436 def test_rejects_conditionally_wrapped_cross_build_helper(self) -> None:437 mutated = replace_once(438 GATE_SCRIPT,439 " ./scripts/run-linux-cross-build-gate.sh\n",440 " if false; then\n ./scripts/run-linux-cross-build-gate.sh\n fi\n",441 )442 with self.assertRaisesRegex(PolicyError, "invoked unconditionally"):443 verify_gate_script(mutated)444445 def test_rejects_duplicated_or_removed_cross_build_helper(self) -> None:446 removed = replace_once(447 GATE_SCRIPT, " ./scripts/run-linux-cross-build-gate.sh\n", ""448 )449 with self.assertRaisesRegex(PolicyError, "exactly once"):450 verify_gate_script(removed)451452 duplicated = replace_once(453 GATE_SCRIPT,454 " ./scripts/run-linux-cross-build-gate.sh\n",455 " ./scripts/run-linux-cross-build-gate.sh\n"456 " ./scripts/run-linux-cross-build-gate.sh\n",457 )458 with self.assertRaisesRegex(PolicyError, "exactly once"):459 verify_gate_script(duplicated)460461 def test_rejects_failure_control_tokens_anywhere(self) -> None:462 for token in ("set +e", "trap 'exit 0' EXIT", "exec true"):463 with self.subTest(token=token):464 mutated = replace_once(465 GATE_SCRIPT,466 "run_stage_quick() {\n",467 f"run_stage_quick() {{\n {token}\n",468 )469 with self.assertRaisesRegex(PolicyError, "forbidden token"):470 verify_gate_script(mutated)471472 def test_rejects_dropped_stage(self) -> None:473 mutated = replace_once(474 GATE_SCRIPT,475 "readonly ALL_STAGES=(quick cross native gstreamer repo-tests loopback containment)",476 "readonly ALL_STAGES=(quick native gstreamer repo-tests loopback containment)",477 )478 with self.assertRaisesRegex(PolicyError, "stages changed"):479 verify_gate_script(mutated)480481 def test_rejects_undispatched_stage(self) -> None:482 mutated = replace_once(483 GATE_SCRIPT, " cross) run_stage_cross ;;\n", ""484 )485 with self.assertRaisesRegex(PolicyError, "does not dispatch"):486 verify_gate_script(mutated)487488 def test_rejects_missing_stage_function(self) -> None:489 mutated = replace_once(490 GATE_SCRIPT, "run_stage_containment() {", "run_stage_containment_disabled() {"491 )492 with self.assertRaises(PolicyError):493 verify_gate_script(mutated)494495 def test_rejects_default_run_that_skips_stages(self) -> None:496 mutated = replace_once(497 GATE_SCRIPT,498 ' for current_stage in "${ALL_STAGES[@]}"; do',499 " for current_stage in quick; do",500 )501 with self.assertRaisesRegex(PolicyError, "must run every stage"):502 verify_gate_script(mutated)503504505class CacheBudgetTests(unittest.TestCase):506 """A cache entry belongs to the branch that wrote it.507508 Saving from pull requests gave every open pull request its own509 multi-gigabyte set, which exhausted the repository's cache ceiling and510 evicted the entries the legs depend on. These assert that saves stay gated511 to pushes and that a leg cannot smuggle in an unpaired or mismatched entry.512 """513514 def test_rejects_unconditional_cache_save(self) -> None:515 mutated = replace_once(516 WORKFLOW,517 " - name: Save the Cargo build cache\n"518 " if: github.event_name == 'push' && steps.cargo-cache.outputs.cache-hit != 'true'\n"519 " uses: actions/cache/save@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4\n"520 " with:\n"521 " path: |\n"522 " ~/.cargo/registry/index\n"523 " ~/.cargo/registry/cache\n"524 " ~/.cargo/git/db\n"525 " target\n"526 " key: cargo-repo-tests-${{ runner.os }}-${{ hashFiles('Cargo.lock', 'rust-toolchain.toml') }}\n",527 " - name: Save the Cargo build cache\n"528 " uses: actions/cache/save@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4\n"529 " with:\n"530 " path: |\n"531 " ~/.cargo/registry/index\n"532 " ~/.cargo/registry/cache\n"533 " ~/.cargo/git/db\n"534 " target\n"535 " key: cargo-repo-tests-${{ runner.os }}-${{ hashFiles('Cargo.lock', 'rust-toolchain.toml') }}\n",536 )537 with self.assertRaisesRegex(PolicyError, "unconditionally"):538 verify(mutated)539540 def test_rejects_cache_save_on_pull_requests(self) -> None:541 for condition in (542 "if: always()",543 "if: github.event_name == 'pull_request'",544 "if: steps.cargo-cache.outputs.cache-hit != 'true'",545 ):546 with self.subTest(condition=condition):547 mutated = in_job(548 WORKFLOW,549 "native-tests",550 "if: github.event_name == 'push' && steps.cargo-cache.outputs.cache-hit != 'true'",551 condition,552 )553 with self.assertRaisesRegex(PolicyError, "unapproved condition"):554 verify(mutated)555556 def test_rejects_conditional_step_that_is_not_a_cache_save(self) -> None:557 mutated = in_job(558 WORKFLOW,559 "native-tests",560 " - name: Run the native gate stage\n",561 " - name: Run the native gate stage\n"562 " if: github.event_name == 'push' && steps.cargo-cache.outputs.cache-hit != 'true'\n",563 )564 with self.assertRaisesRegex(PolicyError, "conditionally disabled step"):565 verify(mutated)566567 def test_rejects_restore_without_a_paired_save(self) -> None:568 start = WORKFLOW.index(" - name: Save the Cargo build cache\n")569 end = WORKFLOW.index(" native-tests:\n")570 mutated = WORKFLOW[:start] + WORKFLOW[end:]571 with self.assertRaisesRegex(PolicyError, "pair its Cargo cache restore"):572 verify(mutated)573574 def test_rejects_save_under_a_different_key(self) -> None:575 mutated = in_job(576 WORKFLOW,577 "release-containment",578 " - name: Save the Cargo build cache\n"579 " if: github.event_name == 'push' && steps.cargo-cache.outputs.cache-hit != 'true'\n"580 " uses: actions/cache/save@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4\n"581 " with:\n"582 " path: |\n"583 " ~/.cargo/registry/index\n"584 " ~/.cargo/registry/cache\n"585 " ~/.cargo/git/db\n"586 " target\n"587 " key: cargo-containment-",588 " - name: Save the Cargo build cache\n"589 " if: github.event_name == 'push' && steps.cargo-cache.outputs.cache-hit != 'true'\n"590 " uses: actions/cache/save@0400d5f644dc74513175e3cd8d07132dd4860809 # v4.2.4\n"591 " with:\n"592 " path: |\n"593 " ~/.cargo/registry/index\n"594 " ~/.cargo/registry/cache\n"595 " ~/.cargo/git/db\n"596 " target\n"597 " key: cargo-elsewhere-",598 )599 with self.assertRaisesRegex(PolicyError, "different key"):600 verify(mutated)601602 def test_rejects_missing_cargo_cache_id(self) -> None:603 mutated = in_job(604 WORKFLOW,605 "gstreamer-gate",606 " - name: Restore the Cargo build cache\n id: cargo-cache\n",607 " - name: Restore the Cargo build cache\n",608 )609 with self.assertRaisesRegex(PolicyError, "cargo-cache id"):610 verify(mutated)611612 def test_rejects_read_write_cache_beyond_the_zigbuild_binary(self) -> None:613 mutated = in_job(614 WORKFLOW,615 "cross-build",616 " path: ~/.cargo/bin/cargo-zigbuild\n",617 " path: target\n",618 )619 with self.assertRaisesRegex(PolicyError, "cargo-zigbuild binary"):620 verify(mutated)621622 def test_shared_matrix_cache_nominates_exactly_one_writer(self) -> None:623 """Two concurrent legs must not race to reserve the same key.624625 Cases 1-3 share one debug entry and case 4 owns the release entry, so626 exactly one leg per distinct cache key may carry save: "true".627 """628 legs = WORKFLOW[WORKFLOW.index(job_marker("loopback")) :]629 legs = legs[: legs.index(job_marker("release-containment"))]630 writers: dict[str, int] = {}631 cache = None632 for line in legs.splitlines():633 stripped = line.strip()634 if stripped.startswith("- case:") or stripped.startswith("cache:"):635 if stripped.startswith("cache:"):636 cache = stripped.split(":", 1)[1].strip()637 writers.setdefault(cache, 0)638 elif stripped.startswith("save:"):639 if stripped.split(":", 1)[1].strip().strip('"') == "true":640 assert cache is not None641 writers[cache] += 1642 self.assertEqual(writers, {"debug": 1, "release": 1})643644645if __name__ == "__main__":646 unittest.main()