didericis/bot-bottle
1
0
Fork 0
Code Issues 32 Pull Requests 8 Actions Packages Releases Wiki Activity

refactor(firecracker): split the combined rootfs into per-plane artifacts

Browse Source 
Each infra VM now boots its own rootfs instead of a shared combined image, so
the exposed gateway VM no longer carries buildah + the control-plane code it
never runs (a fatter, less-isolated exposed surface — the opposite of the plane
split's intent). The combined image bought only "one artifact"; each VM already
kept a full copy of the shared rootfs, so nothing was saved at boot.

  * orchestrator rootfs — control plane + buildah (Dockerfile.orchestrator.fc,
    FROM orchestrator); the slim gateway rootfs boots bot-bottle-gateway:latest
    directly. Dockerfile.infra.fc (the combined image) is deleted.
  * `_infra_init` splits into `_orchestrator_init` / `_gateway_init` (shared
    preamble via `_init_head`); each per-plane rootfs bakes only its role init,
    so the `bb_role` cmdline branch is gone.
  * infra_artifact is role-parametrized: a per-role package
    (bot-bottle-firecracker-<role>), version hash, URL, cache dir, and candidate
    subdir. `ensure_built` pulls both; `_expected_version` combines both markers.
  * publish_infra builds the images once, then builds + publishes an
    orchestrator and a gateway artifact under DIR/<role>/.

coverage.sh's candidate-dir plumbing is layout-agnostic (publish_infra --output
now fills DIR/<role>/, which ensure_artifact_gz reads). Full suite green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
didericis
2026-07-25 15:21:03 -04:00
parent b7599ed146
commit d0d1da612e
8 changed files with 497 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_firecracker_infra_vm.py
+46 -35
View File
@@ -68,48 +68,55 @@ class TestInfraEndpoint(unittest.TestCase):
ca.assert_called_once_with(timeout=1)
class TestBuildInfraRootfs(unittest.TestCase):
def test_uses_infra_variant_and_role_branched_init(self):
with patch.object(infra_vm.util, "build_base_rootfs_dir") as build:
build.return_value = Path("/cache/rootfs/x-infra")
infra_vm.build_infra_rootfs_dir()
build.assert_called_once()
self.assertEqual(infra_vm._INFRA_IMAGE, build.call_args.args[0])
# variant is "-infra-<init-hash>" so an init change rebuilds the rootfs.
self.assertTrue(build.call_args.kwargs["variant"].startswith("-infra-"))
init = build.call_args.kwargs["init_script"]
# One shared init, role-branched off the kernel cmdline: it starts the
# control plane OR the gateway data plane, and exports PATH so the
# gateway daemons' subprocesses find python3.
self.assertIn("bb_role=", init)
self.assertIn('if [ "$ROLE" = gateway ]; then', init)
self.assertIn("bot_bottle.orchestrator", init)
self.assertIn("bot_bottle.gateway.bootstrap", init)
self.assertIn("export PATH=", init)
class TestRoleInits(unittest.TestCase):
def test_orchestrator_init_starts_only_the_control_plane(self):
init = infra_vm.role_init("orchestrator")
# No bb_role branch — the rootfs *is* the role.
self.assertNotIn("bb_role=", init)
self.assertNotIn("bot_bottle.gateway.bootstrap", init)
self.assertIn("export PATH=", init) # shared preamble
# Persistent registry volume mounted at the DB dir on the orchestrator.
self.assertIn("/dev/vdb", init)
# VM backend uses git-http (9420); the git:// daemon is left out.
self.assertIn("BOT_BOTTLE_GATEWAY_DAEMONS=egress,git-http,supervise", init)
# Role-scoped auth (#469): the orchestrator gets the host-seeded signing
# key; the gateway gets the host-minted `gateway` JWT (NOT the key — the
# JWT is minted on the host now, so the init never touches the key to
# mint it); each plane refuses to start without its secret.
self.assertIn(f"cat {infra_vm._GUEST_SIGNING_KEY_PATH}", init) # host-seeded key
self.assertIn(f"cat {infra_vm._GUEST_GATEWAY_JWT_PATH}", init) # host-minted JWT
self.assertNotIn("ROLE_GATEWAY", init) # minting moved to the host
self.assertIn("refusing to start the control plane", init) # no open mode
self.assertIn("refusing to start the data plane", init) # gateway fail-closed
self.assertIn('BOT_BOTTLE_ORCHESTRATOR_TOKEN="$CP_KEY" python3 -m bot_bottle.orchestrator',
init) # key -> orchestrator only
self.assertIn('BOT_BOTTLE_ORCHESTRATOR_AUTH_JWT="$GW_JWT"', init) # JWT -> gateway daemons
def test_gateway_init_starts_only_the_data_plane(self):
init = infra_vm.role_init("gateway")
self.assertNotIn("bb_role=", init)
self.assertNotIn("bot_bottle.orchestrator", init)
self.assertNotIn("/dev/vdb", init) # no registry volume on the data plane
self.assertIn("export PATH=", init) # shared preamble
self.assertIn("BOT_BOTTLE_GATEWAY_DAEMONS=egress,git-http,supervise", init)
self.assertIn(f"cat {infra_vm._GUEST_GATEWAY_JWT_PATH}", init) # host-minted JWT
self.assertNotIn("ROLE_GATEWAY", init) # minting moved to the host
self.assertIn("refusing to start the data plane", init) # fail-closed
self.assertIn('BOT_BOTTLE_ORCHESTRATOR_AUTH_JWT="$GW_JWT"', init) # JWT -> daemons
# The gateway resolves the orchestrator's address off the cmdline
# (bb_orch), so no IP is baked into the artifact.
self.assertIn("bb_orch=", init)
self.assertIn("BOT_BOTTLE_ORCHESTRATOR_URL=http://$ORCH:", init)
class TestBuildRootfsDir(unittest.TestCase):
def test_orchestrator_rootfs_uses_the_fc_image_and_role_variant(self):
with patch.object(infra_vm.util, "build_base_rootfs_dir") as build:
build.return_value = Path("/cache/rootfs/x")
infra_vm.build_rootfs_dir("orchestrator")
self.assertEqual(infra_vm._ORCHESTRATOR_FC_IMAGE, build.call_args.args[0])
self.assertTrue(build.call_args.kwargs["variant"].startswith("-orchestrator-"))
def test_gateway_rootfs_uses_the_gateway_image_directly(self):
with patch.object(infra_vm.util, "build_base_rootfs_dir") as build:
build.return_value = Path("/cache/rootfs/x")
infra_vm.build_rootfs_dir("gateway")
self.assertEqual(infra_vm._GATEWAY_IMAGE, build.call_args.args[0])
self.assertTrue(build.call_args.kwargs["variant"].startswith("-gateway-"))
class TestEnsureBuilt(unittest.TestCase):
def test_default_pulls_artifact_without_docker(self):
def test_default_pulls_both_artifacts_without_docker(self):
# PRD 0069 Stage 2: the launch host pulls the prebuilt rootfs; no Docker.
# Pin BOT_BOTTLE_INFRA_BUILD off: the coverage CI job exports it =local
# for the integration suite, and that ambient value would otherwise send
@@ -119,17 +126,21 @@ class TestEnsureBuilt(unittest.TestCase):
patch.object(infra_vm.infra_artifact, "ensure_artifact_gz") as pull:
infra_vm.ensure_built()
build.assert_not_called()
pull.assert_called_once()
# One pull per plane.
roles = {c.kwargs["role"] for c in pull.call_args_list}
self.assertEqual({"orchestrator", "gateway"}, roles)
def test_local_mode_builds_deps_before_infra(self):
def test_local_mode_builds_orchestrator_before_its_fc_image(self):
with patch.dict(os.environ, {"BOT_BOTTLE_INFRA_BUILD": "local"}), \
patch.object(infra_vm.docker_mod, "build_image") as build:
infra_vm.ensure_built()
tags = [c.args[0] for c in build.call_args_list]
# infra is FROM gateway and COPY --from orchestrator, so both first.
self.assertEqual(infra_vm._INFRA_IMAGE, tags[-1])
self.assertIn(infra_vm._ORCHESTRATOR_IMAGE, tags[:-1])
self.assertIn(infra_vm._GATEWAY_IMAGE, tags[:-1])
# orchestrator-fc is FROM orchestrator, so the base is built first.
self.assertIn(infra_vm._ORCHESTRATOR_IMAGE, tags)
self.assertIn(infra_vm._GATEWAY_IMAGE, tags)
self.assertEqual(infra_vm._ORCHESTRATOR_FC_IMAGE, tags[-1])
self.assertLess(tags.index(infra_vm._ORCHESTRATOR_IMAGE),
tags.index(infra_vm._ORCHESTRATOR_FC_IMAGE))
# The orchestrator + gateway services are imported lazily inside ensure_running
tests/unit/test_infra_artifact.py
+59 -40
View File
@@ -3,6 +3,8 @@
The launch-host half — version hashing and download/verify/decompress — is
what keeps a docker-free host from booting a stale or corrupted rootfs, so the
checksum + fail-closed paths are locked here. Network is mocked; no Docker.
There are two per-plane artifacts now; these exercise one role (orchestrator) —
the pull/verify logic is role-agnostic, keyed only by the package name.
"""
from __future__ import annotations
@@ -21,6 +23,8 @@ from unittest import mock
from bot_bottle.backend.firecracker import infra_artifact as ia
from bot_bottle.log import Die
_ROLE = "orchestrator"
def _gz(data: bytes) -> bytes:
return gzip.compress(data)
@@ -66,29 +70,36 @@ class _CacheMixin(unittest.TestCase):
if sha_text is None:
sha_text = f"{hashlib.sha256(gz_bytes).hexdigest()} rootfs.ext4.gz\n"
net = _FakeNet({
ia.artifact_url(version, "rootfs.ext4.gz"): gz_bytes,
ia.artifact_url(version, "rootfs.ext4.gz.sha256"): sha_text.encode(),
ia.artifact_url(version, "rootfs.ext4.gz", role=_ROLE): gz_bytes,
ia.artifact_url(version, "rootfs.ext4.gz.sha256", role=_ROLE): sha_text.encode(),
})
return mock.patch.object(ia.urllib.request, "urlopen", net.urlopen), net
class TestVersion(unittest.TestCase):
def test_deterministic_16_hex(self) -> None:
v = ia.infra_artifact_version("#!/bin/sh\ntrue\n")
self.assertEqual(v, ia.infra_artifact_version("#!/bin/sh\ntrue\n"))
v = ia.infra_artifact_version("#!/bin/sh\ntrue\n", _ROLE)
self.assertEqual(v, ia.infra_artifact_version("#!/bin/sh\ntrue\n", _ROLE))
self.assertEqual(16, len(v))
int(v, 16) # hex
def test_init_change_bumps_version(self) -> None:
self.assertNotEqual(
ia.infra_artifact_version("a"), ia.infra_artifact_version("b"))
ia.infra_artifact_version("a", _ROLE), ia.infra_artifact_version("b", _ROLE))
def test_role_changes_version(self) -> None:
# The same init under a different role hashes differently (role is folded
# in, and each role hashes its own Dockerfiles).
self.assertNotEqual(
ia.infra_artifact_version("init", "orchestrator"),
ia.infra_artifact_version("init", "gateway"))
class TestVersionInputs(unittest.TestCase):
"""The hash must cover *every* file baked into the rootfs, not just `*.py`
(`COPY bot_bottle` is wholesale) — else a non-Python change (e.g. the egress
entrypoint shell script) leaves the version unchanged and a launch host
boots a rootfs whose code differs from its checkout."""
(the package is baked in wholesale) — else a non-Python change (e.g. the
egress entrypoint shell script) leaves the version unchanged and a launch
host boots a rootfs whose code differs from its checkout."""
def _fake_repo(self, root: Path) -> None:
pkg = root / "bot_bottle"
@@ -96,7 +107,8 @@ class TestVersionInputs(unittest.TestCase):
(pkg / "app.py").write_text("print('hi')\n")
(pkg / "egress_entrypoint.sh").write_text("#!/bin/sh\nexec mitmdump\n")
(pkg / "netpool.defaults.env").write_text("FOO=1\n")
for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra", "Dockerfile.infra.fc"):
for name in ("Dockerfile.orchestrator", "Dockerfile.orchestrator.fc",
"Dockerfile.gateway"):
(root / name).write_text(f"FROM scratch # {name}\n")
(root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
@@ -104,10 +116,10 @@ class TestVersionInputs(unittest.TestCase):
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._fake_repo(root)
before = ia.infra_artifact_version("init", repo_root=root)
before = ia.infra_artifact_version("init", _ROLE, repo_root=root)
(root / "pyproject.toml").write_text(
"[project]\nname = 'bot-bottle'\ndependencies = ['httpx']\n")
after = ia.infra_artifact_version("init", repo_root=root)
after = ia.infra_artifact_version("init", _ROLE, repo_root=root)
self.assertNotEqual(before, after)
def test_dropbear_change_bumps_version(self) -> None:
@@ -119,85 +131,92 @@ class TestVersionInputs(unittest.TestCase):
with mock.patch.dict(os.environ, {
"BOT_BOTTLE_FC_DROPBEAR": str(dropbear),
}):
before = ia.infra_artifact_version("init", repo_root=root)
before = ia.infra_artifact_version("init", _ROLE, repo_root=root)
dropbear.write_bytes(b"dropbear-v2")
after = ia.infra_artifact_version("init", repo_root=root)
after = ia.infra_artifact_version("init", _ROLE, repo_root=root)
self.assertNotEqual(before, after)
def test_non_python_file_change_bumps_version(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._fake_repo(root)
before = ia.infra_artifact_version("init", repo_root=root)
before = ia.infra_artifact_version("init", _ROLE, repo_root=root)
(root / "bot_bottle" / "egress_entrypoint.sh").write_text(
"#!/bin/sh\nexec mitmdump --different\n")
after = ia.infra_artifact_version("init", repo_root=root)
after = ia.infra_artifact_version("init", _ROLE, repo_root=root)
self.assertNotEqual(before, after)
def test_pyc_and_pycache_ignored(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._fake_repo(root)
before = ia.infra_artifact_version("init", repo_root=root)
before = ia.infra_artifact_version("init", _ROLE, repo_root=root)
cache = root / "bot_bottle" / "__pycache__"
cache.mkdir()
(cache / "app.cpython-312.pyc").write_bytes(b"\x00bytecode")
(root / "bot_bottle" / "app.pyc").write_bytes(b"\x00bytecode")
after = ia.infra_artifact_version("init", repo_root=root)
after = ia.infra_artifact_version("init", _ROLE, repo_root=root)
self.assertEqual(before, after)
class TestEnsureArtifact(_CacheMixin):
def _candidate(self, root: Path, version: str, gz: bytes, sha_text: str | None = None) -> Path:
"""Stage a candidate bundle for `_ROLE` under root/<role>/."""
role_dir = root / _ROLE
role_dir.mkdir(parents=True, exist_ok=True)
(role_dir / "version.txt").write_text(version + "\n")
(role_dir / "rootfs.ext4.gz").write_bytes(gz)
if sha_text is None:
sha_text = f"{hashlib.sha256(gz).hexdigest()} rootfs.ext4.gz\n"
(role_dir / "rootfs.ext4.gz.sha256").write_text(sha_text)
return role_dir
def test_uses_verified_ci_candidate_without_network(self) -> None:
version = "deadbeef00000000"
gz = _gz(b"candidate ext4")
with tempfile.TemporaryDirectory() as d:
root = Path(d)
(root / "version.txt").write_text(version + "\n")
(root / "rootfs.ext4.gz").write_bytes(gz)
digest = hashlib.sha256(gz).hexdigest()
(root / "rootfs.ext4.gz.sha256").write_text(
f"{digest} rootfs.ext4.gz\n")
role_dir = self._candidate(root, version, gz)
with mock.patch.dict(os.environ, {
"BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
}), mock.patch.object(ia.urllib.request, "urlopen") as net:
path = ia.ensure_artifact_gz(version)
self.assertEqual(root / "rootfs.ext4.gz", path)
path = ia.ensure_artifact_gz(version, role=_ROLE)
self.assertEqual(role_dir / "rootfs.ext4.gz", path)
net.assert_not_called()
def test_rejects_candidate_for_another_version(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
(root / "version.txt").write_text("wrong\n")
(root / _ROLE).mkdir()
(root / _ROLE / "version.txt").write_text("wrong\n")
with mock.patch.dict(os.environ, {
"BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
}):
with self.assertRaises(Die) as ctx:
ia.ensure_artifact_gz("expected")
ia.ensure_artifact_gz("expected", role=_ROLE)
self.assertIn("version mismatch", str(ctx.exception.message))
def test_rejects_incomplete_candidate(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
(root / "version.txt").write_text("v1\n")
(root / _ROLE).mkdir()
(root / _ROLE / "version.txt").write_text("v1\n")
with mock.patch.dict(os.environ, {
"BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
}):
with self.assertRaises(Die) as ctx:
ia.ensure_artifact_gz("v1")
ia.ensure_artifact_gz("v1", role=_ROLE)
self.assertIn("incomplete", str(ctx.exception.message))
def test_rejects_candidate_checksum_mismatch(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
(root / "version.txt").write_text("v1\n")
(root / "rootfs.ext4.gz").write_bytes(b"bad")
(root / "rootfs.ext4.gz.sha256").write_text("0" * 64 + " rootfs.ext4.gz\n")
self._candidate(root, "v1", b"bad", sha_text="0" * 64 + " rootfs.ext4.gz\n")
with mock.patch.dict(os.environ, {
"BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
}):
with self.assertRaises(Die) as ctx:
ia.ensure_artifact_gz("v1")
ia.ensure_artifact_gz("v1", role=_ROLE)
self.assertIn("checksum mismatch", str(ctx.exception.message))
def test_downloads_verifies_and_caches(self) -> None:
@@ -205,12 +224,12 @@ class TestEnsureArtifact(_CacheMixin):
gz = _gz(b"fake ext4 bytes")
patcher, net = self._serve(version, gz)
with patcher:
path = ia.ensure_artifact_gz(version)
path = ia.ensure_artifact_gz(version, role=_ROLE)
self.assertTrue(path.is_file())
self.assertEqual(gz, path.read_bytes())
first_calls = len(net.calls)
# Second call is a cache hit — no further network.
ia.ensure_artifact_gz(version)
ia.ensure_artifact_gz(version, role=_ROLE)
self.assertEqual(first_calls, len(net.calls))
def test_checksum_mismatch_fails_closed(self) -> None:
@@ -219,17 +238,17 @@ class TestEnsureArtifact(_CacheMixin):
patcher, _ = self._serve(version, gz, sha_text="0" * 64 + " rootfs.ext4.gz\n")
with patcher:
with self.assertRaises(Die) as ctx:
ia.ensure_artifact_gz(version)
ia.ensure_artifact_gz(version, role=_ROLE)
self.assertIn("checksum mismatch", str(ctx.exception.message))
# nothing left cached to accidentally boot
self.assertFalse((ia._cache_root(version) / "rootfs.ext4.gz").exists())
self.assertFalse((ia._cache_root(version, _ROLE) / "rootfs.ext4.gz").exists())
def test_missing_artifact_points_at_publish(self) -> None:
version = "0000000000000000"
net = _FakeNet({}) # everything 404s
with mock.patch.object(ia.urllib.request, "urlopen", net.urlopen):
with self.assertRaises(Die) as ctx:
ia.ensure_artifact_gz(version)
ia.ensure_artifact_gz(version, role=_ROLE)
self.assertIn("publish_infra", str(ctx.exception.message))
def test_materialize_gunzips_to_dest(self) -> None:
@@ -238,7 +257,7 @@ class TestEnsureArtifact(_CacheMixin):
patcher, _ = self._serve(version, _gz(raw))
with patcher, tempfile.TemporaryDirectory() as d:
dest = Path(d) / "rootfs.ext4"
ia.materialize_ext4(version, dest)
ia.materialize_ext4(version, dest, role=_ROLE)
self.assertEqual(raw, dest.read_bytes())
@@ -248,10 +267,10 @@ class TestConfig(unittest.TestCase):
"BOT_BOTTLE_INFRA_ARTIFACT_BASE": "https://mirror.example/",
"BOT_BOTTLE_INFRA_ARTIFACT_OWNER": "acme",
}):
url = ia.artifact_url("v1", "rootfs.ext4.gz")
url = ia.artifact_url("v1", "rootfs.ext4.gz", role="gateway")
self.assertEqual(
"https://mirror.example/api/packages/acme/generic/"
"bot-bottle-firecracker-infra/v1/rootfs.ext4.gz", url)
"bot-bottle-firecracker-gateway/v1/rootfs.ext4.gz", url)
def test_local_build_flag(self) -> None:
with mock.patch.dict(os.environ, {"BOT_BOTTLE_INFRA_BUILD": "local"}):
tests/unit/test_publish_infra.py
+74 -80
View File
@@ -61,155 +61,149 @@ class TestPut(unittest.TestCase):
self.assertEqual(b"abc123 rootfs\n", captured[0].data)
_ROLE = "orchestrator"
class TestPublishBundle(unittest.TestCase):
def _bundle(self, root: Path, version: str) -> None:
def _bundle(self, role_dir: Path, version: str) -> None:
payload = b"candidate"
(root / "version.txt").write_text(version + "\n")
(root / "rootfs.ext4.gz").write_bytes(payload)
role_dir.mkdir(parents=True, exist_ok=True)
(role_dir / "version.txt").write_text(version + "\n")
(role_dir / "rootfs.ext4.gz").write_bytes(payload)
digest = hashlib.sha256(payload).hexdigest()
(root / "rootfs.ext4.gz.sha256").write_text(
(role_dir / "rootfs.ext4.gz.sha256").write_text(
f"{digest} rootfs.ext4.gz\n")
def test_existing_identical_artifact_is_success(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._bundle(root, "v1")
sha = (root / "rootfs.ext4.gz.sha256").read_bytes()
role_dir = Path(d) / _ROLE
self._bundle(role_dir, "v1")
sha = (role_dir / "rootfs.ext4.gz.sha256").read_bytes()
response = mock.MagicMock()
response.__enter__.return_value.read.return_value = sha
with mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="v1"
), mock.patch.object(
pub.urllib.request, "urlopen", return_value=response
), mock.patch.object(pub, "_put") as put:
self.assertEqual("v1", pub._publish_bundle(root, "token"))
with mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(
pub.urllib.request, "urlopen", return_value=response
), mock.patch.object(pub, "_put") as put:
self.assertEqual("v1", pub._publish_bundle(_ROLE, role_dir, "token"))
put.assert_not_called()
def test_partial_artifact_is_replaced(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._bundle(root, "v1")
role_dir = Path(d) / _ROLE
self._bundle(role_dir, "v1")
missing = urllib.error.HTTPError("u", 404, "missing", Message(), None)
with mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="v1"
), mock.patch.object(
pub.urllib.request, "urlopen", side_effect=missing
), mock.patch.object(pub, "_delete") as delete, \
with mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(
pub.urllib.request, "urlopen", side_effect=missing
), mock.patch.object(pub, "_delete") as delete, \
mock.patch.object(pub, "_put") as put:
pub._publish_bundle(root, "token")
pub._publish_bundle(_ROLE, role_dir, "token")
self.assertEqual(3, delete.call_count)
self.assertEqual(3, put.call_count)
def test_rejects_bundle_for_different_checkout(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._bundle(root, "old")
with mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="new"
):
role_dir = Path(d) / _ROLE
self._bundle(role_dir, "old")
with mock.patch.object(pub, "_role_version", return_value="new"):
with self.assertRaises(SystemExit) as ctx:
pub._publish_bundle(root, "token")
pub._publish_bundle(_ROLE, role_dir, "token")
self.assertIn("does not match checkout", str(ctx.exception))
def test_rejects_bad_bundle_checksum(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._bundle(root, "v1")
(root / "rootfs.ext4.gz").write_bytes(b"tampered")
with mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="v1"
):
role_dir = Path(d) / _ROLE
self._bundle(role_dir, "v1")
(role_dir / "rootfs.ext4.gz").write_bytes(b"tampered")
with mock.patch.object(pub, "_role_version", return_value="v1"):
with self.assertRaises(SystemExit) as ctx:
pub._publish_bundle(root, "token")
pub._publish_bundle(_ROLE, role_dir, "token")
self.assertIn("checksum mismatch", str(ctx.exception))
def test_registry_lookup_failure_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d)
self._bundle(root, "v1")
role_dir = Path(d) / _ROLE
self._bundle(role_dir, "v1")
failure = urllib.error.URLError("offline")
with mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="v1"
), mock.patch.object(pub.urllib.request, "urlopen", side_effect=failure):
with mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(pub.urllib.request, "urlopen", side_effect=failure):
with self.assertRaises(SystemExit) as ctx:
pub._publish_bundle(root, "token")
pub._publish_bundle(_ROLE, role_dir, "token")
self.assertIn("registry unreachable", str(ctx.exception))
class TestTryDownloadPublished(unittest.TestCase):
def test_downloads_existing_artifact(self) -> None:
with tempfile.TemporaryDirectory() as d, \
mock.patch.object(pub.infra_vm, "_infra_init", return_value="init"), \
mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(
pub.infra_artifact, "infra_artifact_version", return_value="v1"
), mock.patch.object(
pub.urllib.request, "urlopen", return_value=_Resp()
), mock.patch.object(pub.infra_artifact, "_download") as download:
root = Path(d)
result = pub._try_download_published(root)
self.assertEqual(
("v1", root / "rootfs.ext4.gz", root / "rootfs.ext4.gz.sha256"),
result,
)
self.assertEqual(2, download.call_count)
role_dir = Path(d) / _ROLE
role_dir.mkdir()
result = pub._try_download_published(_ROLE, role_dir)
self.assertEqual("v1", result)
self.assertEqual("v1\n", (role_dir / "version.txt").read_text())
self.assertEqual(2, download.call_count)
def test_missing_artifact_returns_none(self) -> None:
missing = urllib.error.HTTPError("u", 404, "missing", Message(), None)
with tempfile.TemporaryDirectory() as d, mock.patch.object(
pub.urllib.request, "urlopen", side_effect=missing
):
self.assertIsNone(pub._try_download_published(Path(d)))
with tempfile.TemporaryDirectory() as d, \
mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(pub.urllib.request, "urlopen", side_effect=missing):
self.assertIsNone(pub._try_download_published(_ROLE, Path(d)))
def test_registry_http_failure_is_reported(self) -> None:
failure = urllib.error.HTTPError("u", 500, "failed", Message(), None)
with tempfile.TemporaryDirectory() as d, mock.patch.object(
pub.urllib.request, "urlopen", side_effect=failure
):
with tempfile.TemporaryDirectory() as d, \
mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(pub.urllib.request, "urlopen", side_effect=failure):
with self.assertRaises(SystemExit) as ctx:
pub._try_download_published(Path(d))
pub._try_download_published(_ROLE, Path(d))
self.assertIn("registry check failed (HTTP 500)", str(ctx.exception))
def test_registry_connection_failure_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as d, mock.patch.object(
pub.urllib.request, "urlopen", side_effect=urllib.error.URLError("offline")
):
with tempfile.TemporaryDirectory() as d, \
mock.patch.object(pub, "_role_version", return_value="v1"), \
mock.patch.object(
pub.urllib.request, "urlopen",
side_effect=urllib.error.URLError("offline")):
with self.assertRaises(SystemExit) as ctx:
pub._try_download_published(Path(d))
pub._try_download_published(_ROLE, Path(d))
self.assertIn("registry unreachable", str(ctx.exception))
class TestMain(unittest.TestCase):
def test_output_builds_candidate_and_records_version(self) -> None:
def test_output_builds_a_candidate_per_role(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d) / "candidate"
with mock.patch.object(
pub, "build_artifact", return_value=("v1", root / "g", root / "s")
) as build:
with mock.patch.object(pub.infra_vm, "build_infra_images_with_docker") as images, \
mock.patch.object(pub, "build_role_artifact", return_value="v1") as build:
self.assertEqual(0, pub.main(["--output", str(root)]))
build.assert_called_once_with(root)
self.assertEqual("v1\n", (root / "version.txt").read_text())
images.assert_called_once()
built_roles = {c.args[0] for c in build.call_args_list}
self.assertEqual({"orchestrator", "gateway"}, built_roles)
def test_output_reuses_published_candidate(self) -> None:
def test_output_reuses_published_candidates(self) -> None:
with tempfile.TemporaryDirectory() as d:
root = Path(d) / "candidate"
reused = ("v1", root / "rootfs.ext4.gz", root / "rootfs.ext4.gz.sha256")
with mock.patch.object(
pub, "_try_download_published", return_value=reused
) as reuse, mock.patch.object(pub, "build_artifact") as build:
with mock.patch.object(pub, "_try_download_published", return_value="v1") as reuse, \
mock.patch.object(pub.infra_vm, "build_infra_images_with_docker") as images, \
mock.patch.object(pub, "build_role_artifact") as build:
self.assertEqual(
0, pub.main(["--output", str(root), "--reuse-published"])
)
reuse.assert_called_once_with(root)
0, pub.main(["--output", str(root), "--reuse-published"]))
self.assertEqual(2, reuse.call_count) # once per role
images.assert_not_called()
build.assert_not_called()
self.assertEqual("v1\n", (root / "version.txt").read_text())
def test_publish_dir_publishes_existing_candidate(self) -> None:
def test_publish_dir_publishes_both_roles(self) -> None:
with tempfile.TemporaryDirectory() as d, \
mock.patch.object(pub.infra_artifact, "_config", return_value=("", "", "t")), \
mock.patch.object(pub, "_publish_bundle", return_value="v1") as publish:
self.assertEqual(0, pub.main(["--publish-dir", d]))
publish.assert_called_once_with(Path(d), "t")
published_roles = {c.args[0] for c in publish.call_args_list}
self.assertEqual({"orchestrator", "gateway"}, published_roles)
if __name__ == "__main__":