from __future__ import annotations import copy import hashlib import importlib.util import json import sys from contextlib import contextmanager, nullcontext from pathlib import Path from types import SimpleNamespace import pytest @pytest.fixture def launcher(monkeypatch, tmp_path): root = Path(__file__).resolve().parents[2] / "ops/gitea-deploy-broker" monkeypatch.syspath_prepend(str(root)) spec = importlib.util.spec_from_file_location( "build_launcher_refresh_test", root / "runner_fleet_specialize.py" ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) monkeypatch.setattr(module.os, "geteuid", lambda: 0, raising=False) monkeypatch.setattr(module, "SPECIALIZED_STATE_ROOT", tmp_path) monkeypatch.setattr(module, "SPECIALIZED_STATE_PATH", tmp_path / "marker.json") monkeypatch.setattr(module, "MIGRATION_JOURNAL_PATH", tmp_path / "journal.json") for path in (module.SPECIALIZED_STATE_PATH, module.MIGRATION_JOURNAL_PATH): path.write_bytes(b"{}") monkeypatch.setattr( module, "_trusted_private_file", lambda path, **_k: path.read_bytes() if path.exists() else b"config", ) monkeypatch.setattr( module, "_read_json_file", lambda path, **_k: json.loads(path.read_bytes()) ) monkeypatch.setattr(module, "_read_journal", lambda: {}) monkeypatch.setattr( module, "_write_private_atomic", lambda path, raw: path.write_bytes(raw) ) ids = {profile: 71 + index for index, profile in enumerate(module.PROFILE_IDS)} monkeypatch.setattr(module, "_active_profile_registration_ids", lambda _j: ids) containers, rows = {}, {} for index, profile in enumerate(module.RUNNER_PROFILES.values()): name = profile.container_name containers[name] = { "Name": "/" + name, "Id": str(index + 1) * 64, "Image": module.RUNNER_IMAGE_ID, "Config": { "Image": module.RUNNER_IMAGE, "Labels": { "com.docker.compose.project": module.PROJECT, "com.docker.compose.service": "runner-" + profile.id, "tech.itworx.role": "gitea-actions-runner", "tech.itworx.ci.profile": profile.id, }, "Env": [ "CONFIG_FILE=/runner-config.yaml", "DOCKER_HOST=" + module._DIND_DOCKER_HOST, ], }, "State": {"Running": False}, "HostConfig": { "CgroupParent": "itworx-ci", "NanoCpus": 500_000_000, "Memory": 512 * module._MIB, "PidsLimit": 192, "PidMode": "", "NetworkMode": module._RUNNER_NETWORK_NAME, "Privileged": False, "ExtraHosts": ["gitea.example.com:198.19.104.1"], "RestartPolicy": {"Name": "no"}, }, "Mounts": [ { "Source": str(module._host_volume_path(source)), "Destination": destination, "RW": writable, } for source, destination, writable in ( (module._SPECIALIZED_DATA_ROOT / profile.id, "/data", True), ( module.RUNNER_CONFIG_ROOT / (profile.id + ".yaml"), "/runner-config.yaml", False, ), ( module._DIND_CONTROL_ROOT, module._DIND_CONTROL_CONTAINER_ROOT, False, ), ( module._SPECIALIZED_WORKSPACE_ROOT / name, "/" + name + "-workspace", True, ), ) ], } rows[name] = { "name": name, "id": ids[profile.id], "disabled": True, "busy": False, } model = { "services": { "runner-" + profile: {"image": module.RUNNER_IMAGE, "unchanged": profile} for profile in module.PROFILE_IDS } } monkeypatch.setattr( module, "_active_release_bundle", lambda: ({}, {}, model, module._canonical_bytes(model)), ) monkeypatch.setattr(module, "_docker_outer_inventory", lambda: (containers, [])) monkeypatch.setattr(module, "_runner_rows", lambda: list(rows.values())) monkeypatch.setattr(module, "_inspect", lambda name: containers.get(name)) monkeypatch.setattr(module, "_guard_state_unchanged", lambda _j: True) monkeypatch.setattr(module, "_fresh_recovery_host_gate", lambda: None) monkeypatch.setattr(module, "_resource_proof", lambda *_a: {"ready": True}) monkeypatch.setattr(module, "verify_profile_task_images", lambda *_a, **_k: True) monkeypatch.setattr( module, "verified_topology", lambda: { "ready": all( module._validate_outer_item(name, item, profile) for profile, name in ( (p.id, p.container_name) for p in module.RUNNER_PROFILES.values() ) for item in [containers[name]] ) }, ) admission = {"pending": None, "bootstrapReceiptSha256": "a" * 64, "revision": 7} controller = SimpleNamespace( _read_state=lambda: admission, _process_lock=lambda: nullcontext() ) monkeypatch.setitem( sys.modules, "runner_admission_controller", SimpleNamespace(RunnerAdmissionController=lambda: controller), ) release = { "sourceRevision": "b" * 40, "sourceTree": "c" * 40, "artifactSha256": "d" * 64, } monkeypatch.setitem( sys.modules, "status_server", SimpleNamespace(_require_controller_release=lambda: release), ) monkeypatch.setattr(module, "_migration_lock", lambda **_k: nullcontext()) monkeypatch.setattr( module, "_open_fixed_release_volume", lambda: nullcontext(SimpleNamespace(assert_current=lambda: None)), ) @contextmanager def maintenance(): yield lambda: None monkeypatch.setattr(module, "_rollback_maintenance", maintenance) effects = [] original = copy.deepcopy(containers["itworx-ci-build"]) def docker(args, **_k): if args[:2] == ["docker", "exec"]: assert args == [ "docker", "exec", "-e", "DOCKER_HOST=" + module._DIND_DOCKER_HOST, module.DIND_NAME, "docker", "ps", "-a", "-q", ] return "" if args[:3] == ["docker", "image", "inspect"]: return json.dumps([{"Id": args[-1]}]) if args[:2] == ["docker", "inspect"]: result = [ item for name, item in containers.items() if name in args[2:] or item["Id"] in args[2:] ] if not result: raise module.RunnerFleetSpecializeError( "runner_fleet_specialization_command_failed" ) return json.dumps(result) if args[:2] == ["docker", "rename"]: name, item = next( (name, item) for name, item in containers.items() if item["Id"] == args[2] ) assert args[3] not in containers containers.pop(name) item["Name"] = "/" + args[3] containers[args[3]] = item effects.append(tuple(args)) elif args[:2] == ["docker", "create"]: effects.append(tuple(args)) item = copy.deepcopy(original) item["Id"] = "a" * 64 item["Image"] = module.BUILD_LAUNCHER_IMAGE item["Config"]["Image"] = module.BUILD_LAUNCHER_IMAGE for index, token in enumerate(args): if token == "--label": key, value = args[index + 1].split("=", 1) item["Config"]["Labels"][key] = value assert "itworx-ci-build" not in containers containers["itworx-ci-build"] = item return item["Id"] elif args[:2] == ["docker", "rm"]: name = next( name for name, item in containers.items() if item["Id"] == args[2] ) containers.pop(name) effects.append(tuple(args)) return "" monkeypatch.setattr(module, "_run", docker) monkeypatch.setattr( module, "_run_compose", lambda *_a, **_k: pytest.fail( "Archived Compose must not recreate the launcher" ), ) return module, containers, rows, admission, effects, model def test_refresh_uses_real_outer_image_postcondition_in_created_phase(launcher, monkeypatch): module, containers, rows, _, _, _ = launcher before = copy.deepcopy(containers) registrations = copy.deepcopy(rows) monkeypatch.setattr( module, "verify_profile_task_images", lambda profiles, **kwargs: module._verify_specialized_outer_images(profiles, **kwargs), ) preflight = module._build_launcher_preflight("refresh") result = module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert result["status"] == "applied" assert rows == registrations for name in before: if name != "itworx-ci-build": assert containers[name] == before[name] @pytest.mark.parametrize("change", [ "forged_receipt", "partial_profiles", "changed_release", "wrong_stage", "running_candidate", "foreign_candidate_id", "wrong_candidate_image", "wrong_operation_label", "changed_mount", "running_backup", "changed_other_limit", ]) def test_created_launcher_scope_rejects_drift_and_keeps_normal_admission_blocked(launcher, monkeypatch, change): module, containers, _, _, effects, _ = launcher def checked(profiles, **kwargs): receipt = module._build_launcher_receipt() if receipt is None or receipt["status"] != "created": return module._verify_specialized_outer_images(profiles, **kwargs) assert module._verify_specialized_outer_images(profiles) is False assert module.verified_topology()["ready"] is False assert module._verify_specialized_outer_images(profiles, **kwargs) is True original_containers = copy.deepcopy(containers) original_receipt = copy.deepcopy(receipt) release = sys.modules["status_server"]._require_controller_release() original_release = copy.deepcopy(release) effects_before = copy.deepcopy(effects) candidate = containers["itworx-ci-build"] scoped = dict(kwargs) selected = profiles if change == "forged_receipt": scoped["pending_build_launcher"] = {**receipt, "operationId": "9" * 64} elif change == "partial_profiles": selected = ("build",) elif change == "changed_release": release["sourceRevision"] = "9" * 40 elif change == "wrong_stage": modified = {**receipt, "status": "prepared"} module._write_private_atomic(module._build_launcher_receipt_path(), module._canonical_bytes(modified)) scoped["pending_build_launcher"] = modified elif change == "running_candidate": candidate["State"]["Running"] = True elif change == "foreign_candidate_id": candidate["Id"] = "9" * 64 elif change == "wrong_candidate_image": candidate["Image"] = module.RUNNER_IMAGE_ID elif change == "wrong_operation_label": candidate["Config"]["Labels"]["tech.itworx.ci.launcher-operation"] = "9" * 64 elif change == "changed_mount": candidate["Mounts"][1]["RW"] = True elif change == "running_backup": containers[receipt["backupName"]]["State"]["Running"] = True else: containers["itworx-ci-backend"]["HostConfig"]["Memory"] += 1 try: assert module._verify_specialized_outer_images(selected, **scoped) is False assert effects == effects_before finally: containers.clear() containers.update(original_containers) release.clear() release.update(original_release) module._write_private_atomic(module._build_launcher_receipt_path(), module._canonical_bytes(original_receipt)) return module._verify_specialized_outer_images(profiles, **kwargs) monkeypatch.setattr(module, "verify_profile_task_images", checked) preflight = module._build_launcher_preflight("refresh") result = module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert result["status"] == "applied" def test_final_created_snapshot_refuses_other_runner_identity_change(launcher, monkeypatch): module, containers, _, _, _, _ = launcher def check_then_drift(profiles, **kwargs): valid = module._verify_specialized_outer_images(profiles, **kwargs) if kwargs.get("pending_build_launcher") is not None: containers["itworx-ci-backend"]["Id"] = "9" * 64 return valid monkeypatch.setattr(module, "verify_profile_task_images", check_then_drift) preflight = module._build_launcher_preflight("refresh") with pytest.raises(module.RunnerFleetSpecializeError, match="runner_build_launcher_outcome_unknown"): module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert module._build_launcher_receipt()["status"] == "rollback_unknown" receipt = module._build_launcher_receipt() diagnostic = json.loads((module.SPECIALIZED_STATE_ROOT / ("build-launcher-failure-" + receipt["operationId"] + ".json")).read_bytes()) assert diagnostic["failureStage"] == "final_fences" assert diagnostic["failureCode"] == "runner_build_launcher_recovery_fleet_changed" @pytest.mark.parametrize("failure,expected_code", [ ("runner_build_launcher_postcondition_failed", "runner_build_launcher_postcondition_failed"), ("synthetic_private_detail_not_for_receipts", "runner_build_launcher_internal_failure"), ]) def test_private_rollback_diagnostics_retain_only_finite_codes(launcher, monkeypatch, failure, expected_code): module, _, _, _, _, _ = launcher def fail_after_creation(_profiles, **kwargs): if kwargs.get("pending_build_launcher") is not None: raise RuntimeError(failure) return True monkeypatch.setattr(module, "verify_profile_task_images", fail_after_creation) preflight = module._build_launcher_preflight("refresh") with pytest.raises(module.RunnerFleetSpecializeError, match="runner_build_launcher_failed_rolled_back"): module._apply_build_launcher("refresh", preflight["preflightSha256"]) receipt = module._build_launcher_receipt() assert receipt["status"] == "rolled_back" diagnostic = json.loads((module.SPECIALIZED_STATE_ROOT / ("build-launcher-failure-" + receipt["operationId"] + ".json")).read_bytes()) assert diagnostic["failureStage"] == "task_images" assert diagnostic["failureCode"] == expected_code assert "failureStage" not in receipt and "failureCode" not in receipt assert "synthetic_private_detail_not_for_receipts" not in module._build_launcher_receipt_path().read_text() def test_diagnostic_storage_failure_does_not_prevent_guarded_rollback(launcher, monkeypatch): module, containers, _, _, _, _ = launcher before = copy.deepcopy(containers) original_write = module._write_private_atomic def write_without_diagnostics(path, raw): if path.name.startswith("build-launcher-failure-"): raise OSError("synthetic_private_storage_detail") original_write(path, raw) def fail_after_creation(_profiles, **kwargs): return kwargs.get("pending_build_launcher") is None monkeypatch.setattr(module, "_write_private_atomic", write_without_diagnostics) monkeypatch.setattr(module, "verify_profile_task_images", fail_after_creation) preflight = module._build_launcher_preflight("refresh") with pytest.raises(module.RunnerFleetSpecializeError, match="runner_build_launcher_failed_rolled_back"): module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert module._build_launcher_receipt()["status"] == "rolled_back" assert containers == before assert not list(module.SPECIALIZED_STATE_ROOT.glob("build-launcher-failure-*.json")) def test_refresh_and_explicit_rollback_preserve_fleet_and_archived_model(launcher): module, containers, rows, _, effects, model = launcher before = copy.deepcopy(containers) original_model = copy.deepcopy(model) registrations = copy.deepcopy(rows) preflight = module._build_launcher_preflight("refresh") assert not effects and preflight["executed"] is False result = module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert result["status"] == "applied" and result["registrationId"] == 76 assert rows == registrations and model == original_model assert effects[0][:3] == ("docker", "rename", before["itworx-ci-build"]["Id"]) assert effects[1][:2] == ("docker", "create") assert ( module._build_launcher_original(module._build_launcher_receipt())["Id"] == before["itworx-ci-build"]["Id"] ) for name in before: if name != "itworx-ci-build": assert containers[name] == before[name] rollback = module._build_launcher_preflight("rollback") result = module._apply_build_launcher("rollback", rollback["preflightSha256"]) assert ( result["status"] == "rolled_back" and result["imageId"] == module.RUNNER_IMAGE_ID ) assert rows == registrations and model == original_model assert module.verified_topology()["ready"] is True @pytest.mark.parametrize( "change", [ "busy", "enabled", "running", "registration", "image", "pending", "bootstrap", "private_jobs", "guard", "host", "source", ], ) def test_preflight_refuses_defective_conditions_without_effects( launcher, monkeypatch, change ): module, containers, rows, admission, effects, _ = launcher row, item = rows["itworx-ci-build"], containers["itworx-ci-build"] if change == "busy": row["busy"] = True elif change == "enabled": row["disabled"] = False elif change == "running": item["State"]["Running"] = True elif change == "registration": row["id"] = 99 elif change == "image": item["Image"] = module.BUILD_LAUNCHER_IMAGE elif change == "pending": admission["pending"] = {"profileId": "build"} elif change == "bootstrap": admission["bootstrapReceiptSha256"] = None elif change == "private_jobs": monkeypatch.setattr(module, "_run", lambda *_a, **_k: "some-container") elif change == "guard": monkeypatch.setattr(module, "_guard_state_unchanged", lambda _j: False) else: def refuse(*_a, **_k): raise module.RunnerFleetSpecializeError("fixed-refusal") monkeypatch.setattr( module, "_fresh_recovery_host_gate" if change == "host" else "_active_release_bundle", refuse, ) with pytest.raises(module.RunnerFleetSpecializeError): module._build_launcher_preflight("refresh") assert not effects and not module._build_launcher_receipt_path().exists() def test_preflight_cas_refuses_new_admission_revision(launcher): module, _, _, state, effects, _ = launcher preflight = module._build_launcher_preflight("refresh") state["revision"] += 1 with pytest.raises(module.RunnerFleetSpecializeError, match="preflight_changed"): module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert not effects and not module._build_launcher_receipt_path().exists() @pytest.mark.parametrize("foreign", [False, True]) def test_failed_replacement_restores_only_owned_idle_identity( launcher, monkeypatch, foreign ): module, containers, _, _, effects, _ = launcher original = module._run def fail(args, **kwargs): result = original(args, **kwargs) if args[:2] == ["docker", "create"]: if foreign: containers["itworx-ci-build"]["HostConfig"]["Privileged"] = True raise RuntimeError("Bearer secret-must-never-escape") return result monkeypatch.setattr(module, "_run", fail) preflight = module._build_launcher_preflight("refresh") with pytest.raises( module.RunnerFleetSpecializeError, match="outcome_unknown" if foreign else "failed_rolled_back", ): module._apply_build_launcher("refresh", preflight["preflightSha256"]) assert len(effects) == (2 if foreign else 4) assert module._build_launcher_receipt()["status"] == ( "rollback_unknown" if foreign else "rolled_back" ) assert module.verified_topology()["ready"] is (not foreign) def test_variant_is_bound_only_to_build_and_exact_receipt_cid(launcher): module, containers, _, _, _, _ = launcher module._apply_build_launcher( "refresh", module._build_launcher_preflight("refresh")["preflightSha256"] ) for name, item in containers.items(): altered = copy.deepcopy(item) altered["Image"] = module.BUILD_LAUNCHER_IMAGE altered["Config"]["Image"] = module.BUILD_LAUNCHER_IMAGE profile = name.removeprefix("itworx-ci-") assert module._validate_outer_item(name, altered, profile) is ( profile == "build" ) containers["itworx-ci-build"]["Id"] = "f" * 64 assert module.verified_topology()["ready"] is False @pytest.mark.parametrize( "field,value", [ ("patchSha256", "e" * 64), ("markerSha256", "e" * 64), ("registrationId", 99), ("status", "fake"), ], ) def test_corrupt_receipt_fails_closed(launcher, field, value): module, *_ = launcher module._apply_build_launcher( "refresh", module._build_launcher_preflight("refresh")["preflightSha256"] ) path = module._build_launcher_receipt_path() receipt = json.loads(path.read_bytes()) receipt[field] = value path.write_text(json.dumps(receipt)) assert module.verified_topology()["ready"] is False def test_launcher_cli_redacts_unexpected_error_and_requires_exact_confirmation( launcher, monkeypatch, capsys ): module, *_ = launcher def refused(*_a): raise RuntimeError("Bearer secret-must-never-escape") monkeypatch.setattr(module, "_build_launcher_preflight", refused) assert module._main(("--preflight-build-launcher-refresh",)) == 1 assert "secret-must-never-escape" not in capsys.readouterr().out assert ( module._main( ( "--apply-build-launcher-refresh", "--expected-preflight-sha256", "a" * 64, "--confirm", "wrong", ) ) == 64 ) def test_provider_patch_matches_controller_pin(launcher): module, *_ = launcher patch = ( Path(__file__).resolve().parents[2] / "ops/runner-launcher/trusted-systempaths.patch" ) assert ( hashlib.sha256(patch.read_bytes()).hexdigest() == module.BUILD_LAUNCHER_PATCH_SHA256 ) class Interrupted(BaseException): pass @pytest.mark.parametrize("operation", ["refresh", "recovery"]) @pytest.mark.parametrize("inventory", ["stopped_container", "connection_failure"]) def test_launcher_inventory_uses_trusted_socket_and_fails_closed( launcher, monkeypatch, operation, inventory ): module, containers, rows, _, effects, _ = launcher if operation == "recovery": preflight = module._build_launcher_preflight("refresh") module._apply_build_launcher("refresh", preflight["preflightSha256"]) receipt = module._build_launcher_receipt() receipt["status"] = "created" module._write_private_atomic(module._build_launcher_receipt_path(), module._canonical_bytes(receipt)) before = copy.deepcopy(containers), copy.deepcopy(rows) effects.clear() original = module._run def inventory_result(args, **kwargs): if args[:2] == ["docker", "exec"]: assert args == [ "docker", "exec", "-e", "DOCKER_HOST=" + module._DIND_DOCKER_HOST, module.DIND_NAME, "docker", "ps", "-a", "-q", ] if inventory == "connection_failure": raise module.RunnerFleetSpecializeError("runner_fleet_specialization_command_failed") return "f" * 64 return original(args, **kwargs) monkeypatch.setattr(module, "_run", inventory_result) with pytest.raises(module.RunnerFleetSpecializeError): if operation == "recovery": module._build_launcher_recovery_preflight() else: module._build_launcher_preflight("refresh") assert effects == [] assert (containers, rows) == before @pytest.mark.parametrize( "point", ["backup", "prepared", "rename", "create", "created", "before_applied", "applied"], ) def test_each_durable_refresh_boundary_has_supported_recovery( launcher, monkeypatch, point ): module, containers, rows, _, effects, _ = launcher original_id = containers["itworx-ci-build"]["Id"] original_rows = copy.deepcopy(rows) write, docker = module._write_private_atomic, module._run def interrupt_write(path, raw): status = json.loads(raw).get("status") if point == "before_applied" and status == "applied": raise Interrupted() write(path, raw) if ( point == "backup" and path.name.startswith("build-launcher-before-") or point == status ): raise Interrupted() def interrupt_docker(args, **kwargs): result = docker(args, **kwargs) if args[:2] == ["docker", point]: raise Interrupted() return result monkeypatch.setattr(module, "_write_private_atomic", interrupt_write) monkeypatch.setattr(module, "_run", interrupt_docker) preflight = module._build_launcher_preflight("refresh") with pytest.raises(Interrupted): module._apply_build_launcher("refresh", preflight["preflightSha256"]) monkeypatch.setattr(module, "_write_private_atomic", write) monkeypatch.setattr(module, "_run", docker) if point == "backup": assert not effects and module._build_launcher_receipt() is None module._apply_build_launcher("refresh", preflight["preflightSha256"]) point = "applied" if point == "applied": rollback = module._build_launcher_preflight("rollback") module._apply_build_launcher("rollback", rollback["preflightSha256"]) else: recovery = module._build_launcher_recovery_preflight() assert recovery["executed"] is False module._recover_build_launcher(recovery["preflightSha256"]) assert containers["itworx-ci-build"]["Id"] == original_id assert module._build_launcher_receipt()["status"] == "rolled_back" assert rows == original_rows and module.verified_topology()["ready"] is True @pytest.mark.parametrize( "point", ["rolling_back", "rm", "rename", "before_rolled_back"] ) def test_each_rollback_boundary_recovers_original_instance( launcher, monkeypatch, point ): module, containers, _, _, _, _ = launcher original_id = containers["itworx-ci-build"]["Id"] module._apply_build_launcher( "refresh", module._build_launcher_preflight("refresh")["preflightSha256"] ) write, docker = module._write_private_atomic, module._run def interrupt_write(path, raw): status = json.loads(raw).get("status") if point == "before_rolled_back" and status == "rolled_back": raise Interrupted() write(path, raw) if point == status: raise Interrupted() def interrupt_docker(args, **kwargs): result = docker(args, **kwargs) if args[:2] == ["docker", point]: raise Interrupted() return result monkeypatch.setattr(module, "_write_private_atomic", interrupt_write) monkeypatch.setattr(module, "_run", interrupt_docker) rollback = module._build_launcher_preflight("rollback") with pytest.raises(Interrupted): module._apply_build_launcher("rollback", rollback["preflightSha256"]) monkeypatch.setattr(module, "_write_private_atomic", write) monkeypatch.setattr(module, "_run", docker) recovery = module._build_launcher_recovery_preflight() module._recover_build_launcher(recovery["preflightSha256"]) assert containers["itworx-ci-build"]["Id"] == original_id assert module.verified_topology()["ready"] is True def test_foreign_same_shape_and_image_with_new_cid_is_never_removed( launcher, monkeypatch ): module, containers, _, _, effects, _ = launcher original_verify = module._resource_proof def replace_and_refuse(*args): original_verify(*args) containers["itworx-ci-build"]["Id"] = "f" * 64 raise RuntimeError("postcondition failure") monkeypatch.setattr(module, "_resource_proof", replace_and_refuse) with pytest.raises(module.RunnerFleetSpecializeError, match="outcome_unknown"): module._apply_build_launcher( "refresh", module._build_launcher_preflight("refresh")["preflightSha256"] ) assert containers["itworx-ci-build"]["Id"] == "f" * 64 assert not any(args[:2] == ("docker", "rm") for args in effects) with pytest.raises(module.RunnerFleetSpecializeError, match="foreign_container"): module._build_launcher_recovery_preflight()