from __future__ import annotations import ast import io import json import os import re import shutil import subprocess import sys import tarfile from dataclasses import replace from pathlib import Path from typing import Any import pytest from scripts import bootstrap_controller_artifact_promoter as bootstrap ROOT = Path(__file__).resolve().parents[2] LEGACY_REVISION = "5553c04ce497c08a35ed301052dc5e0b02e5d9b4" LEGACY_TREE = "1624e1748053f998d8427c8c8a4203277d22f13c" APPOPS_CONTAINER_ID = "72daa9f638defc4f8897d8aad23c4181054b2a1bc04c993078a5ac9597eeb0ec" APPOPS_IMAGE_ID = "sha256:9b5eddf32c8eb5361106fc35cc0e056c49d287a20d1fb26490662cc8447f09d4" CONTROLLER_CONTAINER_ID = "e4e7550405aa2a9a91901faf1177d46a35b0fc43c304976ec8971865d43e30eb" CONTROLLER_IMAGE_ID = "sha256:c474d435a4f55ac5205570da0caa47e103f89a6c2ef96ea45861a23eb7b005dc" CONTROLLER_IMAGE_REF = "docker.gitea.com/runner-images:ubuntu-latest" CONTROLLER_REVISION = "034c9f844d2455850ba37b31379b7edaa786583e" CONTROLLER_TREE = "6b8e92942e6e5eca8a9aa9da0a1509b2019d7b7b" MODERN_CONTROLLER_CONTAINER_ID = "9093c8e29d9064f9948918e5884f9adc5db6b740991e2934902f93c6ea813e01" MODERN_CONTROLLER_REVISION = "e854334923cca574482d22b8424ba3ce46ec6aa1" MODERN_CONTROLLER_TREE = "cf2a117f9222b952deb9902c646116612f3a7ea0" MODERN_CONTROLLER_ARTIFACT_SHA256 = "e88739bf091c8bac77ea0db5b97300666a1baa8ba984bbf6ae91e3bf25719e08" REVISION = "a" * 40 TREE = "b" * 40 CALLER_ID = "c" * 64 MODERN_HELPER_ID = "f" * 64 RETIRED_HELPER_ID = bootstrap.RETIRED_FAILED_MODERN_HELPER_ID HOST_HOSTNAME = "Tower" RETIRED_MODERN_HELPER_SOURCE = (ROOT / "backend/tests/fixtures/controller_artifact_helper_spec_2_1_14.txt").read_text(encoding="utf-8") LEGACY_HELPER_SOURCE_FIXTURE = ''' def helper_container_spec(*, revision: str, image_id: str) -> dict[str, Any]: if not _REVISION.fullmatch(revision): raise ValueError("controller_promoter_revision_invalid") if not _IMAGE_ID.fullmatch(image_id): raise ValueError("controller_promoter_image_invalid") command = f"""set -Eeuo pipefail revision='{revision}' version='{CONTROLLER_VERSION}' repo='/gitea-repositories/jens/unraid-appops-gateway.git' work='/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-{revision[:12]}' rm -rf -- "$work" install -d -m 0700 "$work" cleanup() {{ rm -rf -- "$work"; }} trap cleanup EXIT command -v git >/dev/null command -v docker >/dev/null docker compose version >/dev/null command -v python3 >/dev/null command -v tar >/dev/null command -v curl >/dev/null git --git-dir="$repo" cat-file -e "$revision^{{commit}}" tree="$(git --git-dir="$repo" rev-parse "$revision^{{tree}}")" git --git-dir="$repo" archive --format=tar "$revision" \\ ops/gitea-deploy-broker \\ scripts/prepare_controller_artifact.py \\ scripts/reconcile_controller_state.py \\ scripts/run_repopulse_adoption.py > "$work/controller.tar" archive_sha="$(sha256sum "$work/controller.tar" | cut -d ' ' -f 1)" git --git-dir="$repo" show "$revision:scripts/upgrade_deployment_controller.sh" > "$work/upgrade.sh" chmod 0700 "$work/upgrade.sh" bash "$work/upgrade.sh" \\ --archive "$work/controller.tar" \\ --archive-sha256 "$archive_sha" \\ --source-repository '{SOURCE_REPOSITORY}' \\ --source-revision "$revision" \\ --source-tree "$tree" \\ --controller-version "$version" \\ --confirm UPGRADE_DEPLOYMENT_CONTROLLER_V2 """ return { "Image": image_id, "User": "0:0", "Cmd": ["/bin/bash", "-lc", command], "Labels": { HELPER_LABEL: "true", "itworx.source.repository": SOURCE_REPOSITORY, "itworx.source.revision": revision, "itworx.controller.version": CONTROLLER_VERSION, }, "HostConfig": { "Binds": [ "/mnt/cache:/mnt/cache", "/mnt/user:/mnt/user", "/srv/example/appdata/gitea/git/repositories:/gitea-repositories:ro", "/var/run/docker.sock:/var/run/docker.sock", ], "NetworkMode": "host", "Privileged": False, "SecurityOpt": ["no-new-privileges:true"], "RestartPolicy": {"Name": "no"}, }, } ''' def _legacy_helper_spec(revision: str = REVISION) -> dict[str, Any]: tree = ast.parse(LEGACY_HELPER_SOURCE_FIXTURE) function = next( node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "helper_container_spec" ) namespace: dict[str, Any] = { "re": re, "Any": Any, "_REVISION": re.compile(r"^[0-9a-f]{40}$"), "_IMAGE_ID": re.compile(r"^sha256:[0-9a-f]{64}$"), "SOURCE_REPOSITORY": "jens/unraid-appops-gateway", "HELPER_NAME": "itworx-controller-artifact-promoter", "CONTROLLER_VERSION": "2.1.5", "HELPER_LABEL": "itworx.controller.promoter", } exec( compile(ast.Module(body=[function], type_ignores=[]), "legacy-promoter", "exec"), namespace, ) return namespace["helper_container_spec"](revision=revision, image_id=CONTROLLER_IMAGE_ID) def _image(labels: dict[str, str]) -> dict[str, Any]: return {"Id": APPOPS_IMAGE_ID, "Config": {"Labels": dict(labels)}} def _appops_labels() -> dict[str, str]: return { "org.opencontainers.image.source": "https://gitea.example.com/Jens/Unraid-AppOps-Gateway", "org.opencontainers.image.revision": LEGACY_REVISION, "org.opencontainers.image.source-tree": LEGACY_TREE, "org.opencontainers.image.version": "1.8.0-platform-5553c04ce497", "org.opencontainers.image.created": "2026-10-01T16:32:01Z", "com.docker.compose.project": "unraid-appops-gateway", "com.docker.compose.service": "appops-gateway", "com.docker.compose.project.working_dir": "/srv/example/appdata/unraid-appops-gateway/source", "com.docker.compose.project.config_files": "/srv/example/appdata/unraid-appops-gateway/source/docker-compose.yml", } def _controller_labels() -> dict[str, str]: return { "io.buildah.version": "1.35.0", "org.opencontainers.image.authors": "Gitea", "org.opencontainers.image.created": "2026-10-01T00:01:57Z", "org.opencontainers.image.description": "Official docker images used by Gitea runner to run workflows.", "org.opencontainers.image.documentation": "https://gitea.com/gitea/runner-images/src/branch/main/README.md", "org.opencontainers.image.licenses": "MIT", "org.opencontainers.image.revision": "317f8e3c510e5ec773a103f5d9a9e5769105c59b", "org.opencontainers.image.source": "https://gitea.com/gitea/runner-images", "org.opencontainers.image.title": "Official Gitea Actions runner images", "org.opencontainers.image.url": "https://gitea.com/gitea/runner-images", "org.opencontainers.image.vendor": "Gitea", "org.opencontainers.image.version": "v26.10.01", } def _controller_container_labels() -> dict[str, str]: return { **_controller_labels(), "com.docker.compose.config-hash": "577d77247471213db8a5c465ee2450765ba52cb4048f983c42083ac45a16ddb7", "com.docker.compose.container-number": "1", "com.docker.compose.depends_on": "", "com.docker.compose.image": "sha256:adbb51b2873bf05546019c341caf6fd4368159546c36a5aea81b4d8bf088804e", "com.docker.compose.oneoff": "False", "com.docker.compose.project": "gitea-deploy-runners", "com.docker.compose.project.config_files": "/mnt/cache/appdata/gitea-deploy-runners/docker-compose.yml", "com.docker.compose.project.working_dir": "/mnt/cache/appdata/gitea-deploy-runners", "com.docker.compose.replace": "gitea-deploy-control", "com.docker.compose.service": "deploy-control", "com.docker.compose.version": "2.40.3", "tech.itworx.deployment.target": "unraid", "tech.itworx.role": "gitea-deploy-controller", "tech.itworx.gitea.scope": "Jens", } def _modern_controller_labels() -> dict[str, str]: return { **_controller_labels(), "com.docker.compose.config-hash": "a" * 64, "com.docker.compose.container-number": "1", "com.docker.compose.depends_on": "", "com.docker.compose.image": CONTROLLER_IMAGE_ID, "com.docker.compose.oneoff": "False", "com.docker.compose.project": "gitea-deploy-runners", "com.docker.compose.project.config_files": "-", "com.docker.compose.project.environment_file": "/dev/null", "com.docker.compose.project.working_dir": "/var/lib/docker/volumes/itworx-controller-hub-releases/_data", "com.docker.compose.replace": "gitea-deploy-control", "com.docker.compose.service": "deploy-control", "com.docker.compose.version": "2.40.3", "tech.itworx.deployment.target": "unraid", "tech.itworx.gitea.scope": "Jens", "tech.itworx.role": "gitea-deploy-controller", } def _modern_controller_mounts(source_revision: str) -> list[dict[str, Any]]: artifact_source = ( "/var/lib/docker/volumes/itworx-controller-hub-releases/_data/" f"controller-artifacts/{source_revision}" ) return [ {"Type": "bind", "Source": "/boot", "Destination": "/boot", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/var/lib/docker", "Destination": "/var/lib/docker", "RW": False, "Propagation": "rslave"}, { "Type": "bind", "Source": "/srv/example/appdata/gitea/git/repositories", "Destination": "/gitea-repositories", "RW": False, "Propagation": "rprivate", }, {"Type": "bind", "Source": artifact_source, "Destination": "/opt/gitea-deploy", "RW": False, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/var/run/docker.sock", "Destination": "/var/run/docker.sock", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/mnt/user", "Destination": "/mnt/user", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/mnt/cache", "Destination": "/mnt/cache", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/mnt/disks", "Destination": "/mnt/disks", "RW": False, "Propagation": "rprivate"}, { "Type": "volume", "Name": "itworx-controller-hub-releases", "Source": "/var/lib/docker/volumes/itworx-controller-hub-releases/_data", "Destination": "/controller-hub-releases", "RW": True, "Propagation": "", }, { "Type": "bind", "Source": "/var/lib/docker/volumes/itworx-controller-hub-releases/_data/controller-config", "Destination": "/controller-config", "RW": False, "Propagation": "rprivate", }, { "Type": "volume", "Name": "itworx-forge-master-key-backup", "Source": "/var/lib/docker/volumes/itworx-forge-master-key-backup/_data", "Destination": "/forge-master-key-backup", "RW": True, "Propagation": "", }, ] def _modern_controller_source_mounts(source_revision: str) -> list[dict[str, Any]]: mounts = [dict(mount) for mount in _modern_controller_mounts(source_revision)] mounts = [mount for mount in mounts if mount["Destination"] != "/forge-master-key-backup"] for mount in mounts: if mount["Destination"] in {"/opt/gitea-deploy", "/controller-config"}: mount["Propagation"] = "rslave" return mounts def _modern_controller_inspect( source_revision: str, *, identifier: str = MODERN_CONTROLLER_CONTAINER_ID, ) -> dict[str, Any]: controller = _controller_inspect() controller.update({"Id": identifier, "Image": CONTROLLER_IMAGE_ID}) controller["State"].update({"Dead": False, "Paused": False, "Restarting": False}) controller["Config"]["Image"] = CONTROLLER_IMAGE_ID controller["Config"]["Labels"] = _modern_controller_labels() controller["Mounts"] = _modern_controller_mounts(source_revision) return controller def _modern_controller_source_inspect( source_revision: str, *, identifier: str = MODERN_CONTROLLER_CONTAINER_ID, ) -> dict[str, Any]: controller = _modern_controller_inspect(source_revision, identifier=identifier) controller["Mounts"] = _modern_controller_source_mounts(source_revision) return controller def _appops_inspect() -> dict[str, Any]: labels = _appops_labels() return { "Id": APPOPS_CONTAINER_ID, "Image": APPOPS_IMAGE_ID, "Name": "/Unraid-AppOps-Gateway", "State": {"Running": True, "Health": {"Status": "healthy"}}, "Config": { "Hostname": "unraid-appops-gateway", "Image": "unraid-appops-gateway:platform-5553c04ce497", "Labels": labels, }, "HostConfig": {"RestartPolicy": {"Name": "unless-stopped"}}, "Mounts": [ { "Type": "bind", "Source": "/srv/example/appdata/unraid-appops-gateway-1216", "Destination": "/data", "RW": True, }, { "Type": "bind", "Source": "/srv/example/appdata/comfyui-nvidia/basedir/models", "Destination": "/mnt/privatemuse-comfyui-models", "RW": True, }, ], } def _controller_inspect() -> dict[str, Any]: return { "Id": CONTROLLER_CONTAINER_ID, "Image": CONTROLLER_IMAGE_ID, "Name": "/gitea-deploy-control", "State": {"Running": True}, "Config": {"Image": CONTROLLER_IMAGE_REF, "Labels": _controller_container_labels()}, "HostConfig": { "Binds": [ "/mnt/user:/mnt/user:rw", "/mnt/cache:/mnt/cache:rw", "/mnt/disks:/mnt/disks:ro", "/boot:/boot:rw", "/var/lib/docker:/var/lib/docker:ro", "/srv/example/appdata/gitea/git/repositories:/gitea-repositories:ro", "/mnt/cache/appdata/gitea-deploy-runners/controller:/opt/gitea-deploy:ro", "/var/run/docker.sock:/var/run/docker.sock:rw", ], "Mounts": None, }, "Mounts": [ { "Type": "bind", "Source": "/boot", "Destination": "/boot", "RW": True, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/var/lib/docker", "Destination": "/var/lib/docker", "RW": False, "Propagation": "rslave", }, { "Type": "bind", "Source": "/srv/example/appdata/gitea/git/repositories", "Destination": "/gitea-repositories", "RW": False, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/mnt/cache/appdata/gitea-deploy-runners/controller", "Destination": "/opt/gitea-deploy", "RW": False, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/var/run/docker.sock", "Destination": "/var/run/docker.sock", "RW": True, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/mnt/user", "Destination": "/mnt/user", "RW": True, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/mnt/cache", "Destination": "/mnt/cache", "RW": True, "Propagation": "rprivate", }, { "Type": "bind", "Source": "/mnt/disks", "Destination": "/mnt/disks", "RW": False, "Propagation": "rprivate", }, ], } def _inspect_from_spec( name: str, identifier: str, image_id: str, spec: dict[str, Any], *, state: dict[str, Any], hostname: str | None = None, ) -> dict[str, Any]: return { "Id": identifier, "Image": image_id, "Name": f"/{name}", "State": dict(state), "Config": { "Image": spec["Image"], "User": spec["User"], "Cmd": spec["Cmd"], "Labels": {**_controller_labels(), **spec["Labels"]}, "Hostname": hostname if hostname is not None else identifier[:12], }, "HostConfig": {"ReadonlyRootfs": False, **spec["HostConfig"]}, } class FakeDocker: def __init__(self, caller_spec: dict[str, Any], *, existing_child: dict[str, Any] | None = None): legacy_caller = _inspect_from_spec( "itworx-controller-artifact-promoter", CALLER_ID, CONTROLLER_IMAGE_ID, caller_spec, state={"Running": True}, hostname=HOST_HOSTNAME, ) self.containers = { APPOPS_CONTAINER_ID: _appops_inspect(), "gitea-deploy-control": _controller_inspect(), bootstrap.LEGACY_HELPER_NAME: legacy_caller, CALLER_ID[:12]: legacy_caller, } self.images = { APPOPS_IMAGE_ID: _image(_appops_labels()), CONTROLLER_IMAGE_ID: { "Id": CONTROLLER_IMAGE_ID, "Config": {"Labels": _controller_labels()}, }, } self.existing_child = existing_child self.created: list[dict[str, Any]] = [] self.started: list[str] = [] self.removed: list[str] = [] self.inspected: list[str] = [] self.host_name = HOST_HOSTNAME self.log_bytes = b"controller_v2_ready revision=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n" self.named_inspect_override: dict[str, Any] | None = None self.named_inspect_override_at = 0 self.retired_id_inspect_override: dict[str, Any] | None = None self.appops_inspect_count = 0 self.appops_inspect_override: dict[str, Any] | None = None self.appops_inspect_override_at = 0 self._inspect_counts: dict[str, int] = {} def inspect_container(self, reference: str) -> dict[str, Any] | None: self.inspected.append(reference) self._inspect_counts[reference] = self._inspect_counts.get(reference, 0) + 1 if reference == "itworx-controller-artifact-promoter-modern": if ( self.named_inspect_override is not None and self._inspect_counts[reference] == self.named_inspect_override_at ): return self.named_inspect_override return self.existing_child if ( reference == RETIRED_HELPER_ID and self.retired_id_inspect_override is not None ): return self.retired_id_inspect_override if ( reference == RETIRED_HELPER_ID and self.existing_child is not None and self.existing_child.get("Id") == RETIRED_HELPER_ID ): return self.existing_child if reference == APPOPS_CONTAINER_ID: self.appops_inspect_count += 1 if self.appops_inspect_count == self.appops_inspect_override_at: return self.appops_inspect_override return self.containers.get(reference) def daemon_name(self) -> str: return self.host_name def inspect_image(self, image_id: str) -> dict[str, Any] | None: return self.images.get(image_id) def create_container(self, name: str, spec: dict[str, Any]) -> str: assert name == "itworx-controller-artifact-promoter-modern" self.created.append(spec) container = _inspect_from_spec( name, MODERN_HELPER_ID, spec["Image"], spec, state={"Running": False, "ExitCode": 0, "Status": "exited"}, ) self.containers[name] = container self.containers[MODERN_HELPER_ID] = container return MODERN_HELPER_ID def start_container(self, identifier: str) -> None: self.started.append(identifier) source_revision = self.created[-1]["Labels"]["itworx.source.revision"] self.containers["gitea-deploy-control"] = _modern_controller_inspect( source_revision, identifier="e" * 64 ) def remove_retired_failed_container(self, identifier: str) -> None: assert identifier == RETIRED_HELPER_ID assert self.existing_child is not None assert self.existing_child["Id"] == RETIRED_HELPER_ID self.removed.append(identifier) self.existing_child = None def container_logs(self, identifier: str, *, tail: int, max_bytes: int) -> bytes: assert identifier in {MODERN_HELPER_ID, RETIRED_HELPER_ID} assert tail <= 40 assert max_bytes <= 8192 return self.log_bytes[:max_bytes] class FakeProtectedGit: def __init__(self, helper_source: str): self.revision = REVISION self.tree = TREE self.helper_source = helper_source self.blobs_read: list[tuple[str, str]] = [] self.main_checks = 0 self.fail_current_main_at: int | None = None self.retired_tree_valid = True self.controller_release_checks: list[bootstrap.ControllerBaseline] = [] self.invalid_controller_release: bootstrap.ControllerBaseline | None = None def verify_current_main(self, revision: str, tree: str) -> None: self.main_checks += 1 if self.fail_current_main_at == self.main_checks: raise bootstrap.BootstrapError("controller_promoter_source_not_current_main") if (revision, tree) != (self.revision, self.tree): raise bootstrap.BootstrapError("controller_promoter_source_not_current_main") def verify_legacy_revision(self, revision: str, tree: str) -> None: if (revision, tree) != (LEGACY_REVISION, LEGACY_TREE): raise bootstrap.BootstrapError("controller_promoter_legacy_provenance_invalid") def verify_controller_release(self, baseline: bootstrap.ControllerBaseline) -> None: self.controller_release_checks.append(baseline) if baseline not in bootstrap.CONTROLLER_BASELINES or baseline == self.invalid_controller_release: raise bootstrap.BootstrapError("controller_promoter_controller_provenance_invalid") def verify_retired_modern_revision(self, revision: str, tree: str) -> None: if (revision, tree) != ( bootstrap.RETIRED_FAILED_MODERN_REVISION, bootstrap.RETIRED_FAILED_MODERN_TREE, ) or not self.retired_tree_valid: raise bootstrap.BootstrapError( "controller_promoter_retired_helper_provenance_invalid" ) def read_blob(self, revision: str, path: str) -> bytes: self.blobs_read.append((revision, path)) if revision not in {self.revision, bootstrap.RETIRED_FAILED_MODERN_REVISION} or path != "backend/app/services/controller_artifact_helper_spec.py": raise bootstrap.BootstrapError("controller_promoter_source_blob_invalid") if revision == bootstrap.RETIRED_FAILED_MODERN_REVISION: return RETIRED_MODERN_HELPER_SOURCE.encode("utf-8") return self.helper_source.encode("utf-8") def read_legacy_blob(self, revision: str, path: str) -> bytes: self.blobs_read.append((revision, path)) if (revision, path) != ( LEGACY_REVISION, "backend/app/services/controller_artifact_promoter.py", ): raise bootstrap.BootstrapError("controller_promoter_legacy_blob_invalid") return LEGACY_HELPER_SOURCE_FIXTURE.encode("utf-8") def _health( *, revision: str = REVISION, tree: str = TREE, version: str = "2.1.21", artifact_sha256: str = "1" * 64, ) -> dict[str, Any]: return { "status": "ok", "protocolVersion": "itworx-controller-request/v2", "artifact": { "apiVersion": "itworx.controller-artifact/v1", "sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": revision, "sourceTree": tree, "controllerVersion": version, "artifactSha256": artifact_sha256, }, "delivery": { "health": { "configuration": "healthy", "inbox": "healthy", "artifact": "healthy", } }, } def _modern_baseline_health() -> dict[str, Any]: return _health( revision=MODERN_CONTROLLER_REVISION, tree=MODERN_CONTROLLER_TREE, version="2.1.14", artifact_sha256=MODERN_CONTROLLER_ARTIFACT_SHA256, ) def _run_bootstrap( docker: FakeDocker, git: FakeProtectedGit, *, health: dict[str, Any] | None = None, initial_health: dict[str, Any] | None = None, health_sequence: list[dict[str, Any]] | None = None, hostname: str = HOST_HOSTNAME, source_revision: str = REVISION, source_tree: str = TREE, ): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) git.helper_source = source initial = initial_health or _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) health_reads = health_sequence or [initial, health or _health()] def read_health() -> dict[str, Any]: return health_reads.pop(0) if health_reads else health or _health() return bootstrap.run_bootstrap( "jens/unraid-appops-gateway", source_revision, source_tree, docker=docker, protected_git=git, hostname=hostname, health_reader=read_health, monotonic=lambda: 0.0, sleeper=lambda _seconds: None, child_timeout_seconds=0.0, ) def test_protected_git_reproduces_pinned_release_with_exact_handoff_crlf_conversion( monkeypatch: pytest.MonkeyPatch, ): git_executable = shutil.which("git") if git_executable is None: pytest.skip("git is required to validate the pinned source archives") probe = subprocess.run( [git_executable, "cat-file", "-e", f"{CONTROLLER_REVISION}^{{commit}}"], cwd=ROOT, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15, check=False, ) if probe.returncode != 0: pytest.skip("the checkout does not expose the retained historical commits to Git") def local_git_output(arguments: list[str], *, limit: int = bootstrap.MAX_GIT_BLOB_BYTES) -> bytes: result = subprocess.run( [git_executable, "-c", f"safe.directory={ROOT}", *arguments], cwd=ROOT, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=15, check=False, ) if result.returncode != 0: raise bootstrap.BootstrapError("controller_promoter_protected_git_invalid") if len(result.stdout) > limit: raise bootstrap.BootstrapError("controller_promoter_protected_git_invalid") return result.stdout monkeypatch.setattr(bootstrap, "_git_output", local_git_output) repository = bootstrap.ProtectedGitRepository() for baseline in ( bootstrap.LEGACY_CONTROLLER_BASELINE, bootstrap.MODERN_CONTROLLER_BASELINE, ): repository.verify_controller_release(baseline) with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): repository.verify_controller_release( replace(baseline, tree="f" * 40) ) archive_paths = [ "scripts/run_repopulse_adoption.py" if name == "run_repopulse_adoption.py" else f"ops/gitea-deploy-broker/{name}" for name in baseline.artifact_files ] archive = local_git_output( ["archive", "--format=tar", baseline.revision, "--", *archive_paths], limit=bootstrap.MAX_CONTROLLER_ARTIFACT_ARCHIVE_BYTES, ) assert bootstrap._controller_artifact_archive_digest( archive, baseline ) == baseline.artifact_sha256 with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): bootstrap._controller_artifact_source_bytes( replace(baseline, revision="f" * 40), bootstrap.CONTROLLER_CRLF_HANDOFF, b"source", ) corrupted_archive = _rewrite_controller_archive( archive, mutate_path="ops/gitea-deploy-broker/deploy.py", mutate=lambda content: content + b"# changed\n", ) assert bootstrap._controller_artifact_archive_digest( corrupted_archive, baseline ) != baseline.artifact_sha256 renamed_archive = _rewrite_controller_archive( archive, rename_path="scripts/run_repopulse_adoption.py", new_path="scripts/unreviewed.py", ) with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): bootstrap._controller_artifact_archive_digest(renamed_archive, baseline) handoff_path = "scripts/run_repopulse_adoption.py" corrupted_handoff = _rewrite_controller_archive( archive, mutate_path=handoff_path, mutate=lambda content: content + b"# changed\n", ) with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): bootstrap._controller_artifact_archive_digest(corrupted_handoff, baseline) crlf_source_archive = _rewrite_controller_archive( archive, mutate_path=handoff_path, mutate=lambda content: content.replace(b"\n", b"\r\r\n"), ) with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): bootstrap._controller_artifact_archive_digest(crlf_source_archive, baseline) def _rewrite_controller_archive( archive: bytes, *, rename_path: str | None = None, new_path: str | None = None, mutate_path: str | None = None, mutate: Any | None = None, ) -> bytes: output = io.BytesIO() with tarfile.open(fileobj=io.BytesIO(archive), mode="r:") as original: with tarfile.open(fileobj=output, mode="w:") as rewritten: for member in original.getmembers(): if member.isdir(): rewritten.addfile(member) continue body = original.extractfile(member) assert body is not None content = body.read() if member.name == rename_path: assert new_path is not None member.name = new_path if member.name == mutate_path: assert mutate is not None content = mutate(content) member.size = len(content) rewritten.addfile(member, io.BytesIO(content)) return output.getvalue() def test_controller_source_provenance_failure_precedes_health_and_side_effects(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) docker = FakeDocker(_legacy_helper_spec()) git = FakeProtectedGit(helper_source) git.invalid_controller_release = bootstrap.MODERN_CONTROLLER_BASELINE with pytest.raises( bootstrap.BootstrapError, match="controller_promoter_controller_provenance_invalid", ): _run_bootstrap( docker, git, health_sequence=[_modern_baseline_health()], ) assert git.controller_release_checks == [bootstrap.MODERN_CONTROLLER_BASELINE] assert docker.removed == [] assert docker.created == [] assert docker.started == [] def _retired_failed_child() -> dict[str, Any]: spec = bootstrap.render_modern_helper_spec( RETIRED_MODERN_HELPER_SOURCE, revision=bootstrap.RETIRED_FAILED_MODERN_REVISION, image_id=CONTROLLER_IMAGE_ID, current_controller_id=CONTROLLER_CONTAINER_ID, current_controller_revision=CONTROLLER_REVISION, current_controller_tree=CONTROLLER_TREE, current_controller_version="2.1.12", helper_version="2.1.14", ) child = _inspect_from_spec( "itworx-controller-artifact-promoter-modern", RETIRED_HELPER_ID, CONTROLLER_IMAGE_ID, spec, state={"Running": False, "ExitCode": 1, "Status": "exited"}, ) child["Config"]["Volumes"] = None child["Mounts"] = [ {"Type": "bind", "Source": "/mnt/cache", "Destination": "/mnt/cache", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/mnt/user", "Destination": "/mnt/user", "RW": True, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/var/lib/docker", "Destination": "/var/lib/docker", "RW": False, "Propagation": "rslave"}, {"Type": "bind", "Source": "/srv/example/appdata/gitea/git/repositories", "Destination": "/gitea-repositories", "RW": False, "Propagation": "rprivate"}, {"Type": "bind", "Source": "/var/run/docker.sock", "Destination": "/var/run/docker.sock", "RW": True, "Propagation": "rprivate"}, ] child["HostConfig"].update( { "Mounts": None, "CapAdd": None, "CapDrop": None, "Devices": None, "DeviceRequests": None, "PortBindings": None, "VolumesFrom": None, "PidMode": "", "UTSMode": "", "UsernsMode": "", "CgroupParent": "", "IpcMode": "private", "CgroupnsMode": "private", "Runtime": "runc", "AutoRemove": False, "PublishAllPorts": False, } ) return child def test_initial_controller_runtime_accepts_exact_current_compose_container(): docker = FakeDocker(_legacy_helper_spec()) base_image_labels = bootstrap._inspect_controller_base_image(docker) health = _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) assert ( bootstrap._verify_controller_runtime( docker, health_payload=health, base_image_labels=base_image_labels, ) == base_image_labels ) def test_modern_source_baseline_accepts_its_fixed_source_mount_profile(): docker = FakeDocker(_legacy_helper_spec()) docker.containers["gitea-deploy-control"] = _modern_controller_source_inspect( MODERN_CONTROLLER_REVISION ) base_image_labels = bootstrap._inspect_controller_base_image(docker) assert ( bootstrap._verify_controller_runtime( docker, health_payload=_modern_baseline_health(), base_image_labels=base_image_labels, ) == base_image_labels ) def test_modern_source_mount_profile_is_distinct_from_postpromotion_topology(): source_mounts = _modern_controller_source_mounts(MODERN_CONTROLLER_REVISION) desired_mounts = _modern_controller_mounts(MODERN_CONTROLLER_REVISION) assert len(source_mounts) == 10 assert len(desired_mounts) == 11 assert bootstrap._modern_controller_source_mounts_match( source_mounts, MODERN_CONTROLLER_REVISION ) assert not bootstrap._modern_controller_source_mounts_match( desired_mounts, MODERN_CONTROLLER_REVISION ) assert not bootstrap._modern_controller_source_mounts_match( source_mounts, "9" * 40 ) assert bootstrap._modern_controller_mounts_match( desired_mounts, MODERN_CONTROLLER_REVISION ) docker = FakeDocker(_legacy_helper_spec()) docker.containers["gitea-deploy-control"] = _modern_controller_source_inspect( MODERN_CONTROLLER_REVISION ) assert not bootstrap._verified_live_controller( docker, source_revision=MODERN_CONTROLLER_REVISION ) @pytest.mark.parametrize( "mutation", [ "missing_mount", "extra_backup_mount", "artifact_propagation", "config_propagation", "integer_rw", "duplicate_mount", ], ) def test_modern_source_mount_profile_rejects_topology_drift(mutation: str): mounts = _modern_controller_source_mounts(MODERN_CONTROLLER_REVISION) if mutation == "missing_mount": mounts.pop() elif mutation == "extra_backup_mount": backup = next( mount for mount in _modern_controller_mounts(MODERN_CONTROLLER_REVISION) if mount["Destination"] == "/forge-master-key-backup" ) mounts.append(backup) elif mutation == "artifact_propagation": mounts[3]["Propagation"] = "rprivate" elif mutation == "config_propagation": mounts[9]["Propagation"] = "rprivate" elif mutation == "integer_rw": mounts[1]["RW"] = 0 elif mutation == "duplicate_mount": mounts.append(dict(mounts[0])) assert not bootstrap._modern_controller_source_mounts_match( mounts, MODERN_CONTROLLER_REVISION ) def test_immutable_legacy_helper_source_fixture_passes_bootstrap_identity_guard(): legacy_spec = _legacy_helper_spec() docker = FakeDocker(legacy_spec) git = FakeProtectedGit("") result = _run_bootstrap(docker, git) assert result["status"] == "promoted" assert docker.created assert git.blobs_read == [ (REVISION, "backend/app/services/controller_artifact_helper_spec.py"), (LEGACY_REVISION, "backend/app/services/controller_artifact_promoter.py"), ] def test_host_network_legacy_caller_is_resolved_by_fixed_name_not_host_hostname(): docker = FakeDocker(_legacy_helper_spec()) git = FakeProtectedGit("") result = _run_bootstrap(docker, git, hostname=HOST_HOSTNAME) assert result["status"] == "promoted" assert bootstrap.LEGACY_HELPER_NAME in docker.inspected assert HOST_HOSTNAME not in docker.inspected assert docker.created def test_docker_info_hostname_reader_returns_only_bounded_name(monkeypatch: pytest.MonkeyPatch): client = object.__new__(bootstrap.DockerSocketClient) requests: list[tuple[str, str, int]] = [] def fake_request(method: str, path: str, *, max_bytes: int) -> tuple[int, bytes]: requests.append((method, path, max_bytes)) return 200, json.dumps({"Name": HOST_HOSTNAME, "ID": "must-not-be-returned"}).encode() monkeypatch.setattr(client, "_request", fake_request) assert client.daemon_name() == HOST_HOSTNAME assert requests == [("GET", "/info", bootstrap.MAX_DOCKER_JSON_BYTES)] @pytest.mark.parametrize( "name", ["", "Tower/other", "Tower\n", "Töwer", "bad..name", "-tower", "tower-", "a" * 254], ) def test_docker_info_hostname_reader_rejects_malformed_or_non_ascii_names( monkeypatch: pytest.MonkeyPatch, name: str ): client = object.__new__(bootstrap.DockerSocketClient) monkeypatch.setattr( client, "_request", lambda *_args, **_kwargs: (200, json.dumps({"Name": name}).encode()), ) with pytest.raises(bootstrap.BootstrapError): client.daemon_name() def test_docker_info_hostname_reader_rejects_oversized_response(monkeypatch: pytest.MonkeyPatch): read_limits: list[int] = [] class Response: status = 200 def read(self, limit: int) -> bytes: read_limits.append(limit) return b"x" * limit class Connection: def __init__(self, _socket_path: str): pass def request(self, *_args: Any, **_kwargs: Any) -> None: pass def getresponse(self) -> Response: return Response() def close(self) -> None: pass monkeypatch.setattr(bootstrap, "_UnixHTTPConnection", Connection) with pytest.raises(bootstrap.BootstrapError, match="response_too_large"): bootstrap.DockerSocketClient().daemon_name() assert read_limits == [bootstrap.MAX_DOCKER_JSON_BYTES + 1] @pytest.mark.parametrize( ("daemon_name", "process_hostname", "caller_hostname"), [ ("Tower", "tower", "Tower"), ("OtherHost", "Tower", "Tower"), ("Tower", "Tower", "OtherHost"), ("bad/name", "bad/name", "bad/name"), ("Töwer", "Töwer", "Töwer"), ], ) def test_legacy_caller_rejects_hostname_source_disagreement_without_creating_child( daemon_name: str, process_hostname: str, caller_hostname: str ): docker = FakeDocker(_legacy_helper_spec()) docker.host_name = daemon_name docker.containers[bootstrap.LEGACY_HELPER_NAME]["Config"]["Hostname"] = caller_hostname git = FakeProtectedGit("") with pytest.raises(bootstrap.BootstrapError): _run_bootstrap(docker, git, hostname=process_hostname) assert bootstrap.LEGACY_HELPER_NAME in docker.inspected assert process_hostname not in docker.inspected assert docker.created == [] def test_legacy_caller_rejects_malformed_container_id_without_creating_child(): docker = FakeDocker(_legacy_helper_spec()) docker.containers[bootstrap.LEGACY_HELPER_NAME]["Id"] = "not-a-container-id" git = FakeProtectedGit("") with pytest.raises(bootstrap.BootstrapError): _run_bootstrap(docker, git) assert docker.created == [] def test_missing_fixed_legacy_name_does_not_fall_back_to_process_hostname(): docker = FakeDocker(_legacy_helper_spec()) docker.containers[HOST_HOSTNAME] = docker.containers.pop(bootstrap.LEGACY_HELPER_NAME) git = FakeProtectedGit("") with pytest.raises(bootstrap.BootstrapError): _run_bootstrap(docker, git, hostname=HOST_HOSTNAME) assert bootstrap.LEGACY_HELPER_NAME in docker.inspected assert HOST_HOSTNAME not in docker.inspected assert docker.created == [] @pytest.mark.parametrize( "mutation", [ "caller-image", "caller-label", "caller-revision", "caller-command", "caller-bind", "appops-image", "appops-provenance", "target-id", "target-source", "target-tree", "target-version", "old-target-id", "old-target-image", "old-target-version", "target-config-image", "target-compose-hash", "target-compose-missing", "target-compose-extra", "target-compose-image", "target-tech-label", "target-tech-extra", "target-path", "target-extra-mount", "target-bind-mode", "base-image-installed-provenance", ], ) def test_bootstrap_rejects_unknown_runtime_identity_without_creating_child(mutation: str): docker = FakeDocker(_legacy_helper_spec()) if mutation == "caller-image": docker.containers[CALLER_ID[:12]]["Image"] = "sha256:" + "9" * 64 elif mutation == "caller-label": docker.containers[CALLER_ID[:12]]["Config"]["Labels"]["itworx.controller.version"] = "2.1.4" elif mutation == "caller-revision": docker.containers[CALLER_ID[:12]]["Config"]["Labels"]["itworx.source.revision"] = "d" * 40 elif mutation == "caller-command": docker.containers[CALLER_ID[:12]]["Config"]["Cmd"].append("--image-id") elif mutation == "caller-bind": docker.containers[CALLER_ID[:12]]["HostConfig"]["Binds"].append("/tmp:/payload") elif mutation == "appops-image": docker.containers[APPOPS_CONTAINER_ID]["Image"] = "sha256:" + "8" * 64 elif mutation == "appops-provenance": docker.images[APPOPS_IMAGE_ID]["Config"]["Labels"]["org.opencontainers.image.revision"] = "f" * 40 elif mutation == "target-id": docker.containers["gitea-deploy-control"]["Id"] = "0" * 64 elif mutation == "old-target-id": docker.containers["gitea-deploy-control"]["Id"] = "5539b672e66890e657d0f3b0e6e94f9c9ca5b47c245e120d330f6f1d5be73fae" elif mutation == "old-target-image": docker.containers["gitea-deploy-control"]["Image"] = "sha256:adbb51b2873bf05546019c341caf6fd4368159546c36a5aea81b4d8bf088804e" elif mutation == "target-source": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["org.opencontainers.image.revision"] = "f" * 40 elif mutation == "target-tree": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["org.opencontainers.image.source-tree"] = "f" * 40 elif mutation == "target-version": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["org.opencontainers.image.version"] = "2.1.11" elif mutation == "old-target-version": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["org.opencontainers.image.version"] = "v26.09.29" elif mutation == "target-config-image": docker.containers["gitea-deploy-control"]["Config"]["Image"] = "docker.gitea.com/runner-images:ubuntu-22.04" elif mutation == "target-compose-hash": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["com.docker.compose.config-hash"] = "0" * 64 elif mutation == "target-compose-missing": docker.containers["gitea-deploy-control"]["Config"]["Labels"].pop("com.docker.compose.service") elif mutation == "target-compose-extra": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["com.docker.compose.container_name"] = "gitea-deploy-control" elif mutation == "target-compose-image": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["com.docker.compose.image"] = CONTROLLER_IMAGE_ID elif mutation == "target-tech-label": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["tech.itworx.role"] = "other" elif mutation == "target-tech-extra": docker.containers["gitea-deploy-control"]["Config"]["Labels"]["tech.itworx.extra"] = "unexpected" elif mutation == "target-path": docker.containers["gitea-deploy-control"]["Mounts"][3]["Destination"] = "/controller-config" elif mutation == "target-extra-mount": docker.containers["gitea-deploy-control"]["Mounts"].append( { "Type": "bind", "Source": "/tmp", "Destination": "/tmp", "RW": True, "Propagation": "rprivate", } ) docker.containers["gitea-deploy-control"]["HostConfig"]["Binds"].append( "/tmp:/tmp:rw" ) elif mutation == "target-bind-mode": docker.containers["gitea-deploy-control"]["HostConfig"]["Binds"][6] = ( "/mnt/cache/appdata/gitea-deploy-runners/controller:/opt/gitea-deploy:rw" ) docker.containers["gitea-deploy-control"]["Mounts"][3]["RW"] = True elif mutation == "base-image-installed-provenance": labels = docker.images[CONTROLLER_IMAGE_ID]["Config"]["Labels"] labels["org.opencontainers.image.revision"] = CONTROLLER_REVISION labels["org.opencontainers.image.source-tree"] = CONTROLLER_TREE git = FakeProtectedGit("") with pytest.raises(bootstrap.BootstrapError): _run_bootstrap(docker, git) assert docker.created == [] @pytest.mark.parametrize( "mutation", ["revision", "tree", "version", "artifact-sha", "delivery-health"], ) def test_bootstrap_requires_exact_mounted_controller_artifact_health(mutation: str): initial = _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) artifact = initial["artifact"] if mutation == "revision": artifact["sourceRevision"] = "9" * 40 elif mutation == "tree": artifact["sourceTree"] = "9" * 40 elif mutation == "version": artifact["controllerVersion"] = "2.1.11" elif mutation == "artifact-sha": artifact["artifactSha256"] = "9" * 64 elif mutation == "delivery-health": initial["delivery"]["health"]["artifact"] = "degraded" docker = FakeDocker(_legacy_helper_spec()) git = FakeProtectedGit("") with pytest.raises(bootstrap.BootstrapError): _run_bootstrap(docker, git, initial_health=initial) assert docker.created == [] def test_running_modern_child_returns_unknown_without_duplicate_or_stop(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) helper_spec = bootstrap.render_modern_helper_spec( helper_source, revision=REVISION, image_id=CONTROLLER_IMAGE_ID, current_controller_id=CONTROLLER_CONTAINER_ID, current_controller_revision=CONTROLLER_REVISION, current_controller_tree=CONTROLLER_TREE, current_controller_version="2.1.12", ) existing = _inspect_from_spec( "itworx-controller-artifact-promoter-modern", MODERN_HELPER_ID, CONTROLLER_IMAGE_ID, helper_spec, state={"Running": True, "Status": "running"}, ) docker = FakeDocker(_legacy_helper_spec(), existing_child=existing) git = FakeProtectedGit(helper_source) result = _run_bootstrap(docker, git, health=_health()) assert result["status"] == "outcome_unknown" assert result["reason"] == "controller_promoter_modern_helper_running" assert docker.created == [] assert docker.started == [] def test_exact_failed_historical_child_is_removed_before_current_helper_launch(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) legacy_health = _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) failed_child = _retired_failed_child() docker = FakeDocker(_legacy_helper_spec(), existing_child=failed_child) git = FakeProtectedGit(helper_source) result = _run_bootstrap( docker, git, health_sequence=[ legacy_health, legacy_health, legacy_health, legacy_health, _health(), ], ) assert result["status"] == "promoted" assert docker.removed == [RETIRED_HELPER_ID] assert docker.created and docker.started == [MODERN_HELPER_ID] assert RETIRED_HELPER_ID in docker.inspected assert docker.appops_inspect_count == 2 assert git.blobs_read.count( (bootstrap.RETIRED_FAILED_MODERN_REVISION, bootstrap.MODERN_HELPER_SOURCE_PATH) ) == 1 def test_current_214_baseline_is_verified_and_bound_to_new_helper(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) docker = FakeDocker(_legacy_helper_spec()) docker.containers["gitea-deploy-control"] = _modern_controller_source_inspect( MODERN_CONTROLLER_REVISION ) git = FakeProtectedGit(helper_source) result = _run_bootstrap( docker, git, health_sequence=[_modern_baseline_health(), _health(), _health()], ) assert result["status"] == "promoted" assert git.controller_release_checks == [bootstrap.MODERN_CONTROLLER_BASELINE] assert docker.created[0]["Labels"] == { "itworx.controller.promoter": "true", "itworx.source.repository": "jens/unraid-appops-gateway", "itworx.source.revision": REVISION, "itworx.controller.version": "2.1.21", "itworx.controller.bootstrap.target.id": MODERN_CONTROLLER_CONTAINER_ID, "itworx.controller.bootstrap.target.image": CONTROLLER_IMAGE_ID, "itworx.controller.bootstrap.target.revision": MODERN_CONTROLLER_REVISION, "itworx.controller.bootstrap.target.tree": MODERN_CONTROLLER_TREE, "itworx.controller.bootstrap.target.version": "2.1.14", } def test_retired_failed_child_is_removed_only_after_current_214_profile_rechecks(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) failed_child = _retired_failed_child() docker = FakeDocker(_legacy_helper_spec(), existing_child=failed_child) docker.containers["gitea-deploy-control"] = _modern_controller_source_inspect( MODERN_CONTROLLER_REVISION ) git = FakeProtectedGit(helper_source) result = _run_bootstrap( docker, git, health_sequence=[ _modern_baseline_health(), _modern_baseline_health(), _modern_baseline_health(), _modern_baseline_health(), _health(), _health(), ], ) assert result["status"] == "promoted" assert docker.removed == [RETIRED_HELPER_ID] assert docker.created[0]["Labels"]["itworx.controller.bootstrap.target.id"] == ( MODERN_CONTROLLER_CONTAINER_ID ) @pytest.mark.parametrize( "mutation", [ "container_id", "image_id", "source_revision", "source_tree", "artifact_digest", "version", "compose_label", "extra_label", "missing_label", "extra_mount", "missing_mount", "writable_mount", "integer_rw_mount", "missing_rw_mount", "malformed_mount_source", "malformed_mount_name", "mount_alias", "dead_controller", "paused_controller", "restarting_controller", "missing_state_flag", "health_revision", "health_tree", "health_artifact", "health_version", ], ) def test_retired_cleanup_refuses_any_current_214_identity_or_topology_drift(mutation: str): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) failed_child = _retired_failed_child() docker = FakeDocker(_legacy_helper_spec(), existing_child=failed_child) controller = _modern_controller_source_inspect(MODERN_CONTROLLER_REVISION) health = _modern_baseline_health() if mutation == "container_id": controller["Id"] = "9" * 64 elif mutation == "image_id": controller["Image"] = "sha256:" + "9" * 64 elif mutation == "source_revision": controller["Mounts"][3]["Source"] = ( "/var/lib/docker/volumes/itworx-controller-hub-releases/_data/" + "9" * 40 ) elif mutation == "source_tree": health["artifact"]["sourceTree"] = "9" * 40 elif mutation == "artifact_digest": health["artifact"]["artifactSha256"] = "9" * 64 elif mutation == "version": controller["Config"]["Labels"]["org.opencontainers.image.version"] = "forged" elif mutation == "compose_label": controller["Config"]["Labels"]["com.docker.compose.service"] = "other" elif mutation == "extra_label": controller["Config"]["Labels"]["unexpected"] = "value" elif mutation == "missing_label": del controller["Config"]["Labels"]["tech.itworx.role"] elif mutation == "extra_mount": controller["Mounts"].append(dict(controller["Mounts"][0])) controller["Mounts"][-1]["Destination"] = "/unexpected" elif mutation == "missing_mount": controller["Mounts"].pop() elif mutation == "writable_mount": controller["Mounts"][1]["RW"] = True elif mutation == "integer_rw_mount": controller["Mounts"][1]["RW"] = 0 elif mutation == "missing_rw_mount": del controller["Mounts"][1]["RW"] elif mutation == "malformed_mount_source": controller["Mounts"][1]["Source"] = 123 elif mutation == "malformed_mount_name": controller["Mounts"][8]["Name"] = None elif mutation == "mount_alias": controller["Mounts"][9]["Source"] = "/tmp/controller-config" elif mutation == "dead_controller": controller["State"]["Dead"] = True elif mutation == "paused_controller": controller["State"]["Paused"] = True elif mutation == "restarting_controller": controller["State"]["Restarting"] = True elif mutation == "missing_state_flag": del controller["State"]["Dead"] elif mutation == "health_revision": health["artifact"]["sourceRevision"] = "9" * 40 elif mutation == "health_tree": health["artifact"]["sourceTree"] = "9" * 40 elif mutation == "health_artifact": health["artifact"]["artifactSha256"] = "9" * 64 elif mutation == "health_version": health["artifact"]["controllerVersion"] = "2.1.13" docker.containers["gitea-deploy-control"] = controller with pytest.raises(bootstrap.BootstrapError): _run_bootstrap( docker, FakeProtectedGit(helper_source), health_sequence=[health, health, health], ) assert docker.removed == [] assert docker.created == [] assert docker.started == [] @pytest.mark.parametrize( "mutation", [ "running", "successful-exit", "boolean-exit", "other-exit", "wrong-id", "config-volumes", "extra-host-mount", "cap-add", "device-request", "docker-root-writable", "socket-bind-changed", "legacy-controller-stopped", "current-main-changed", "second-name-inspect-changed", "second-id-inspect-changed", "appops-stopped-before-delete", "historical-tree-proof-failed", ], ) def test_retired_helper_cleanup_fails_closed_on_any_identity_or_state_drift(mutation: str): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) failed_child = _retired_failed_child() docker = FakeDocker(_legacy_helper_spec(), existing_child=failed_child) git = FakeProtectedGit(helper_source) if mutation == "running": failed_child["State"].update({"Running": True, "Status": "running"}) elif mutation == "successful-exit": failed_child["State"]["ExitCode"] = 0 elif mutation == "boolean-exit": failed_child["State"]["ExitCode"] = True elif mutation == "other-exit": failed_child["State"]["ExitCode"] = 2 elif mutation == "wrong-id": failed_child["Id"] = "9" * 64 elif mutation == "config-volumes": failed_child["Config"]["Volumes"] = {"/extra": {}} elif mutation == "extra-host-mount": failed_child["HostConfig"]["Mounts"] = [{"Type": "volume"}] elif mutation == "cap-add": failed_child["HostConfig"]["CapAdd"] = ["SYS_ADMIN"] elif mutation == "device-request": failed_child["HostConfig"]["DeviceRequests"] = [{"Capabilities": [["gpu"]]}] elif mutation == "docker-root-writable": failed_child["HostConfig"]["Binds"][2] = "/var/lib/docker:/var/lib/docker:rw" failed_child["Mounts"][2]["RW"] = True elif mutation == "socket-bind-changed": failed_child["HostConfig"]["Binds"][4] = "/tmp/docker.sock:/var/run/docker.sock" failed_child["Mounts"][4]["Source"] = "/tmp/docker.sock" elif mutation == "legacy-controller-stopped": docker.containers["gitea-deploy-control"]["State"]["Running"] = False elif mutation == "current-main-changed": git.fail_current_main_at = 2 elif mutation == "second-name-inspect-changed": docker.named_inspect_override = json.loads(json.dumps(failed_child)) docker.named_inspect_override["Config"]["Cmd"].append("unexpected") docker.named_inspect_override_at = 2 elif mutation == "second-id-inspect-changed": docker.retired_id_inspect_override = json.loads(json.dumps(failed_child)) docker.retired_id_inspect_override["State"]["ExitCode"] = 0 elif mutation == "appops-stopped-before-delete": docker.appops_inspect_override = json.loads( json.dumps(docker.containers[APPOPS_CONTAINER_ID]) ) docker.appops_inspect_override["State"]["Running"] = False docker.appops_inspect_override_at = 2 elif mutation == "historical-tree-proof-failed": git.retired_tree_valid = False legacy_health = _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) with pytest.raises(bootstrap.BootstrapError): _run_bootstrap( docker, git, health_sequence=[legacy_health, legacy_health, legacy_health], ) assert docker.removed == [] assert docker.created == [] assert docker.started == [] def test_retired_helper_remove_uses_non_force_non_volume_docker_delete(): client = bootstrap.DockerSocketClient() calls: list[tuple[str, str]] = [] def request(method: str, path: str, **_kwargs: Any) -> tuple[int, bytes]: calls.append((method, path)) return 204, b"" client._request = request # type: ignore[method-assign] client.remove_retired_failed_container(RETIRED_HELPER_ID) assert calls == [ ("DELETE", f"/containers/{RETIRED_HELPER_ID}?force=false&v=false") ] @pytest.mark.parametrize("status", [404, 409]) def test_retired_helper_delete_refuses_unexpected_docker_status(status: int): client = bootstrap.DockerSocketClient() def request(_method: str, _path: str, **_kwargs: Any) -> tuple[int, bytes]: return status, b"" client._request = request # type: ignore[method-assign] with pytest.raises(bootstrap.BootstrapError): client.remove_retired_failed_container(RETIRED_HELPER_ID) def test_same_failed_source_revision_is_never_retried_or_removed(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) docker = FakeDocker( _legacy_helper_spec(bootstrap.RETIRED_FAILED_MODERN_REVISION), existing_child=_retired_failed_child(), ) git = FakeProtectedGit(helper_source) git.revision = bootstrap.RETIRED_FAILED_MODERN_REVISION git.tree = bootstrap.RETIRED_FAILED_MODERN_TREE legacy_health = _health( revision=CONTROLLER_REVISION, tree=CONTROLLER_TREE, version="2.1.12", artifact_sha256=bootstrap.LEGACY_CONTROLLER_ARTIFACT_SHA256, ) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_retired_helper_retry_refused"): _run_bootstrap( docker, git, health_sequence=[legacy_health], source_revision=git.revision, source_tree=git.tree, ) assert docker.removed == [] assert docker.created == [] def test_completed_modern_child_with_wrong_artifact_is_not_relaunched(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) helper_spec = bootstrap.render_modern_helper_spec( helper_source, revision=REVISION, image_id=CONTROLLER_IMAGE_ID, current_controller_id=CONTROLLER_CONTAINER_ID, current_controller_revision=CONTROLLER_REVISION, current_controller_tree=CONTROLLER_TREE, current_controller_version="2.1.12", ) existing = _inspect_from_spec( "itworx-controller-artifact-promoter-modern", MODERN_HELPER_ID, CONTROLLER_IMAGE_ID, helper_spec, state={"Running": False, "ExitCode": 0, "Status": "exited"}, ) docker = FakeDocker(_legacy_helper_spec(), existing_child=existing) git = FakeProtectedGit(helper_source) result = _run_bootstrap( docker, git, initial_health=_health(tree="9" * 40), ) assert result["status"] == "outcome_unknown" assert result["reason"] == "controller_promoter_postcheck_failed" assert docker.created == [] def test_completed_child_uses_new_health_after_controller_recreation(): helper_source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_text( encoding="utf-8" ) helper_spec = bootstrap.render_modern_helper_spec( helper_source, revision=REVISION, image_id=CONTROLLER_IMAGE_ID, current_controller_id=CONTROLLER_CONTAINER_ID, current_controller_revision=CONTROLLER_REVISION, current_controller_tree=CONTROLLER_TREE, current_controller_version="2.1.12", ) existing = _inspect_from_spec( "itworx-controller-artifact-promoter-modern", MODERN_HELPER_ID, CONTROLLER_IMAGE_ID, helper_spec, state={"Running": False, "ExitCode": 0, "Status": "exited"}, ) docker = FakeDocker(_legacy_helper_spec(), existing_child=existing) docker.containers["gitea-deploy-control"] = _modern_controller_inspect(REVISION) git = FakeProtectedGit(helper_source) result = _run_bootstrap( docker, git, initial_health=_health(), ) assert result["status"] == "already_applied" assert result["revision"] == REVISION assert docker.created == [] assert docker.started == [] @pytest.mark.parametrize( "mutation", [ "post-image-config", "post-compose-hash", "post-compose-image", "post-compose-extra", "post-compose-environment-file", "post-tech-label", "post-tech-extra", "post-artifact-revision", "post-volume-name", "post-volume-source", "post-volume-mode", "post-config-bind-source", "post-config-bind-mode", "post-code-bind-propagation", "post-config-bind-propagation", "post-missing-backup-volume", "post-backup-volume-type", "post-backup-volume-name", "post-backup-volume-source", "post-backup-volume-target", "post-backup-volume-mode", "post-backup-volume-propagation", "post-duplicate-backup-volume", "post-extra-mount", ], ) def test_postactivation_controller_requires_exact_compose_runtime(mutation: str): docker = FakeDocker(_legacy_helper_spec()) controller = _modern_controller_inspect(REVISION) docker.containers["gitea-deploy-control"] = controller if mutation == "post-image-config": controller["Config"]["Image"] = CONTROLLER_IMAGE_REF elif mutation == "post-compose-hash": controller["Config"]["Labels"]["com.docker.compose.config-hash"] = "bad" elif mutation == "post-compose-image": controller["Config"]["Labels"]["com.docker.compose.image"] = "sha256:" + "d" * 64 elif mutation == "post-compose-extra": controller["Config"]["Labels"]["com.docker.compose.extra"] = "unexpected" elif mutation == "post-compose-environment-file": controller["Config"]["Labels"]["com.docker.compose.project.environment_file"] = "/tmp/.env" elif mutation == "post-tech-label": controller["Config"]["Labels"]["tech.itworx.gitea.scope"] = "Other" elif mutation == "post-tech-extra": controller["Config"]["Labels"]["tech.itworx.extra"] = "unexpected" elif mutation == "post-artifact-revision": controller["Mounts"][3]["Source"] = ( "/var/lib/docker/volumes/itworx-controller-hub-releases/_data/" f"controller-artifacts/{'d' * 40}" ) elif mutation == "post-volume-name": controller["Mounts"][8]["Name"] = "foreign-volume" elif mutation == "post-volume-source": controller["Mounts"][8]["Source"] = "/tmp/foreign-volume" elif mutation == "post-volume-mode": controller["Mounts"][8]["RW"] = False elif mutation == "post-config-bind-source": controller["Mounts"][9]["Source"] = "/tmp/config" elif mutation == "post-config-bind-mode": controller["Mounts"][9]["RW"] = True elif mutation == "post-code-bind-propagation": controller["Mounts"][3]["Propagation"] = "rslave" elif mutation == "post-config-bind-propagation": controller["Mounts"][9]["Propagation"] = "shared" elif mutation == "post-missing-backup-volume": controller["Mounts"].pop(10) elif mutation == "post-backup-volume-type": controller["Mounts"][10]["Type"] = "bind" elif mutation == "post-backup-volume-name": controller["Mounts"][10]["Name"] = "foreign-volume" elif mutation == "post-backup-volume-source": controller["Mounts"][10]["Source"] = "/tmp/foreign-volume" elif mutation == "post-backup-volume-target": controller["Mounts"][10]["Destination"] = "/tmp/backup" elif mutation == "post-backup-volume-mode": controller["Mounts"][10]["RW"] = False elif mutation == "post-backup-volume-propagation": controller["Mounts"][10]["Propagation"] = "rprivate" elif mutation == "post-duplicate-backup-volume": controller["Mounts"].append(dict(controller["Mounts"][10])) elif mutation == "post-extra-mount": controller["Mounts"].append(dict(controller["Mounts"][0])) assert not bootstrap._verified_live_controller(docker, source_revision=REVISION) def test_postactivation_accepts_well_formed_compose_hash(): docker = FakeDocker(_legacy_helper_spec()) controller = _modern_controller_inspect(REVISION) controller["Config"]["Labels"]["com.docker.compose.config-hash"] = "b" * 64 docker.containers["gitea-deploy-control"] = controller assert bootstrap._verified_live_controller(docker, source_revision=REVISION) def test_postactivation_accepts_exact_forge_backup_mount_in_eleven_mount_inventory(): docker = FakeDocker(_legacy_helper_spec()) controller = _modern_controller_inspect(REVISION) docker.containers["gitea-deploy-control"] = controller assert bootstrap._verified_live_controller(docker, source_revision=REVISION) def test_successful_child_requires_exact_source_tree_version_and_health(): docker = FakeDocker(_legacy_helper_spec()) git = FakeProtectedGit("") result = _run_bootstrap(docker, git) assert result["status"] == "promoted" assert result["revision"] == REVISION assert result["controllerVersion"] == "2.1.21" assert result["artifactSha256"] == "1" * 64 assert docker.started == [MODERN_HELPER_ID] assert docker.created[0]["Image"] == CONTROLLER_IMAGE_ID assert docker.created[0]["HostConfig"]["ReadonlyRootfs"] is True def test_child_logs_are_bounded_and_redacted_before_result_output(): docker = FakeDocker(_legacy_helper_spec()) docker.log_bytes = ( b"controller_v2_ready revision=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\n" b"GITEA_ADMIN_TOKEN=very-secret-value\n" ) git = FakeProtectedGit("") result = _run_bootstrap(docker, git) assert result["status"] == "promoted" assert "very-secret-value" not in result["logTail"] assert "[redacted]" in result["logTail"] assert len(result["logTail"].encode("utf-8")) <= 4096 @pytest.mark.parametrize( "temporary_root_info,accepted", [ ("1777:0:0", True), ("700:0:0", True), ("777:0:0", False), ("755:0:0", False), ("770:0:0", False), ("1700:0:0", False), ("1777:1000:0", False), ("700:0:1000", False), ], ) def test_legacy_upgrade_signature_fetches_bootstrap_only_from_protected_git( tmp_path: Path, temporary_root_info: str, accepted: bool ): bash = shutil.which("bash") if bash is None: pytest.skip("bash is required to exercise the upgrade script") bin_dir = tmp_path / "bin" bin_dir.mkdir() git_trace = tmp_path / "git-args.jsonl" python_trace = tmp_path / "python-args.json" protected_source = "# trusted protected Git bootstrap source\n" git_shim = bin_dir / "git" git_shim.write_text( f"#!{sys.executable}\n" "import json, os, sys\n" f"trace = {str(git_trace)!r}\n" f"with open(trace, 'a', encoding='utf-8') as stream: stream.write(json.dumps(sys.argv[1:]) + '\\n')\n" "args = [arg for arg in sys.argv[1:] if not arg.startswith('--git-dir=')]\n" f"revision = {REVISION!r}\n" f"tree = {TREE!r}\n" "if args[-3:] == ['rev-parse', 'refs/heads/main', '--'] or args[-2:] == ['rev-parse', 'refs/heads/main']: print(revision)\n" "elif any(arg.endswith('^{tree}') for arg in args): print(tree)\n" f"elif len(args) >= 3 and args[:2] == ['cat-file', '-s']: print({len(protected_source)})\n" "elif len(args) >= 3 and args[0] == 'cat-file' and args[1] == 'blob' and args[-1].endswith(':scripts/bootstrap_controller_artifact_promoter.py'):\n" f" sys.stdout.write({protected_source!r})\n" "else: raise SystemExit(91)\n", encoding="utf-8", ) stat_shim = bin_dir / "stat" stat_shim.write_text( f"#!{sys.executable}\n" "import pathlib, sys\n" "args = sys.argv[1:]\n" "target = args[-1]\n" f"if target == '/tmp': print({temporary_root_info!r})\n" "elif args[1] == '%u:%a': print('0:700')\n" "elif args[1] == '%u:%a:%s': print('0:600:' + str(pathlib.Path(target).stat().st_size))\n" "else: raise SystemExit(1)\n", encoding="utf-8", ) python_shim = bin_dir / "python3" python_shim.write_text( f"#!{sys.executable}\n" "import json, pathlib, sys\n" f"path = pathlib.Path({str(python_trace)!r})\n" "path.write_text(json.dumps({'script': pathlib.Path(sys.argv[1]).read_text(), 'args': sys.argv[2:]}), encoding='utf-8')\n", encoding="utf-8", ) git_shim.chmod(0o755) stat_shim.chmod(0o755) python_shim.chmod(0o755) archive = f"/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-{REVISION[:12]}/controller.tar" result = subprocess.run( [ bash, str(ROOT / "scripts/upgrade_deployment_controller.sh"), "--archive", archive, "--archive-sha256", "a" * 64, "--source-repository", "jens/unraid-appops-gateway", "--source-revision", REVISION, "--source-tree", TREE, "--controller-version", "2.1.5", "--confirm", "UPGRADE_DEPLOYMENT_CONTROLLER_V2", ], cwd=ROOT, env={**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"}, capture_output=True, text=True, check=False, ) if not accepted: assert result.returncode == 66, result.stderr assert "temporary root is unsafe" in result.stderr assert not python_trace.exists() calls = [json.loads(line) for line in git_trace.read_text(encoding="utf-8").splitlines()] assert not any("cat-file" in args for args in calls) return assert result.returncode == 0, result.stderr bootstrap_call = json.loads(python_trace.read_text(encoding="utf-8")) assert bootstrap_call["script"] == protected_source assert bootstrap_call["args"] == [ "--source-repository", "jens/unraid-appops-gateway", "--source-revision", REVISION, "--source-tree", TREE, ] calls = [json.loads(line) for line in git_trace.read_text(encoding="utf-8").splitlines()] assert any(args[-1:] == [f"{REVISION}:scripts/bootstrap_controller_artifact_promoter.py"] for args in calls) assert all(archive not in args for args in calls) @pytest.mark.parametrize( "mutation", ["source", "revision", "tree", "version", "extra", "image-id"], ) def test_upgrade_script_rejects_mutated_legacy_call_without_bootstrap( tmp_path: Path, mutation: str ): bash = shutil.which("bash") if bash is None: pytest.skip("bash is required to exercise the upgrade script") bin_dir = tmp_path / "bin" bin_dir.mkdir() marker = tmp_path / "bootstrap-ran" python_shim = bin_dir / "python3" python_shim.write_text( f"#!/usr/bin/env python3\nfrom pathlib import Path\nPath({str(marker)!r}).write_text('ran')\n", encoding="utf-8", ) python_shim.chmod(0o755) args = [ "--archive", f"/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-{REVISION[:12]}/controller.tar", "--archive-sha256", "a" * 64, "--source-repository", "jens/unraid-appops-gateway", "--source-revision", REVISION, "--source-tree", TREE, "--controller-version", "2.1.5", "--confirm", "UPGRADE_DEPLOYMENT_CONTROLLER_V2", ] if mutation == "source": args[args.index("jens/unraid-appops-gateway")] = "attacker/repo" elif mutation == "revision": args[args.index(REVISION)] = "A" * 40 elif mutation == "tree": args[args.index(TREE)] = "f" * 40 elif mutation == "version": args[args.index("2.1.5")] = "2.1.4" elif mutation == "extra": args.extend(["--unexpected", "value"]) elif mutation == "image-id": args.extend(["--image-id", CONTROLLER_IMAGE_ID]) result = subprocess.run( [bash, str(ROOT / "scripts/upgrade_deployment_controller.sh"), *args], cwd=ROOT, env={**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"}, capture_output=True, text=True, check=False, ) assert result.returncode != 0 assert not marker.exists() @pytest.mark.parametrize("modern_call", ["missing-image", "shared-archive"]) @pytest.mark.parametrize("temporary_root_info", ["1777:0:0", "700:0:0", "777:0:0"]) def test_modern_upgrade_path_still_requires_private_archive_and_image_id( tmp_path: Path, modern_call: str, temporary_root_info: str ): bash = shutil.which("bash") if bash is None: pytest.skip("bash is required to exercise the upgrade script") bin_dir = tmp_path / "bin" bin_dir.mkdir() stat_shim = bin_dir / "stat" stat_shim.write_text( f"#!{sys.executable}\nimport sys\n" f"if sys.argv[-1] == '/tmp': print({temporary_root_info!r})\n" "else: raise SystemExit(91)\n", encoding="utf-8", ) stat_shim.chmod(0o755) archive = ( f"/tmp/itworx-controller-artifact-promoter-{REVISION[:12]}.AAAAAA/controller.tar" if modern_call == "missing-image" else f"/mnt/cache/appdata/gitea-deploy-runners/.controller-promoter-{REVISION[:12]}/controller.tar" ) args = [ "--archive", archive, "--archive-sha256", "a" * 64, "--source-repository", "jens/unraid-appops-gateway", "--source-revision", REVISION, "--source-tree", TREE, "--controller-version", "2.1.14", "--confirm", "UPGRADE_DEPLOYMENT_CONTROLLER_V2", ] if modern_call == "shared-archive": args.extend(["--image-id", CONTROLLER_IMAGE_ID]) result = subprocess.run( [bash, str(ROOT / "scripts/upgrade_deployment_controller.sh"), *args], cwd=ROOT, env={**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"}, capture_output=True, text=True, check=False, ) assert result.returncode != 0 if modern_call == "shared-archive": if temporary_root_info == "777:0:0": assert "helper temporary root is unsafe" in result.stderr else: assert "archive is outside the helper private temporary directory" in result.stderr if modern_call == "missing-image": assert "image ID" in result.stderr def test_current_helper_version_is_distinct_from_fixed_controller_baselines(): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() namespace = bootstrap._safe_modern_spec_namespace(source) assert namespace["CONTROLLER_VERSION"] == "2.1.21" assert bootstrap._verified_historical_helper_version(source) == "2.1.21" assert [baseline.version for baseline in bootstrap.CONTROLLER_BASELINES] == ["2.1.12", "2.1.14"] with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(RETIRED_MODERN_HELPER_SOURCE.encode("utf-8")) assert bootstrap._verified_historical_helper_version(RETIRED_MODERN_HELPER_SOURCE.encode("utf-8")) == "2.1.14" @pytest.mark.parametrize("version", ["2.1.16", "2.1.22", "9.9.9"]) def test_current_and_historical_helper_reject_unreviewed_target_versions(version): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() foreign = source.replace(b'CONTROLLER_VERSION = "2.1.21"', f'CONTROLLER_VERSION = "{version}"'.encode()) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(foreign) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._verified_historical_helper_version(foreign) @pytest.mark.parametrize("version", ["2.1.14", "2.1.16", "2.1.22"]) def test_promoted_health_cannot_substitute_a_baseline_or_unknown_version(version): assert not bootstrap._health_matches(_health(version=version), REVISION, TREE) @pytest.mark.parametrize("version", ["2.1.19", "2.1.20"]) def test_previous_reviewed_helper_is_historical_only(version): source = (ROOT / "backend/app/services/controller_artifact_helper_spec.py").read_bytes() prior = source.replace(b'CONTROLLER_VERSION = "2.1.21"', f'CONTROLLER_VERSION = "{version}"'.encode()) with pytest.raises(bootstrap.BootstrapError, match="controller_promoter_modern_spec_identity_invalid"): bootstrap._safe_modern_spec_namespace(prior) assert bootstrap._verified_historical_helper_version(prior) == version