ci: backend-agnostic integration guards + per-backend preflight #470

Merged
didericis merged 8 commits from ci/backend-agnostic-integration-414 into main 2026-07-24 21:25:23 -04:00
23 changed files with 632 additions and 129 deletions
+8 -6
View File
@@ -103,14 +103,16 @@ jobs:
- name: Install coverage
run: python3 -m pip install --break-system-packages coverage
- name: Show environment
# Fail loudly if the backend this job promises isn't actually usable,
# rather than letting every test silently `unittest.skip` and the job
# go green on zero coverage. `backend status` prints a clear per-check
# summary (docker on PATH, daemon reachable) and exits non-zero when a
# prerequisite is missing — the same readiness check the skip guards
# gate on via `has_backend`.
- name: Preflight — Docker backend is ready
run: |
python3 --version
if command -v docker >/dev/null 2>&1; then
docker version || true
else
echo "docker not on PATH — integration tests will skip"
fi
python3 cli.py backend status --backend=docker
- name: Run integration tests (docker) with coverage
env:
+43 -4
View File
@@ -33,6 +33,7 @@ backend field; the host picks.
from __future__ import annotations
import enum
import os
import shlex
import sys
@@ -59,6 +60,12 @@ if TYPE_CHECKING:
from .freeze import CommitCancelled, Freezer, get_freezer
class BackendStatus(enum.IntEnum):
"""Return codes for BottleBackend.status(). READY == 0 so callsites
can compare against 0 or the named constant interchangeably."""
READY = 0
@dataclass(frozen=True)
class BottleSpec:
"""CLI-supplied intent. Backend-agnostic — each backend's prepare
@@ -611,12 +618,18 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
@classmethod
@abstractmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
"""Report whether this backend's prerequisites are satisfied on
the host — binaries, daemon reachability, network pool, range
conflicts, etc. Prints a human-readable summary; returns 0 when
the backend is ready to launch and non-zero when something is
missing. Invoked by `./cli.py backend status [--backend=…]`."""
conflicts, etc. Returns BackendStatus.READY (0) when the backend
is ready to launch and non-zero when something is missing.
When quiet=False (default) prints a human-readable summary to
stderr. When quiet=True returns the status code silently —
useful for cheap programmatic checks.
Invoked by `./cli.py backend status [--backend=…]` (quiet=False)
and by is_backend_ready() (caller-controlled)."""
@classmethod
@abstractmethod
@@ -811,6 +824,29 @@ def has_backend(name: str) -> bool:
return backends[name].is_available()
def is_backend_available(name: str) -> bool:
"""Cheap availability check: is the backend's binary on PATH?
Suitable for cleanup enumeration and auto-selection — does NOT probe
the daemon or network pool. Use is_backend_ready() for a full
readiness check before launching tests."""
return has_backend(name)
def is_backend_ready(name: str, *, quiet: bool = False) -> bool:
"""Full readiness check: passes all of the backend's status() checks.
When quiet=False the backend prints diagnostic output explaining what
is missing — intended for test-suite guards that run at discovery time
so the operator sees a concrete failure reason for each skip.
Returns False for unknown backend names."""
backends = _get_backends()
if name not in backends:
return False
return backends[name].status(quiet=quiet) == BackendStatus.READY
def enumerate_active_agents() -> list[ActiveAgent]:
"""All currently-running agents, across every available
backend. Used by CLI `list active` and the dashboard's agents
@@ -835,6 +871,7 @@ def enumerate_active_agents() -> list[ActiveAgent]:
__all__ = [
"ActiveAgent",
"BackendStatus",
"Bottle",
"BottleBackend",
"BottleCleanupPlan",
@@ -847,5 +884,7 @@ __all__ = [
"get_bottle_backend",
"get_freezer",
"has_backend",
"is_backend_available",
"is_backend_ready",
"known_backend_names",
]
+6 -2
View File
@@ -20,7 +20,8 @@ infrastructure: CA install and git copy-in.
from __future__ import annotations
import shutil
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -60,8 +61,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+6 -2
View File
@@ -8,7 +8,8 @@ fail-closed nftables egress boundary. Selected by
from __future__ import annotations
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -52,8 +53,11 @@ class FirecrackerBottleBackend(
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+86 -6
View File
@@ -13,6 +13,7 @@ generic `./cli.py backend {setup,status}` command dispatches to.
from __future__ import annotations
import fcntl
import os
import shutil
import subprocess
@@ -22,6 +23,9 @@ from pathlib import Path
from . import netpool
from . import util
# KVM_GET_API_VERSION = _IO(KVMIO=0xAE, 0x00): cheapest proof of KVM access.
_KVM_GET_API_VERSION = 0xAE00
_FC_RELEASES = "https://github.com/firecracker-microvm/firecracker/releases"
_UNIT_PATH = Path("/etc/systemd/system") / netpool.SYSTEMD_UNIT
@@ -219,14 +223,90 @@ def teardown() -> int:
return 0
def _firecracker_binary_ok() -> bool:
"""True iff the firecracker binary is on PATH and `--version` exits 0."""
if shutil.which("firecracker") is None:
return False
try:
return subprocess.run(
["firecracker", "--version"],
capture_output=True, check=False, timeout=5,
).returncode == 0
except (OSError, subprocess.TimeoutExpired):
return False
def _kvm_accessible() -> bool:
"""True iff /dev/kvm can be opened read-write and responds to KVM_GET_API_VERSION.
VM creation requires write access; opening read-only may satisfy the
ioctl but fails at boot time, so O_RDWR is the permission check."""
if not os.path.exists(util._KVM_DEVICE):
return False
try:
fd = os.open(util._KVM_DEVICE, os.O_RDWR | os.O_CLOEXEC)
try:
fcntl.ioctl(fd, _KVM_GET_API_VERSION)
finally:
os.close(fd)
return True
except OSError:
return False
def status() -> int:
# Readiness == what the launch preflight hard-requires: the TAP pool
# present (unprivileged, authoritative) and no range overlap. Listing
# the nft table usually needs root, so — like the preflight — an
# unconfirmable table is reported but NOT treated as not-ready; the
# post-boot isolation probe is the authoritative check. This keeps an
# unprivileged `backend status` usable as a launch gate.
# Readiness == what the launch preflight hard-requires: the binary
# executable, /dev/kvm accessible, the TAP pool present, and no range
# overlap. Listing the nft table usually needs root, so — like the
# preflight — an unconfirmable table is reported but NOT treated as
# not-ready; the post-boot isolation probe is the authoritative check.
# This keeps an unprivileged `backend status` usable as a launch gate.
ok = True
if _firecracker_binary_ok():
sys.stderr.write(f"firecracker binary: ok ({shutil.which('firecracker')})\n")
else:
fc_path = shutil.which("firecracker")
if fc_path is None:
sys.stderr.write("firecracker binary: NOT found on PATH\n")
else:
sys.stderr.write(
f"firecracker binary: found ({fc_path}) but `--version` failed\n"
)
ok = False
if _kvm_accessible():
sys.stderr.write(f"KVM: {util._KVM_DEVICE} accessible\n")
else:
if not os.path.exists(util._KVM_DEVICE):
sys.stderr.write(f"KVM: {util._KVM_DEVICE} not present\n")
else:
sys.stderr.write(
f"KVM: {util._KVM_DEVICE} not accessible (open/ioctl failed)\n"
)
ok = False
kernel = util.kernel_path()
if kernel.is_file():
sys.stderr.write(f"guest kernel: {kernel}\n")
else:
sys.stderr.write(
f"guest kernel: NOT found at {kernel} "
f"(set BOT_BOTTLE_FC_KERNEL or cache a vmlinux there)\n"
)
ok = False
dropbear = util.dropbear_path()
if dropbear.is_file():
sys.stderr.write(f"dropbear: {dropbear}\n")
else:
sys.stderr.write(
f"dropbear: NOT found at {dropbear} "
f"(set BOT_BOTTLE_FC_DROPBEAR or cache a static binary)\n"
)
ok = False
mke2fs = shutil.which("mke2fs")
if mke2fs is not None:
sys.stderr.write(f"mke2fs: {mke2fs}\n")
else:
sys.stderr.write("mke2fs: NOT found on PATH (install e2fsprogs)\n")
ok = False
missing = netpool.missing_taps()
total = netpool.pool_size()
if missing:
@@ -2,7 +2,8 @@
from __future__ import annotations
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -43,8 +44,11 @@ class MacosContainerBottleBackend(
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+13 -3
View File
@@ -1,13 +1,23 @@
# CI
The test workflow lives at [`.gitea/workflows/test.yml`](../.gitea/workflows/test.yml).
It runs `tests/run_tests.py` (full suite — unit + integration) on:
It runs the unit suite plus one integration job per backend
(`integration-docker`, `integration-firecracker`) on:
- every push to a branch with an open pull request, and
- every push to `main`.
Integration tests need Docker on the runner; they skip cleanly via
`tests/_docker.skip_unless_docker` when no daemon is reachable.
Each integration job selects its backend via `BOT_BOTTLE_BACKEND` and
runs a **preflight** (`./cli.py backend status --backend=<name>`) that
prints a clear per-check readiness summary and fails the job when the
backend is missing — so absent infrastructure is visible at the job level
rather than hidden among per-test `unittest.skip` lines. The skip guards in
[`tests/_backend.py`](../tests/_backend.py) gate on the same readiness
check (`bot_bottle.backend.has_backend`): backend-agnostic tests use
`skip_unless_selected_backend_available()` and run through whichever
backend is selected (checking, e.g., Linux + `/dev/kvm` for Firecracker
rather than unrelated Docker availability); Docker-implementation tests use
`skip_unless_backend("docker")` and no-op under a non-Docker run.
A small subset of integration tests skip when running specifically
under Gitea Actions (`GITEA_ACTIONS=true`), because `act_runner` runs
+20 -5
View File
@@ -2,14 +2,15 @@
Plain-Python test suite using stdlib `unittest`. No external
dependencies. Unit tests run anywhere Python 3 is present; integration
tests need Docker and skip cleanly otherwise.
tests run through the backend named by `BOT_BOTTLE_BACKEND` (default
`docker`) and skip cleanly when that backend isn't available on the host.
## Layout
```
tests/
fixtures.py # JSON manifest builders (shared)
_docker.py # docker-availability skip helper (shared)
_backend.py # backend selection + skip guards (shared)
unit/
test_egress.py
test_egress_addon_core.py
@@ -73,7 +74,7 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
## Adding a test
1. Pick the directory: `tests/unit/` for a pure unit test,
`tests/integration/` for one that needs Docker.
`tests/integration/` for one that needs a backend.
2. Filename: `test_<topic>.py`.
3. Boilerplate:
```python
@@ -88,5 +89,19 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
if __name__ == "__main__":
unittest.main()
```
4. For Docker-dependent tests, decorate the class with
`@skip_unless_docker()` from `tests._docker`.
4. Skip guards live in `tests._backend` and gate on the backend's own
readiness check, `bot_bottle.backend.has_backend` — the same probe
behind `./cli.py backend status`:
- Backend-agnostic tests (go through `get_bottle_backend()`) decorate
the class with `@skip_unless_selected_backend_available()` — the test
runs against whichever backend `BOT_BOTTLE_BACKEND` selects and skips
unless that backend is available (checking, e.g., Linux + `/dev/kvm`
for Firecracker rather than unrelated Docker availability).
- Backend-specific tests (exercise `DockerBroker`, `DockerGateway`,
`backend.docker.*`, …) decorate with `@skip_unless_backend("docker")`
so they no-op under a run targeting a different backend.
Each CI integration job runs `./cli.py backend status --backend=<name>`
as a preflight, which prints a clear per-check summary and exits non-zero
when the backend is missing — so absent infrastructure fails the job
instead of hiding among per-test `unittest.skip` lines.
+71
View File
@@ -0,0 +1,71 @@
"""Backend selection + readiness-aware skip guards for the integration suite.
Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND``
(default ``docker``) and gates on that backend's full readiness check —
``is_backend_ready()`` (equivalent to ``./cli.py backend status``), not just
a binary-on-PATH probe. When the backend is not ready, diagnostic output is
printed during test discovery so the operator sees a concrete reason for each
skip.
"""
from __future__ import annotations
import os
import unittest
from bot_bottle.backend import is_backend_ready
# Default when ``BOT_BOTTLE_BACKEND`` is unset. Docker preserves the historical
# Docker-backed CI path (and mirrors the pin in ``test_sandbox_escape``).
DEFAULT_BACKEND = "docker"
def selected_backend() -> str:
"""The backend this test run targets, from ``BOT_BOTTLE_BACKEND``.
Mirrors the CLI's env selector; unset means ``docker`` so an
unconfigured run behaves exactly as the suite did before backends were
pluggable.
"""
return os.environ.get("BOT_BOTTLE_BACKEND") or DEFAULT_BACKEND
def skip_unless_backend(backend: str):
"""Skip a backend-specific test unless the selected backend matches AND
that backend is fully ready.
Docker-implementation tests (``DockerBroker``, ``DockerGateway``,
``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op
under a run targeting a different backend instead of testing Docker
internals that run doesn't exercise — the guard reads
``BOT_BOTTLE_BACKEND`` rather than "is Docker installed".
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
sel = selected_backend()
if sel != backend:
return unittest.skip(
f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})"
)
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"{backend} backend not ready",
)
def skip_unless_selected_backend_available():
"""Skip a backend-agnostic test unless the *selected* backend is fully ready.
The test then runs through whichever backend ``BOT_BOTTLE_BACKEND`` names,
gated on that backend's full status() check (e.g. daemon reachable, TAP
pool present for Firecracker) rather than just a binary-on-PATH probe.
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
backend = selected_backend()
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"selected backend {backend!r} not ready",
)
-45
View File
@@ -1,45 +0,0 @@
"""Docker availability check used by integration tests."""
from __future__ import annotations
import os
import shutil
import subprocess
import unittest
def docker_available() -> bool:
if os.environ.get("SKIP_DOCKER_TESTS"):
return False
if shutil.which("docker") is None:
return False
try:
return (
subprocess.run(
["docker", "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=5,
).returncode
== 0
)
except subprocess.TimeoutExpired:
return False
def skip_unless_docker(reason: str = "docker unreachable"):
return unittest.skipUnless(docker_available(), reason)
def skip_unless_docker_or_firecracker(
reason: str = "neither Docker nor Firecracker selected",
):
"""Skip a backend-agnostic test unless one supported backend can run.
Firecracker does not require the host Docker daemon. The KVM coverage job
deliberately sets ``SKIP_DOCKER_TESTS`` to exclude Docker-only integration
classes while still exercising this path.
"""
firecracker_selected = os.environ.get("BOT_BOTTLE_BACKEND") == "firecracker"
return unittest.skipUnless(firecracker_selected or docker_available(), reason)
+2 -2
View File
@@ -25,14 +25,14 @@ import os
import subprocess
import unittest
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
_IMAGE = "bot-bottle-gateway-test:chunk1"
_DOCKERFILE = "Dockerfile.gateway"
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: multi-stage build pulls a 200+MB "
@@ -31,7 +31,7 @@ from bot_bottle.backend.docker.gateway_net import next_free_ip
from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK
from bot_bottle.orchestrator.lifecycle import OrchestratorService
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
# One upstream reachable under two names; reflects the Authorization header so
# the probe can see exactly what the gateway injected (or didn't).
@@ -72,7 +72,7 @@ _PROBE_SRC = (
)
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -13,12 +13,12 @@ import unittest
from bot_bottle.orchestrator.broker import LaunchRequest, sign_request
from bot_bottle.orchestrator.docker_broker import DockerBroker, container_name
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerBrokerIntegration(unittest.TestCase):
def setUp(self) -> None:
self.secret = secrets.token_bytes(16)
@@ -28,7 +28,7 @@ from pathlib import Path
from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.lifecycle import OrchestratorService
from bot_bottle.paths import host_control_plane_token
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
# Fixed (not per-run-suffixed) so repeated runs reuse the same layer-cached
# image instead of leaking a new dangling tag on every invocation.
@@ -37,7 +37,7 @@ _TEST_GATEWAY_IMAGE = "bot-bottle-gateway:itest"
_TEST_INFRA_IMAGE = "bot-bottle-infra:itest"
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -11,12 +11,12 @@ import subprocess
import unittest
from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerGatewayIntegration(unittest.TestCase):
def setUp(self) -> None:
self.name = "bot-bottle-orch-gateway-itest-" + secrets.token_hex(4)
@@ -12,12 +12,12 @@ import subprocess
import unittest
from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerGatewayImageExists(unittest.TestCase):
def test_image_exists_true_for_present_false_for_absent(self) -> None:
# Ensure the tiny image is present (build_if_missing is disabled here
+2 -2
View File
@@ -18,10 +18,10 @@ from bot_bottle.backend.docker.network import (
network_create_internal,
network_remove,
)
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
@skip_unless_docker()
@skip_unless_backend("docker")
class TestOrphanCleanup(unittest.TestCase):
def setUp(self):
self.slug = f"cb-test-orphan-{os.getpid()}"
+2 -2
View File
@@ -31,7 +31,7 @@ from pathlib import Path
from bot_bottle.backend import BottleSpec, get_bottle_backend
from bot_bottle.bottle_state import cleanup_state
from bot_bottle.manifest import ManifestIndex
from tests._docker import skip_unless_docker_or_firecracker
from tests._backend import skip_unless_selected_backend_available
# Secrets planted in the bottle env as literals (agents substitute via
@@ -67,7 +67,7 @@ _DUMMY_HOST_KEY = (
)
@skip_unless_docker_or_firecracker()
@skip_unless_selected_backend_available()
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true"
and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
+54
View File
@@ -385,6 +385,60 @@ class TestHasBackend(unittest.TestCase):
self.assertFalse(has_backend("nonexistent"))
class TestIsBackendAvailable(unittest.TestCase):
def test_delegates_to_has_backend(self):
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_available("docker"))
def test_known_and_available(self):
class _Ready:
def is_available(self):
return True
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {"docker": _Ready()}):
self.assertTrue(is_backend_available("docker"))
class TestIsBackendReady(unittest.TestCase):
def test_unknown_backend_returns_false(self):
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_ready("docker"))
def test_ready_when_status_returns_zero(self):
class _ReadyBackend:
def status(self, *, quiet: bool = False) -> int:
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _ReadyBackend()}):
self.assertTrue(is_backend_ready("docker"))
def test_not_ready_when_status_nonzero(self):
class _BrokenBackend:
def status(self, *, quiet: bool = False) -> int:
return 1
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _BrokenBackend()}):
self.assertFalse(is_backend_ready("docker"))
def test_quiet_flag_forwarded(self):
calls = []
class _SpyBackend:
def status(self, *, quiet: bool = False) -> int:
calls.append(quiet)
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _SpyBackend()}):
is_backend_ready("docker", quiet=True)
self.assertEqual([True], calls)
class TestEnsureOrchestrator(unittest.TestCase):
"""The backend-agnostic orchestrator bring-up entry point. Docker starts
the orchestrator + gateway containers; firecracker boots the infra VM;
+196
View File
@@ -328,5 +328,201 @@ class TestNetpoolShellRenderers(unittest.TestCase):
self.assertIn("BOT_BOTTLE_FC_POOL_SIZE=4", out)
class TestFirecrackerBinaryCheck(unittest.TestCase):
def test_binary_missing_returns_false(self):
with patch.object(fc.shutil, "which", return_value=None):
self.assertFalse(fc._firecracker_binary_ok())
def test_binary_present_and_runs_ok(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run",
return_value=subprocess.CompletedProcess([], 0)):
self.assertTrue(fc._firecracker_binary_ok())
def test_binary_found_but_exits_nonzero(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run",
return_value=subprocess.CompletedProcess([], 1)):
self.assertFalse(fc._firecracker_binary_ok())
def test_binary_found_but_oserror(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run", side_effect=OSError("exec failed")):
self.assertFalse(fc._firecracker_binary_ok())
class TestFirecrackerKvmCheck(unittest.TestCase):
def test_kvm_device_absent_returns_false(self):
with patch.object(fc.os.path, "exists", return_value=False):
self.assertFalse(fc._kvm_accessible())
def test_kvm_accessible_when_ioctl_succeeds(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", return_value=42), \
patch.object(fc.os, "close"), \
patch.object(fc.fcntl, "ioctl", return_value=12):
self.assertTrue(fc._kvm_accessible())
def test_kvm_open_rdwr_fails(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", side_effect=OSError("Permission denied")):
self.assertFalse(fc._kvm_accessible())
def test_kvm_present_but_ioctl_fails(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", return_value=42), \
patch.object(fc.os, "close"), \
patch.object(fc.fcntl, "ioctl", side_effect=OSError("permission denied")):
self.assertFalse(fc._kvm_accessible())
class TestFirecrackerArtifactCheck(unittest.TestCase):
"""status() reports missing guest kernel, dropbear binary, and mke2fs."""
def _apply_all_ok(self, stack: contextlib.ExitStack) -> None:
"""Stub every status() check to pass except what the test overrides."""
stack.enter_context(patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(patch.object(fc, "_kvm_accessible", return_value=True))
k: MagicMock = MagicMock()
k.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
d: MagicMock = MagicMock()
d.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
stack.enter_context(patch.object(fc.shutil, "which", return_value="/usr/bin/x"))
stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
def test_status_fails_when_kernel_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
k: MagicMock = MagicMock()
k.is_file.return_value = False
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("guest kernel: NOT found", out)
def test_status_fails_when_dropbear_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
d: MagicMock = MagicMock()
d.is_file.return_value = False
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("dropbear: NOT found", out)
def test_status_fails_when_mke2fs_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
stack.enter_context(patch.object(
fc.shutil, "which",
side_effect=lambda cmd: (None if cmd == "mke2fs" else "/usr/bin/x"), # type: ignore[misc]
))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("mke2fs: NOT found", out)
class TestFirecrackerStatusRuntime(unittest.TestCase):
"""status() reports binary and KVM problems and returns non-zero."""
def _apply_pool_ok(self, stack: contextlib.ExitStack) -> None:
kernel_mock = MagicMock()
kernel_mock.is_file.return_value = True
dropbear_mock = MagicMock()
dropbear_mock.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=kernel_mock))
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=dropbear_mock))
stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
def test_status_fails_when_binary_missing(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=False))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value=None))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("NOT found on PATH", out)
def test_status_fails_when_kvm_not_accessible(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=False))
stack.enter_context(
patch.object(fc.os.path, "exists", return_value=True))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("not accessible", out)
def test_status_ok_when_binary_and_kvm_ready(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=True))
stack.enter_context(
patch.object(netpool, "nft_table_present", return_value=True))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(0, rc)
self.assertIn("firecracker binary: ok", out)
self.assertIn("KVM:", out)
class TestBackendStatusQuiet(unittest.TestCase):
"""status(quiet=True) routes the underlying status() call through
redirect_stderr so diagnostic output is suppressed. Verify the return
code is propagated and that calling with quiet=False leaves the non-quiet
path active (covered by the other TestDockerSetupStatus tests)."""
def test_docker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.docker.backend import DockerBottleBackend
with patch.object(dk, "status", return_value=0):
self.assertEqual(0, DockerBottleBackend.status(quiet=True))
def test_docker_quiet_true_suppresses_stderr(self):
import io as _io
from bot_bottle.backend.docker.backend import DockerBottleBackend
def _loud_status():
import sys
print("should be suppressed", file=sys.stderr)
return 0
with patch.object(dk, "status", side_effect=_loud_status):
buf = _io.StringIO()
with contextlib.redirect_stderr(buf):
DockerBottleBackend.status(quiet=True)
self.assertEqual("", buf.getvalue())
def test_firecracker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.firecracker.backend import FirecrackerBottleBackend
with patch.object(fc, "status", return_value=1):
self.assertEqual(1, FirecrackerBottleBackend.status(quiet=True))
def test_macos_quiet_true_propagates_return_code(self):
from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
with patch.object(mc, "status", return_value=0):
self.assertEqual(0, MacosContainerBottleBackend.status(quiet=True))
if __name__ == "__main__":
unittest.main()
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the backend-aware skip guards in ``tests/_backend.py``.
The guards delegate their readiness check to
``bot_bottle.backend.is_backend_ready`` (the probe behind ``./cli.py backend
status``); here that probe is mocked so the unit job asserts the
selection/skip logic without either backend present on the runner.
"""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._backend import (
selected_backend,
skip_unless_backend,
skip_unless_selected_backend_available,
)
def _skipped(decorated: type) -> bool:
return getattr(decorated, "__unittest_skip__", False)
def _new_case() -> type:
return type("Case", (unittest.TestCase,), {})
class TestSelectedBackend(unittest.TestCase):
def test_defaults_to_docker_when_unset(self):
with patch.dict(os.environ, {}, clear=True):
self.assertEqual("docker", selected_backend())
def test_reads_env(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True):
self.assertEqual("firecracker", selected_backend())
class TestSkipUnlessBackend(unittest.TestCase):
def test_skips_when_other_backend_selected(self):
# A different backend is selected — no host probe needed, skip.
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready") as has:
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
has.assert_not_called()
def test_runs_when_selected_and_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True):
decorated = skip_unless_backend("docker")(_new_case())
self.assertFalse(_skipped(decorated))
def test_skips_when_selected_but_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase):
def test_runs_when_selected_backend_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True) as has:
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertFalse(_skipped(decorated))
has.assert_called_once_with("firecracker", quiet=False)
def test_skips_when_selected_backend_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertTrue(_skipped(decorated))
if __name__ == "__main__":
unittest.main()
-35
View File
@@ -1,35 +0,0 @@
"""Tests for integration-test backend selection helpers."""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._docker import skip_unless_docker_or_firecracker
class TestSkipUnlessDockerOrFirecracker(unittest.TestCase):
def test_firecracker_runs_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "firecracker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertFalse(getattr(decorated, "__unittest_skip__", False))
def test_non_firecracker_still_skips_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "docker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertTrue(getattr(decorated, "__unittest_skip__", False))
if __name__ == "__main__":
unittest.main()
+29 -3
View File
@@ -132,30 +132,56 @@ class TestFirecrackerStatus(unittest.TestCase):
rc = fc_setup.status()
return rc, buf.getvalue()
def _stub_artifacts(self, fc_setup: object) -> tuple[MagicMock, MagicMock]:
k: MagicMock = MagicMock()
k.is_file.return_value = True
d: MagicMock = MagicMock()
d.is_file.return_value = True
return k, d
def test_ready_when_taps_present_even_if_nft_unverifiable(self):
from bot_bottle.backend.firecracker import setup as fc_setup
def _which(cmd: str) -> str | None:
return None if cmd == "nft" else f"/usr/bin/{cmd}"
k, d = self._stub_artifacts(fc_setup)
with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", side_effect=_which), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, out = self._run()
self.assertEqual(0, rc)
self.assertIn("unverified", out)
def test_not_ready_when_taps_missing(self):
from bot_bottle.backend.firecracker import setup as fc_setup
k, d = self._stub_artifacts(fc_setup)
with patch.object(fc_setup.netpool, "missing_taps", return_value=["bbfc0"]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", return_value=None), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, _ = self._run()
self.assertEqual(1, rc)
def test_not_ready_on_range_overlap(self):
from bot_bottle.backend.firecracker import netpool
from bot_bottle.backend.firecracker import setup as fc_setup
k, d = self._stub_artifacts(fc_setup)
conflict = netpool.RouteConflict(dst="10.243.0.0/24", dev="eth0")
with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[conflict]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", return_value=None), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, out = self._run()
self.assertEqual(1, rc)
self.assertIn("CLASHES", out)