| 1 | #!/usr/bin/env python3 |
| 2 | """Hermetic tests for migration gates and manual workspace qualification. |
| 3 | |
| 4 | Verifies `.github/workflows/ci.yml` keeps both migration checker commands |
| 5 | (self-tests then live scan) under the same `heavy` condition as the existing |
| 6 | command-contract boundary step, and that the boundary step itself remains |
| 7 | intact. Exercise the GH6698 workflow shell with a fake Cargo and the real |
| 8 | HOME-isolation wrapper so these wiring tests never compile or run Rust. |
| 9 | """ |
| 10 | |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import json |
| 14 | import os |
| 15 | import re |
| 16 | import shutil |
| 17 | import subprocess |
| 18 | import sys |
| 19 | import tempfile |
| 20 | import textwrap |
| 21 | import unittest |
| 22 | from pathlib import Path |
| 23 | |
| 24 | ROOT = Path(__file__).resolve().parents[1] |
| 25 | CI_PATH = ROOT / ".github" / "workflows" / "ci.yml" |
| 26 | SHARED_PROCESS_MODE = ( |
| 27 | "github.event_name == 'workflow_dispatch' && " |
| 28 | "inputs.workspace_test_mode == 'shared-process-twice'" |
| 29 | ) |
| 30 | ISSUE_6698_TESTS = ( |
| 31 | "remote_control::tests::classic_recovery_uses_persisted_seq_floor_and_ignores_older_terminal", |
| 32 | "remote_control::tests::actual_start_reclaims_runtime_chat_writer_before_worker_spawn", |
| 33 | "remote_control::tests::separate_predispatch_crashes_on_one_run_get_distinct_recovery_turn_ids", |
| 34 | "runtime_api::tests::events_endpoint_respects_since_seq_cursor", |
| 35 | "runtime_threads::tests::approval_required_awaits_external_decision_allow", |
| 36 | "runtime_threads::tests::approval_remember_grants_tool_class_without_changing_posture", |
| 37 | "tools::subagent::budget_handback_tests::budget_handback_inflight_wall_timeout_persists_unreported_usage", |
| 38 | "tools::subagent::tests::child_permission_gate::wall_deadline_ends_pending_wait_with_receipt", |
| 39 | "tools::subagent::tests::resume_keeps_recorded_reasoning_in_manifest_and_request_after_parent_changes", |
| 40 | ) |
| 41 | |
| 42 | |
| 43 | def load_ci() -> str: |
| 44 | return CI_PATH.read_text(encoding="utf-8") |
| 45 | |
| 46 | |
| 47 | def boundary_step_block(ci: str) -> str: |
| 48 | """Extract the 'Check command-contract prototype boundary' step block.""" |
| 49 | marker = "Check command-contract prototype boundary" |
| 50 | start = ci.index(marker) |
| 51 | step_start = ci.rindex("- name:", 0, start) |
| 52 | # The step ends at the next "- name:" after the marker. |
| 53 | next_step = ci.index("- name:", start + len(marker)) |
| 54 | return ci[step_start:next_step] |
| 55 | |
| 56 | |
| 57 | def migration_step_block(ci: str) -> str: |
| 58 | marker = "Check command migration manifest" |
| 59 | start = ci.index(marker) |
| 60 | step_start = ci.rindex("- name:", 0, start) |
| 61 | next_step = ci.index("- name:", start + len(marker)) |
| 62 | return ci[step_start:next_step] |
| 63 | |
| 64 | |
| 65 | def named_step(ci: str, name: str) -> str: |
| 66 | start = ci.index(f" - name: {name}\n") |
| 67 | return ci[start:ci.index(" - name:", start + 1)] |
| 68 | |
| 69 | |
| 70 | def run_shared_process_fixture(mode: str = "pass", expected_sha: str | None = None) -> dict: |
| 71 | """Execute the actual workflow shell and HOME wrapper, never real Cargo.""" |
| 72 | block = named_step(load_ci(), "Shared-process workspace qualification") |
| 73 | script = textwrap.dedent(block.split(" run: |\n", 1)[1]) |
| 74 | with tempfile.TemporaryDirectory(prefix="shared-process-wiring-") as temp: |
| 75 | fixture = Path(temp) |
| 76 | repo = fixture / "repo" |
| 77 | binary = fixture / "bin" |
| 78 | (repo / "scripts").mkdir(parents=True) |
| 79 | binary.mkdir() |
| 80 | shutil.copyfile( |
| 81 | ROOT / "scripts/with-hermetic-test-home.sh", |
| 82 | repo / "scripts/with-hermetic-test-home.sh", |
| 83 | ) |
| 84 | trace = fixture / "cargo-calls.jsonl" |
| 85 | outside = fixture / "outside" |
| 86 | outside.mkdir() |
| 87 | outside_payload = outside / "payload.txt" |
| 88 | outside_payload.write_text("outside fixture must remain unchanged", encoding="utf-8") |
| 89 | outside_payload.chmod(0o400) |
| 90 | outside.chmod(0o500) |
| 91 | fake_cargo = f"""#!{sys.executable} |
| 92 | import json, os, sys |
| 93 | from pathlib import Path |
| 94 | if sys.argv[1:] == ['-V']: |
| 95 | print('cargo fixture (no compiler invoked)') |
| 96 | raise SystemExit(0) |
| 97 | trace = Path(os.environ['FIXTURE_TRACE']) |
| 98 | calls = trace.read_text().splitlines() if trace.exists() else [] |
| 99 | call = {{'args': sys.argv[1:], 'home': os.environ['HOME'], |
| 100 | 'tmpdir': os.environ['TMPDIR'], 'threads': os.environ.get('RUST_TEST_THREADS'), |
| 101 | 'stack': os.environ['RUST_MIN_STACK'], |
| 102 | 'credential_present': bool(os.environ.get('OPENAI_API_KEY'))}} |
| 103 | with trace.open('a') as out: |
| 104 | out.write(json.dumps(call) + '\\n') |
| 105 | identities = {ISSUE_6698_TESTS!r} |
| 106 | if os.environ['FIXTURE_MODE'] == 'missing_identity': |
| 107 | identities = identities[:-1] |
| 108 | for identity in identities: |
| 109 | print('test ' + identity + ' ... ok') |
| 110 | print('fixture complete: integration targets and doctests reached') |
| 111 | if os.environ['FIXTURE_MODE'] == 'source_changed': |
| 112 | Path('scripts/with-hermetic-test-home.sh').write_text('# changed fixture source\\n') |
| 113 | if os.environ['FIXTURE_MODE'] in ('readonly_cleanup', 'cleanup_failure'): |
| 114 | root = Path(os.environ['TMPDIR']) / 'readonly-runtime' |
| 115 | nested = root / 'nested' |
| 116 | nested.mkdir(parents=True) |
| 117 | (nested / 'payload.txt').write_text('read-only plugin fixture') |
| 118 | (nested / 'payload.txt').chmod(0o400) |
| 119 | outside = Path(os.environ['FIXTURE_OUTSIDE']) |
| 120 | (root / 'external-directory').symlink_to(outside, target_is_directory=True) |
| 121 | (root / 'external-file').symlink_to(outside / 'payload.txt') |
| 122 | os.link(outside / 'payload.txt', root / 'external-hardlink') |
| 123 | nested.chmod(0o500) |
| 124 | root.chmod(0o500) |
| 125 | raise SystemExit(101 if os.environ['FIXTURE_MODE'] in ('first_failure', 'cleanup_failure') and not calls else 0) |
| 126 | """ |
| 127 | fake_rm = f"""#!{sys.executable} |
| 128 | import os, sys |
| 129 | from pathlib import Path |
| 130 | target = Path(sys.argv[-1]) |
| 131 | trace = Path(os.environ['FIXTURE_TRACE']) |
| 132 | if (os.environ['FIXTURE_MODE'] == 'cleanup_failure' |
| 133 | and target.parent == Path('/tmp') and target.name.startswith('cw69.') |
| 134 | and len(trace.read_text().splitlines()) == 1): |
| 135 | print('fixture: qualification cleanup refused', file=sys.stderr) |
| 136 | raise SystemExit(73) |
| 137 | os.execv({shutil.which('rm')!r}, ['rm', *sys.argv[1:]]) |
| 138 | """ |
| 139 | for name, content in { |
| 140 | "cargo": fake_cargo, |
| 141 | "rm": fake_rm, |
| 142 | "rustc": "#!/bin/sh\nprintf '%s\\n' 'rustc fixture (no compiler invoked)'\n", |
| 143 | "rustup": '#!/bin/sh\nprintf "%s\\n" "$FIXTURE_BIN/rustc"\n', |
| 144 | }.items(): |
| 145 | path = binary / name |
| 146 | path.write_text(content, encoding="utf-8") |
| 147 | path.chmod(0o700) |
| 148 | env = { |
| 149 | "PATH": f"{binary}{os.pathsep}{os.environ['PATH']}", |
| 150 | "HOME": str(fixture / "outer-home"), |
| 151 | "CARGO_HOME": str(fixture / "cargo-home"), |
| 152 | "RUSTUP_HOME": str(fixture / "rustup-home"), |
| 153 | "GIT_CONFIG_NOSYSTEM": "1", |
| 154 | "RUNNER_TEMP": str(fixture / "evidence"), |
| 155 | "RUNNER_OS": "Linux", "RUNNER_ARCH": "X64", |
| 156 | "RUST_MIN_STACK": "16777216", "RUST_TEST_THREADS": "1", |
| 157 | "OPENAI_API_KEY": "synthetic-untrusted-fixture-key", |
| 158 | "FIXTURE_BIN": str(binary), "FIXTURE_TRACE": str(trace), |
| 159 | "FIXTURE_MODE": mode, "FIXTURE_OUTSIDE": str(outside), |
| 160 | } |
| 161 | for args in ( |
| 162 | ["init", "-q"], ["add", "scripts/with-hermetic-test-home.sh"], |
| 163 | ["-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", |
| 164 | "commit", "-qm", "fixture", "--no-gpg-sign"], |
| 165 | ): |
| 166 | subprocess.run(["git", *args], cwd=repo, env=env, check=True, capture_output=True) |
| 167 | sha = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, env=env, text=True).strip() |
| 168 | env.update(EXPECTED_SHA=expected_sha or sha, GITHUB_SHA=sha) |
| 169 | result = subprocess.run( |
| 170 | ["bash", "-c", script], cwd=repo, env=env, text=True, capture_output=True, timeout=30, |
| 171 | ) |
| 172 | evidence = fixture / "evidence/shared-process-workspace" |
| 173 | calls = [json.loads(line) for line in trace.read_text().splitlines()] if trace.exists() else [] |
| 174 | proof = { |
| 175 | "result": result, |
| 176 | "calls": calls, |
| 177 | "evidence": {path.name: path.read_text() for path in evidence.iterdir()}, |
| 178 | "qualification_removed": [not Path(call["tmpdir"]).exists() for call in calls], |
| 179 | "outside": { |
| 180 | "directory_mode": outside.stat().st_mode & 0o777, |
| 181 | "file_mode": outside_payload.stat().st_mode & 0o777, |
| 182 | "contents": outside_payload.read_text(encoding="utf-8"), |
| 183 | }, |
| 184 | } |
| 185 | # Only the injected rm failure leaves a root; the workflow has already |
| 186 | # made its directories writable. Do not leave the deliberate fixture. |
| 187 | for call in calls: |
| 188 | if Path(call["tmpdir"]).exists(): |
| 189 | shutil.rmtree(call["tmpdir"]) |
| 190 | outside.chmod(0o700) |
| 191 | return proof |
| 192 | |
| 193 | |
| 194 | class CiWiringTests(unittest.TestCase): |
| 195 | def test_boundary_step_still_present(self) -> None: |
| 196 | ci = load_ci() |
| 197 | self.assertIn("Check command-contract prototype boundary", ci) |
| 198 | block = boundary_step_block(ci) |
| 199 | self.assertIn("test_check_command_crate_boundaries.py", block) |
| 200 | self.assertIn("check-command-crate-boundaries.py", block) |
| 201 | |
| 202 | def test_migration_self_test_present(self) -> None: |
| 203 | ci = load_ci() |
| 204 | block = migration_step_block(ci) |
| 205 | self.assertIn("test_check_command_migration_manifest.py", block) |
| 206 | |
| 207 | def test_migration_live_scan_present(self) -> None: |
| 208 | ci = load_ci() |
| 209 | block = migration_step_block(ci) |
| 210 | self.assertIn("check-command-migration-manifest.py", block) |
| 211 | |
| 212 | def test_migration_live_scan_receives_fetched_baseline(self) -> None: |
| 213 | block = migration_step_block(load_ci()) |
| 214 | self.assertIn("PR_BASE_SHA", block) |
| 215 | self.assertIn("PUSH_BEFORE_SHA", block) |
| 216 | self.assertIn("git fetch --no-tags origin", block) |
| 217 | self.assertNotIn( |
| 218 | "--depth", |
| 219 | block, |
| 220 | "baseline fetch must preserve the full ancestry used by later CI range checks", |
| 221 | ) |
| 222 | self.assertIn('--baseline-ref "${baseline}"', block) |
| 223 | |
| 224 | def test_migration_commands_are_ordered_self_test_first(self) -> None: |
| 225 | ci = load_ci() |
| 226 | block = migration_step_block(ci) |
| 227 | self.assertLess( |
| 228 | block.index("test_check_command_migration_manifest.py"), |
| 229 | block.index("check-command-migration-manifest.py"), |
| 230 | "checker self-tests must run before the live migration scan", |
| 231 | ) |
| 232 | |
| 233 | def test_migration_step_uses_heavy_condition(self) -> None: |
| 234 | ci = load_ci() |
| 235 | block = migration_step_block(ci) |
| 236 | self.assertIn("needs.changes.outputs.heavy == 'true'", block) |
| 237 | |
| 238 | def test_boundary_step_condition_matches_migration_step(self) -> None: |
| 239 | ci = load_ci() |
| 240 | boundary = boundary_step_block(ci) |
| 241 | migration = migration_step_block(ci) |
| 242 | self.assertIn("needs.changes.outputs.heavy == 'true'", boundary) |
| 243 | self.assertEqual( |
| 244 | "needs.changes.outputs.heavy == 'true'" in boundary, |
| 245 | "needs.changes.outputs.heavy == 'true'" in migration, |
| 246 | ) |
| 247 | |
| 248 | def test_migration_step_does_not_remove_boundary_step(self) -> None: |
| 249 | ci = load_ci() |
| 250 | # Both steps must coexist (the migration step is added beside, never |
| 251 | # replacing, the boundary step). |
| 252 | self.assertLess( |
| 253 | ci.index("Check command-contract prototype boundary"), |
| 254 | ci.index("Check command migration manifest"), |
| 255 | ) |
| 256 | |
| 257 | def test_main_ci_concurrency_is_keyed_by_sha(self) -> None: |
| 258 | ci = load_ci() |
| 259 | self.assertIn("github.workflow, github.sha", ci) |
| 260 | self.assertIn("ci-pr-{0}", ci) |
| 261 | self.assertNotIn( |
| 262 | "github.event.pull_request.number || github.ref", |
| 263 | ci, |
| 264 | "main must not share one concurrency group per ref — pending runs get cancelled", |
| 265 | ) |
| 266 | self.assertIn("name: Safety gate", ci) |
| 267 | self.assertIn("Hermetic safety and authorization tests", ci) |
| 268 | |
| 269 | def test_safety_gate_is_hermetic_for_config_home(self) -> None: |
| 270 | ci = load_ci() |
| 271 | start = ci.index("Hermetic safety and authorization tests") |
| 272 | next_step = ci.index("- name:", start + 1) |
| 273 | block = ci[start:next_step] |
| 274 | self.assertIn( |
| 275 | "sh scripts/with-hermetic-test-home.sh cargo nextest run " |
| 276 | "-p codewhale-tui -p codewhale-runtime --lib --locked --no-tests=fail " |
| 277 | "-E 'test(auto_review) | test(authority) | test(sandbox)'", block |
| 278 | ) |
| 279 | self.assertIn( |
| 280 | "sh scripts/with-hermetic-test-home.sh cargo test -p codewhale-execpolicy --locked", |
| 281 | block, |
| 282 | ) |
| 283 | self.assertNotIn("CODEWHALE_HOME:", block) |
| 284 | self.assertIn("sh scripts/with-hermetic-test-home.test.sh", ci) |
| 285 | |
| 286 | def test_valid_wiring_passes_all_assertions(self) -> None: |
| 287 | # The live workflow must satisfy every structural invariant above. |
| 288 | ci = load_ci() |
| 289 | self.assertIn("test_check_command_migration_manifest.py", ci) |
| 290 | self.assertIn("check-command-migration-manifest.py", ci) |
| 291 | self.assertIn("needs.changes.outputs.heavy == 'true'", ci) |
| 292 | |
| 293 | def test_shared_process_mode_is_manual_hosted_and_preserves_default_runner(self) -> None: |
| 294 | ci = load_ci() |
| 295 | self.assertIn("default: nextest", ci) |
| 296 | matrix = ci.split(" test:\n", 1)[1].split(" steps:\n", 1)[0] |
| 297 | self.assertIn( |
| 298 | SHARED_PROCESS_MODE + " && '[\"ubuntu-latest\"]' || " |
| 299 | "'[\"ubuntu-latest\",\"macos-latest\",\"windows-latest\"]'", matrix, |
| 300 | ) |
| 301 | self.assertIn("needs.changes.outputs.trusted == 'true'", matrix) |
| 302 | qualification = named_step(ci, "Shared-process workspace qualification") |
| 303 | self.assertIn("if: " + SHARED_PROCESS_MODE + " && matrix.os == 'ubuntu-latest'", qualification) |
| 304 | for name, command in ( |
| 305 | ("Run tests", "cargo nextest run --workspace --all-features --locked --profile ci"), |
| 306 | ("Run doctests", "cargo test --workspace --all-features --locked --doc"), |
| 307 | ): |
| 308 | step = named_step(ci, name) |
| 309 | self.assertIn(command, step) |
| 310 | self.assertIn(f"!({SHARED_PROCESS_MODE})", step) |
| 311 | upload = named_step(ci, "Retain shared-process qualification evidence") |
| 312 | self.assertIn("if: always() && " + SHARED_PROCESS_MODE, upload) |
| 313 | self.assertIn("if-no-files-found: error", upload) |
| 314 | |
| 315 | def test_shared_process_runs_exact_command_twice_with_fresh_hermetic_homes(self) -> None: |
| 316 | proof = run_shared_process_fixture() |
| 317 | self.assertEqual(proof["result"].returncode, 0, proof["result"].stderr) |
| 318 | self.assertEqual(len(proof["calls"]), 2) |
| 319 | self.assertEqual(set(proof["evidence"]["required-tests.txt"].splitlines()), set(ISSUE_6698_TESTS)) |
| 320 | for call in proof["calls"]: |
| 321 | self.assertEqual(call["args"], ["test", "--workspace", "--all-features", "--locked", "--", "--format=pretty"]) |
| 322 | self.assertIsNone(call["threads"]) |
| 323 | self.assertEqual(call["stack"], "16777216") |
| 324 | self.assertFalse(call["credential_present"]) |
| 325 | self.assertTrue(call["tmpdir"].startswith("/tmp/cw69.")) |
| 326 | self.assertLess(len(call["tmpdir"]), 30) |
| 327 | self.assertFalse(Path(call["home"]).exists(), "the actual wrapper must clean up its private home") |
| 328 | self.assertNotEqual(proof["calls"][0]["home"], proof["calls"][1]["home"]) |
| 329 | self.assertNotEqual(proof["calls"][0]["tmpdir"], proof["calls"][1]["tmpdir"]) |
| 330 | for run in (1, 2): |
| 331 | self.assertIn(f"run={run} cargo_exit=0 log_exit=0", proof["evidence"]["results.txt"]) |
| 332 | for identity in ISSUE_6698_TESTS: |
| 333 | self.assertIn(f"test {identity} ... ok", proof["evidence"][f"run-{run}.log"]) |
| 334 | |
| 335 | def test_shared_process_keeps_macos_budget_when_self_hosted_test_leg_is_absent(self) -> None: |
| 336 | job = load_ci().split(" macos-budget:\n", 1)[1].split(" steps:\n", 1)[0] |
| 337 | condition = next(line.strip()[4:] for line in job.splitlines() if line.strip().startswith("if: ")) |
| 338 | self.assertIn("runs-on: macos-latest", job) |
| 339 | for event in ("pull_request", "push", "schedule", "workflow_dispatch"): |
| 340 | for mode in ("nextest", "shared-process-twice"): |
| 341 | for heavy, trusted, self_hosted in ( |
| 342 | (True, True, True), (True, True, False), |
| 343 | (True, False, True), (False, True, True), |
| 344 | ): |
| 345 | with self.subTest(event=event, mode=mode, heavy=heavy, trusted=trusted, self_hosted=self_hosted): |
| 346 | expression = condition |
| 347 | for name, value in { |
| 348 | "github.event_name": event, |
| 349 | "inputs.workspace_test_mode": mode, |
| 350 | "needs.changes.outputs.heavy": str(heavy).lower(), |
| 351 | "needs.changes.outputs.trusted": str(trusted).lower(), |
| 352 | "vars.CW_SELF_HOSTED_MAC": str(self_hosted).lower(), |
| 353 | }.items(): |
| 354 | expression = expression.replace(name, repr(value)) |
| 355 | expression = re.sub(r"!(?!=)", "not ", expression.replace("&&", "and").replace("||", "or")) |
| 356 | actual = eval(expression, {"__builtins__": {}}, {}) |
| 357 | previous = heavy and event != "pull_request" and not (trusted and self_hosted) |
| 358 | qualification = heavy and event == "workflow_dispatch" and mode == "shared-process-twice" |
| 359 | self.assertEqual(actual, previous or qualification) |
| 360 | |
| 361 | def test_shared_process_retains_failed_first_run_and_still_runs_second_once(self) -> None: |
| 362 | proof = run_shared_process_fixture("first_failure") |
| 363 | self.assertNotEqual(proof["result"].returncode, 0) |
| 364 | self.assertEqual(len(proof["calls"]), 2) |
| 365 | self.assertIn("run=1 cargo_exit=101 log_exit=0", proof["evidence"]["results.txt"]) |
| 366 | self.assertIn("run=2 cargo_exit=0 log_exit=0", proof["evidence"]["results.txt"]) |
| 367 | self.assertIn("run-1.log", proof["evidence"]) |
| 368 | self.assertIn("run-2.log", proof["evidence"]) |
| 369 | |
| 370 | def test_shared_process_cleans_readonly_directories_without_following_links(self) -> None: |
| 371 | proof = run_shared_process_fixture("readonly_cleanup") |
| 372 | self.assertEqual(proof["result"].returncode, 0, proof["result"].stderr) |
| 373 | self.assertEqual(len(proof["calls"]), 2) |
| 374 | self.assertEqual(proof["qualification_removed"], [True, True]) |
| 375 | self.assertEqual(proof["outside"], { |
| 376 | "directory_mode": 0o500, "file_mode": 0o400, |
| 377 | "contents": "outside fixture must remain unchanged", |
| 378 | }) |
| 379 | for run in (1, 2): |
| 380 | self.assertIn(f"run={run} permission_exit=0 cleanup_exit=0", proof["evidence"]["results.txt"]) |
| 381 | |
| 382 | def test_shared_process_cleanup_failure_retains_cargo_failure_and_runs_second(self) -> None: |
| 383 | proof = run_shared_process_fixture("cleanup_failure") |
| 384 | self.assertNotEqual(proof["result"].returncode, 0) |
| 385 | self.assertEqual(len(proof["calls"]), 2) |
| 386 | self.assertEqual(proof["qualification_removed"], [False, True]) |
| 387 | results = proof["evidence"]["results.txt"] |
| 388 | self.assertIn("run=1 cargo_exit=101 log_exit=0", results) |
| 389 | self.assertIn("run=1 permission_exit=0 cleanup_exit=73", results) |
| 390 | self.assertIn("run=2 cargo_exit=0 log_exit=0", results) |
| 391 | self.assertIn("run=2 permission_exit=0 cleanup_exit=0", results) |
| 392 | self.assertIn("run-1.log", proof["evidence"]) |
| 393 | self.assertIn("run-2.log", proof["evidence"]) |
| 394 | |
| 395 | def test_shared_process_missing_identity_is_not_a_pass(self) -> None: |
| 396 | proof = run_shared_process_fixture("missing_identity") |
| 397 | self.assertNotEqual(proof["result"].returncode, 0) |
| 398 | self.assertEqual(len(proof["calls"]), 2) |
| 399 | self.assertEqual(proof["evidence"]["results.txt"].count("result=missing_pass"), 2) |
| 400 | |
| 401 | def test_shared_process_refuses_wrong_revision_and_changed_source(self) -> None: |
| 402 | for wrong_sha in ("not-a-sha", "0" * 40): |
| 403 | with self.subTest(expected_sha=wrong_sha): |
| 404 | proof = run_shared_process_fixture(expected_sha=wrong_sha) |
| 405 | self.assertNotEqual(proof["result"].returncode, 0) |
| 406 | self.assertEqual(proof["calls"], []) |
| 407 | self.assertIn("preflight-error.txt", proof["evidence"]) |
| 408 | proof = run_shared_process_fixture("source_changed") |
| 409 | self.assertNotEqual(proof["result"].returncode, 0) |
| 410 | self.assertEqual(len(proof["calls"]), 1) |
| 411 | self.assertIn("run=1 source_changed=true", proof["evidence"]["results.txt"]) |
| 412 | self.assertIn("run=2 not_run=source_changed", proof["evidence"]["results.txt"]) |
| 413 | |
| 414 | |
| 415 | if __name__ == "__main__": |
| 416 | unittest.main() |
| 417 |