From 9fdaba4bd4adc3408121070150f30a9ab0c7b694 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 01:15:45 +0000 Subject: [PATCH 1/8] ci: backend-agnostic integration guards + per-backend preflight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integration tests now select their backend from BOT_BOTTLE_BACKEND and skip on the capability that backend actually needs, instead of gating every backend on unrelated Docker availability. Task 1 — backend-agnostic guards (tests/_backend.py): - Capability probes: docker_capability() (reachable daemon) and firecracker_capability() (accessible /dev/kvm + firecracker on PATH, Docker-independent). backend_capability()/selected_backend() resolve the target from BOT_BOTTLE_BACKEND (default docker). - skip_unless_selected_backend_available() for backend-agnostic tests (test_sandbox_escape) — runs through whichever backend is selected and checks that backend's real capability. - skip_unless_backend("docker") for Docker-implementation tests (DockerBroker, DockerGateway, backend.docker.*) — they no-op under a non-Docker run rather than testing internals that run doesn't target. - Retires tests/_docker.py; the KVM job no longer needs SKIP_DOCKER_TESTS to steer Docker-only classes. Task 2 — explicit per-backend skip visibility: - tests/backend_preflight.py prints a clear PASS/FAIL capability line and exits non-zero when the selected backend is missing. - Both integration jobs run it as a preflight, so absent infrastructure is surfaced at the job level instead of hidden among unittest.skip lines. The docker job replaces its soft "Show environment" step; the firecracker job keeps its richer backend-status check. Docs (tests/README.md, docs/ci.md) updated; unit coverage for the probes, guards, and preflight in test_backend_skip_guards.py. Closes #414 Co-Authored-By: Claude Opus 4.8 --- .gitea/workflows/test.yml | 22 +- docs/ci.md | 16 +- tests/README.md | 23 ++- tests/_backend.py | 153 ++++++++++++++ tests/_docker.py | 45 ----- tests/backend_preflight.py | 27 +++ tests/integration/test_gateway_image.py | 4 +- .../integration/test_multitenant_isolation.py | 4 +- .../test_orchestrator_docker_broker.py | 4 +- ..._orchestrator_docker_control_plane_auth.py | 4 +- .../test_orchestrator_docker_gateway.py | 4 +- .../test_orchestrator_docker_gateway_build.py | 4 +- tests/integration/test_orphan_cleanup.py | 4 +- tests/integration/test_sandbox_escape.py | 4 +- tests/unit/test_backend_skip_guards.py | 188 ++++++++++++++++++ tests/unit/test_docker_test_helpers.py | 35 ---- 16 files changed, 428 insertions(+), 113 deletions(-) create mode 100644 tests/_backend.py delete mode 100644 tests/_docker.py create mode 100644 tests/backend_preflight.py create mode 100644 tests/unit/test_backend_skip_guards.py delete mode 100644 tests/unit/test_docker_test_helpers.py diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index e28fc2a..f2e1a16 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -103,14 +103,15 @@ 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. Same capability probe the skip guards use. + - name: Preflight — Docker backend is ready + env: + BOT_BOTTLE_BACKEND: docker 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 -m tests.backend_preflight docker - name: Run integration tests (docker) with coverage env: @@ -157,10 +158,13 @@ jobs: uses: actions/checkout@v4 - name: Preflight — Firecracker host is ready + env: + BOT_BOTTLE_BACKEND: firecracker run: | - command -v firecracker >/dev/null || { - echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; } - test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; } + # Standardized capability line — same probe (and PASS/FAIL wording) + # the integration skip guards use, so a missing backend is visible + # at the job level rather than hidden among `unittest.skip` lines. + python3 -m tests.backend_preflight firecracker # `backend status` exits non-zero unless the TAP pool is up + no # range overlap; it prints the exact `backend setup` fix. python3 cli.py backend status --backend=firecracker diff --git a/docs/ci.md b/docs/ci.md index 51d7e8d..c0230c2 100644 --- a/docs/ci.md +++ b/docs/ci.md @@ -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** (`python3 -m tests.backend_preflight `) +that prints a clear PASS/FAIL capability line 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 same +capability probes back the skip guards in +[`tests/_backend.py`](../tests/_backend.py): backend-agnostic tests use +`skip_unless_selected_backend_available()` and run through whichever +backend is selected (checking, e.g., `/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 diff --git a/tests/README.md b/tests/README.md index 07857d5..25391fc 100644 --- a/tests/README.md +++ b/tests/README.md @@ -2,14 +2,16 @@ 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 capability probes + skip guards + backend_preflight.py # `python -m tests.backend_preflight` (CI) unit/ test_egress.py test_egress_addon_core.py @@ -73,7 +75,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_.py`. 3. Boilerplate: ```python @@ -88,5 +90,16 @@ 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`: + - 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's capability is present (a reachable Docker daemon, + or an accessible `/dev/kvm` + `firecracker` for Firecracker). + - 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. + + The same capability probes back the CI preflight, + `python -m tests.backend_preflight []`, which prints a clear + PASS/FAIL line and exits non-zero when the selected backend is missing. diff --git a/tests/_backend.py b/tests/_backend.py new file mode 100644 index 0000000..39ed57c --- /dev/null +++ b/tests/_backend.py @@ -0,0 +1,153 @@ +"""Backend-aware capability probes and skip guards for the integration suite. + +Every integration test selects its backend from ``BOT_BOTTLE_BACKEND`` +(default ``docker``) and exercises the full test through that backend. Skip +guards check the *capability* the selected backend actually needs — a +reachable Docker daemon for ``docker``, an accessible ``/dev/kvm`` plus a +``firecracker`` binary for ``firecracker`` — rather than gating every backend +on unrelated Docker availability. + +The same probes back the CI preflight (``python3 -m tests.backend_preflight``) +so a job surfaces missing infrastructure at the job level with a clear +PASS/FAIL line instead of turning every test into a silent ``unittest.skip``. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import unittest +from dataclasses import dataclass + +# 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" + +# `/dev/kvm` must exist and be openable by the invoking user for the +# Firecracker backend to boot a guest (mirrors +# ``bot_bottle.backend.firecracker.util``). +_KVM_DEVICE = "/dev/kvm" + + +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 + + +@dataclass(frozen=True) +class Capability: + """Outcome of a backend capability probe. + + ``ok`` gates the tests; ``detail`` is a one-line human string reused for + both preflight output and ``unittest.skip`` reasons so the same wording + shows up at the job level and next to a skipped test. + """ + + backend: str + ok: bool + detail: str + + +def docker_capability() -> Capability: + """Whether a Docker daemon is reachable (and not opted out).""" + name = "docker" + if os.environ.get("SKIP_DOCKER_TESTS"): + return Capability(name, False, "SKIP_DOCKER_TESTS is set") + if shutil.which("docker") is None: + return Capability(name, False, "docker not on PATH") + try: + returncode = subprocess.run( + ["docker", "info"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + timeout=5, + ).returncode + except subprocess.TimeoutExpired: + return Capability(name, False, "docker info timed out") + if returncode != 0: + return Capability(name, False, "docker daemon unreachable") + return Capability(name, True, "docker daemon reachable") + + +def firecracker_capability() -> Capability: + """Whether this host can boot a Firecracker guest: an accessible + ``/dev/kvm`` and a ``firecracker`` binary on ``PATH``. + + Deliberately does not probe unrelated Docker availability — the + Firecracker backend is Docker-independent at runtime. + """ + name = "firecracker" + if not os.path.exists(_KVM_DEVICE): + return Capability(name, False, f"{_KVM_DEVICE} missing — KVM unavailable") + if not os.access(_KVM_DEVICE, os.R_OK | os.W_OK): + return Capability( + name, False, f"{_KVM_DEVICE} not accessible — add your user to the kvm group" + ) + if shutil.which("firecracker") is None: + return Capability(name, False, "firecracker not on PATH") + return Capability(name, True, f"{_KVM_DEVICE} accessible; firecracker on PATH") + + +_PROBES = { + "docker": docker_capability, + "firecracker": firecracker_capability, +} + + +def backend_capability(backend: str | None = None) -> Capability: + """Probe ``backend`` (default: the selected backend) for readiness.""" + backend = backend or selected_backend() + probe = _PROBES.get(backend) + if probe is None: + return Capability(backend, False, f"no capability probe for backend {backend!r}") + return probe() + + +# ---- back-compat boolean probe ------------------------------------------- + + +def docker_available() -> bool: + """Boolean Docker probe (kept for callers that only need the flag).""" + return docker_capability().ok + + +# ---- skip guards --------------------------------------------------------- + + +def skip_unless_backend(backend: str): + """Skip a backend-specific test unless the selected backend matches AND + that backend's capability is present. + + Docker-implementation tests (``DockerBroker``, ``DockerGateway``, + ``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op + under a Firecracker run instead of testing Docker internals that run + doesn't target — the guard reads ``BOT_BOTTLE_BACKEND`` rather than + "is Docker installed". + """ + sel = selected_backend() + if sel != backend: + return unittest.skip( + f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})" + ) + cap = backend_capability(backend) + return unittest.skipUnless(cap.ok, f"{backend} backend unavailable: {cap.detail}") + + +def skip_unless_selected_backend_available(): + """Skip a backend-agnostic test unless the *selected* backend can run it. + + The test then exercises whichever backend ``BOT_BOTTLE_BACKEND`` names, + checking that backend's real capability (e.g. ``/dev/kvm`` for + Firecracker) rather than unrelated Docker availability. + """ + cap = backend_capability() + return unittest.skipUnless( + cap.ok, f"selected backend {cap.backend!r} unavailable: {cap.detail}" + ) diff --git a/tests/_docker.py b/tests/_docker.py deleted file mode 100644 index 4cb7a25..0000000 --- a/tests/_docker.py +++ /dev/null @@ -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) diff --git a/tests/backend_preflight.py b/tests/backend_preflight.py new file mode 100644 index 0000000..0659586 --- /dev/null +++ b/tests/backend_preflight.py @@ -0,0 +1,27 @@ +"""Backend capability preflight for the integration CI jobs. + +Run as ``python3 -m tests.backend_preflight`` (optionally with a backend name +argument; default: ``BOT_BOTTLE_BACKEND`` or ``docker``). Prints a single +clear PASS/FAIL line for the selected backend's capability and exits non-zero +on failure, so a job surfaces missing infrastructure at the job level instead +of hiding it among ``unittest.skip`` lines further down the output. +""" + +from __future__ import annotations + +import sys + +from tests._backend import backend_capability, selected_backend + + +def main(argv: list[str] | None = None) -> int: + argv = sys.argv[1:] if argv is None else argv + backend = argv[0] if argv else selected_backend() + cap = backend_capability(backend) + status = "PASS" if cap.ok else "FAIL" + print(f"[preflight] backend={backend} {status}: {cap.detail}") + return 0 if cap.ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/integration/test_gateway_image.py b/tests/integration/test_gateway_image.py index 87e5d97..e12cb08 100644 --- a/tests/integration/test_gateway_image.py +++ b/tests/integration/test_gateway_image.py @@ -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 " diff --git a/tests/integration/test_multitenant_isolation.py b/tests/integration/test_multitenant_isolation.py index 7ed3031..e8b9ae5 100644 --- a/tests/integration/test_multitenant_isolation.py +++ b/tests/integration/test_multitenant_isolation.py @@ -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 " diff --git a/tests/integration/test_orchestrator_docker_broker.py b/tests/integration/test_orchestrator_docker_broker.py index 8c4da42..b502422 100644 --- a/tests/integration/test_orchestrator_docker_broker.py +++ b/tests/integration/test_orchestrator_docker_broker.py @@ -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) diff --git a/tests/integration/test_orchestrator_docker_control_plane_auth.py b/tests/integration/test_orchestrator_docker_control_plane_auth.py index 2c66887..8aac63c 100644 --- a/tests/integration/test_orchestrator_docker_control_plane_auth.py +++ b/tests/integration/test_orchestrator_docker_control_plane_auth.py @@ -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 " diff --git a/tests/integration/test_orchestrator_docker_gateway.py b/tests/integration/test_orchestrator_docker_gateway.py index 38e4b76..888b8da 100644 --- a/tests/integration/test_orchestrator_docker_gateway.py +++ b/tests/integration/test_orchestrator_docker_gateway.py @@ -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) diff --git a/tests/integration/test_orchestrator_docker_gateway_build.py b/tests/integration/test_orchestrator_docker_gateway_build.py index 5de444f..346e142 100644 --- a/tests/integration/test_orchestrator_docker_gateway_build.py +++ b/tests/integration/test_orchestrator_docker_gateway_build.py @@ -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 diff --git a/tests/integration/test_orphan_cleanup.py b/tests/integration/test_orphan_cleanup.py index 7094117..d170273 100644 --- a/tests/integration/test_orphan_cleanup.py +++ b/tests/integration/test_orphan_cleanup.py @@ -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()}" diff --git a/tests/integration/test_sandbox_escape.py b/tests/integration/test_sandbox_escape.py index de52e5f..e710506 100644 --- a/tests/integration/test_sandbox_escape.py +++ b/tests/integration/test_sandbox_escape.py @@ -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", diff --git a/tests/unit/test_backend_skip_guards.py b/tests/unit/test_backend_skip_guards.py new file mode 100644 index 0000000..54265a1 --- /dev/null +++ b/tests/unit/test_backend_skip_guards.py @@ -0,0 +1,188 @@ +"""Tests for the backend-aware capability probes and skip guards +(``tests/_backend.py``) and the CI preflight (``tests/backend_preflight.py``). + +The probes shell out to Docker / inspect ``/dev/kvm``; here they are driven +entirely through mocks so the unit job asserts the decision logic without +needing either backend present on the runner. +""" + +from __future__ import annotations + +import os +import unittest +from unittest.mock import patch + +from tests import _backend, backend_preflight +from tests._backend import ( + Capability, + backend_capability, + docker_capability, + firecracker_capability, + 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 TestDockerCapability(unittest.TestCase): + def test_opt_out_via_skip_docker_tests(self): + with patch.dict(os.environ, {"SKIP_DOCKER_TESTS": "1"}, clear=True): + cap = docker_capability() + self.assertFalse(cap.ok) + self.assertIn("SKIP_DOCKER_TESTS", cap.detail) + + def test_not_on_path(self): + with patch.dict(os.environ, {}, clear=True), \ + patch("tests._backend.shutil.which", return_value=None): + cap = docker_capability() + self.assertFalse(cap.ok) + self.assertIn("PATH", cap.detail) + + def test_daemon_reachable(self): + with patch.dict(os.environ, {}, clear=True), \ + patch("tests._backend.shutil.which", return_value="/usr/bin/docker"), \ + patch("tests._backend.subprocess.run") as run: + run.return_value.returncode = 0 + cap = docker_capability() + self.assertTrue(cap.ok) + + def test_daemon_unreachable(self): + with patch.dict(os.environ, {}, clear=True), \ + patch("tests._backend.shutil.which", return_value="/usr/bin/docker"), \ + patch("tests._backend.subprocess.run") as run: + run.return_value.returncode = 1 + cap = docker_capability() + self.assertFalse(cap.ok) + self.assertIn("unreachable", cap.detail) + + +class TestFirecrackerCapability(unittest.TestCase): + def test_kvm_missing(self): + with patch("tests._backend.os.path.exists", return_value=False): + cap = firecracker_capability() + self.assertFalse(cap.ok) + self.assertIn("KVM", cap.detail) + + def test_kvm_inaccessible(self): + with patch("tests._backend.os.path.exists", return_value=True), \ + patch("tests._backend.os.access", return_value=False): + cap = firecracker_capability() + self.assertFalse(cap.ok) + self.assertIn("kvm group", cap.detail) + + def test_binary_missing(self): + with patch("tests._backend.os.path.exists", return_value=True), \ + patch("tests._backend.os.access", return_value=True), \ + patch("tests._backend.shutil.which", return_value=None): + cap = firecracker_capability() + self.assertFalse(cap.ok) + self.assertIn("firecracker not on PATH", cap.detail) + + def test_ready(self): + with patch("tests._backend.os.path.exists", return_value=True), \ + patch("tests._backend.os.access", return_value=True), \ + patch("tests._backend.shutil.which", return_value="/usr/bin/firecracker"): + cap = firecracker_capability() + self.assertTrue(cap.ok) + + +class TestBackendCapability(unittest.TestCase): + def test_unknown_backend(self): + cap = backend_capability("qemu") + self.assertFalse(cap.ok) + self.assertIn("no capability probe", cap.detail) + + def test_defaults_to_selected(self): + def probe() -> Capability: + return Capability("docker", True, "ok") + + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ + patch.dict("tests._backend._PROBES", {"docker": probe}): + cap = backend_capability() + self.assertEqual("docker", cap.backend) + self.assertTrue(cap.ok) + + +class TestSkipUnlessBackend(unittest.TestCase): + def test_skips_when_other_backend_selected(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True): + decorated = skip_unless_backend("docker")(_new_case()) + self.assertTrue(_skipped(decorated)) + + def test_runs_when_selected_and_capable(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ + patch("tests._backend.backend_capability", + return_value=Capability("docker", True, "ok")): + decorated = skip_unless_backend("docker")(_new_case()) + self.assertFalse(_skipped(decorated)) + + def test_skips_when_selected_but_incapable(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ + patch("tests._backend.backend_capability", + return_value=Capability("docker", False, "docker daemon unreachable")): + decorated = skip_unless_backend("docker")(_new_case()) + self.assertTrue(_skipped(decorated)) + + +class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase): + def test_runs_when_selected_backend_capable(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \ + patch("tests._backend.backend_capability", + return_value=Capability("firecracker", True, "ok")): + decorated = skip_unless_selected_backend_available()(_new_case()) + self.assertFalse(_skipped(decorated)) + + def test_skips_when_selected_backend_incapable(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \ + patch("tests._backend.backend_capability", + return_value=Capability("firecracker", False, "/dev/kvm missing")): + decorated = skip_unless_selected_backend_available()(_new_case()) + self.assertTrue(_skipped(decorated)) + + +class TestPreflightMain(unittest.TestCase): + def test_pass_exits_zero(self): + with patch.object(backend_preflight, "backend_capability", + return_value=Capability("docker", True, "docker daemon reachable")): + rc = backend_preflight.main(["docker"]) + self.assertEqual(0, rc) + + def test_fail_exits_nonzero(self): + with patch.object(backend_preflight, "backend_capability", + return_value=Capability("firecracker", False, "/dev/kvm missing")): + rc = backend_preflight.main(["firecracker"]) + self.assertEqual(1, rc) + + def test_defaults_to_selected_backend(self): + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ + patch.object(backend_preflight, "backend_capability", + return_value=Capability("docker", True, "ok")) as cap: + rc = backend_preflight.main([]) + self.assertEqual(0, rc) + cap.assert_called_once_with("docker") + + def test_module_exposes_selected_backend(self): + # Import-surface guard: the preflight re-exports the selector it uses. + self.assertIs(backend_preflight.selected_backend, _backend.selected_backend) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_docker_test_helpers.py b/tests/unit/test_docker_test_helpers.py deleted file mode 100644 index a6c1e37..0000000 --- a/tests/unit/test_docker_test_helpers.py +++ /dev/null @@ -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() -- 2.52.0 From d4d45f835e06dd67322940303eb7fe1233a1bfcb Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 01:41:36 +0000 Subject: [PATCH 2/8] ci: reuse backend.has_backend / `backend status` for readiness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback: the hand-rolled `/dev/kvm` + docker-daemon capability probes duplicated logic the CLI already owns. Delegate instead. - tests/_backend.py skip guards now gate on `bot_bottle.backend.has_backend` (each backend's `is_available()` classmethod — the same probe behind `./cli.py backend status`), dropping the bespoke `Capability` probes. - Remove tests/backend_preflight.py; the docker integration job runs `./cli.py backend status --backend=docker` as its preflight (clear per-check summary, non-zero exit when unready), matching the firecracker job. The firecracker preflight reverts to its original binary/KVM checks (backend status doesn't cover those). - Unit test + docs updated to match. Co-Authored-By: Claude Opus 4.8 --- .gitea/workflows/test.yml | 18 ++- docs/ci.md | 14 +-- tests/README.md | 18 +-- tests/_backend.py | 130 ++++------------------ tests/backend_preflight.py | 27 ----- tests/unit/test_backend_skip_guards.py | 146 +++---------------------- 6 files changed, 62 insertions(+), 291 deletions(-) delete mode 100644 tests/backend_preflight.py diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index f2e1a16..1fbf962 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -105,13 +105,14 @@ jobs: # 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. Same capability probe the skip guards use. + # 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 - env: - BOT_BOTTLE_BACKEND: docker run: | python3 --version - python3 -m tests.backend_preflight docker + python3 cli.py backend status --backend=docker - name: Run integration tests (docker) with coverage env: @@ -158,13 +159,10 @@ jobs: uses: actions/checkout@v4 - name: Preflight — Firecracker host is ready - env: - BOT_BOTTLE_BACKEND: firecracker run: | - # Standardized capability line — same probe (and PASS/FAIL wording) - # the integration skip guards use, so a missing backend is visible - # at the job level rather than hidden among `unittest.skip` lines. - python3 -m tests.backend_preflight firecracker + command -v firecracker >/dev/null || { + echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; } + test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; } # `backend status` exits non-zero unless the TAP pool is up + no # range overlap; it prints the exact `backend setup` fix. python3 cli.py backend status --backend=firecracker diff --git a/docs/ci.md b/docs/ci.md index c0230c2..3cc5e74 100644 --- a/docs/ci.md +++ b/docs/ci.md @@ -8,15 +8,15 @@ It runs the unit suite plus one integration job per backend - every push to `main`. Each integration job selects its backend via `BOT_BOTTLE_BACKEND` and -runs a **preflight** (`python3 -m tests.backend_preflight `) -that prints a clear PASS/FAIL capability line and fails the job when the +runs a **preflight** (`./cli.py backend status --backend=`) 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 same -capability probes back the skip guards in -[`tests/_backend.py`](../tests/_backend.py): backend-agnostic tests use +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., `/dev/kvm` for Firecracker rather -than unrelated Docker availability); Docker-implementation tests use +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 diff --git a/tests/README.md b/tests/README.md index 25391fc..28985cd 100644 --- a/tests/README.md +++ b/tests/README.md @@ -10,8 +10,7 @@ tests run through the backend named by `BOT_BOTTLE_BACKEND` (default ``` tests/ fixtures.py # JSON manifest builders (shared) - _backend.py # backend capability probes + skip guards - backend_preflight.py # `python -m tests.backend_preflight` (CI) + _backend.py # backend selection + skip guards (shared) unit/ test_egress.py test_egress_addon_core.py @@ -90,16 +89,19 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v if __name__ == "__main__": unittest.main() ``` -4. Skip guards live in `tests._backend`: +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's capability is present (a reachable Docker daemon, - or an accessible `/dev/kvm` + `firecracker` for Firecracker). + 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. - The same capability probes back the CI preflight, - `python -m tests.backend_preflight []`, which prints a clear - PASS/FAIL line and exits non-zero when the selected backend is missing. + Each CI integration job runs `./cli.py backend status --backend=` + 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. diff --git a/tests/_backend.py b/tests/_backend.py index 39ed57c..00f5f38 100644 --- a/tests/_backend.py +++ b/tests/_backend.py @@ -1,34 +1,22 @@ -"""Backend-aware capability probes and skip guards for the integration suite. +"""Backend selection + capability-aware skip guards for the integration suite. -Every integration test selects its backend from ``BOT_BOTTLE_BACKEND`` -(default ``docker``) and exercises the full test through that backend. Skip -guards check the *capability* the selected backend actually needs — a -reachable Docker daemon for ``docker``, an accessible ``/dev/kvm`` plus a -``firecracker`` binary for ``firecracker`` — rather than gating every backend -on unrelated Docker availability. - -The same probes back the CI preflight (``python3 -m tests.backend_preflight``) -so a job surfaces missing infrastructure at the job level with a clear -PASS/FAIL line instead of turning every test into a silent ``unittest.skip``. +Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND`` +(default ``docker``) and skips on that backend's own readiness check — +``bot_bottle.backend.has_backend`` (the same probe behind +``./cli.py backend status``), not an unrelated "is Docker installed" test. """ from __future__ import annotations import os -import shutil -import subprocess import unittest -from dataclasses import dataclass + +from bot_bottle.backend import has_backend # 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" -# `/dev/kvm` must exist and be openable by the invoking user for the -# Firecracker backend to boot a guest (mirrors -# ``bot_bottle.backend.firecracker.util``). -_KVM_DEVICE = "/dev/kvm" - def selected_backend() -> str: """The backend this test run targets, from ``BOT_BOTTLE_BACKEND``. @@ -40,114 +28,34 @@ def selected_backend() -> str: return os.environ.get("BOT_BOTTLE_BACKEND") or DEFAULT_BACKEND -@dataclass(frozen=True) -class Capability: - """Outcome of a backend capability probe. - - ``ok`` gates the tests; ``detail`` is a one-line human string reused for - both preflight output and ``unittest.skip`` reasons so the same wording - shows up at the job level and next to a skipped test. - """ - - backend: str - ok: bool - detail: str - - -def docker_capability() -> Capability: - """Whether a Docker daemon is reachable (and not opted out).""" - name = "docker" - if os.environ.get("SKIP_DOCKER_TESTS"): - return Capability(name, False, "SKIP_DOCKER_TESTS is set") - if shutil.which("docker") is None: - return Capability(name, False, "docker not on PATH") - try: - returncode = subprocess.run( - ["docker", "info"], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False, - timeout=5, - ).returncode - except subprocess.TimeoutExpired: - return Capability(name, False, "docker info timed out") - if returncode != 0: - return Capability(name, False, "docker daemon unreachable") - return Capability(name, True, "docker daemon reachable") - - -def firecracker_capability() -> Capability: - """Whether this host can boot a Firecracker guest: an accessible - ``/dev/kvm`` and a ``firecracker`` binary on ``PATH``. - - Deliberately does not probe unrelated Docker availability — the - Firecracker backend is Docker-independent at runtime. - """ - name = "firecracker" - if not os.path.exists(_KVM_DEVICE): - return Capability(name, False, f"{_KVM_DEVICE} missing — KVM unavailable") - if not os.access(_KVM_DEVICE, os.R_OK | os.W_OK): - return Capability( - name, False, f"{_KVM_DEVICE} not accessible — add your user to the kvm group" - ) - if shutil.which("firecracker") is None: - return Capability(name, False, "firecracker not on PATH") - return Capability(name, True, f"{_KVM_DEVICE} accessible; firecracker on PATH") - - -_PROBES = { - "docker": docker_capability, - "firecracker": firecracker_capability, -} - - -def backend_capability(backend: str | None = None) -> Capability: - """Probe ``backend`` (default: the selected backend) for readiness.""" - backend = backend or selected_backend() - probe = _PROBES.get(backend) - if probe is None: - return Capability(backend, False, f"no capability probe for backend {backend!r}") - return probe() - - -# ---- back-compat boolean probe ------------------------------------------- - - -def docker_available() -> bool: - """Boolean Docker probe (kept for callers that only need the flag).""" - return docker_capability().ok - - -# ---- skip guards --------------------------------------------------------- - - def skip_unless_backend(backend: str): """Skip a backend-specific test unless the selected backend matches AND - that backend's capability is present. + that backend is available on the host. Docker-implementation tests (``DockerBroker``, ``DockerGateway``, ``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op - under a Firecracker run instead of testing Docker internals that run - doesn't target — the guard reads ``BOT_BOTTLE_BACKEND`` rather than - "is Docker installed". + 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". """ sel = selected_backend() if sel != backend: return unittest.skip( f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})" ) - cap = backend_capability(backend) - return unittest.skipUnless(cap.ok, f"{backend} backend unavailable: {cap.detail}") + return unittest.skipUnless( + has_backend(backend), f"{backend} backend prerequisites unavailable" + ) def skip_unless_selected_backend_available(): - """Skip a backend-agnostic test unless the *selected* backend can run it. + """Skip a backend-agnostic test unless the *selected* backend is available. - The test then exercises whichever backend ``BOT_BOTTLE_BACKEND`` names, - checking that backend's real capability (e.g. ``/dev/kvm`` for + The test then runs through whichever backend ``BOT_BOTTLE_BACKEND`` names, + gated on that backend's real prerequisites (e.g. Linux + KVM for Firecracker) rather than unrelated Docker availability. """ - cap = backend_capability() + backend = selected_backend() return unittest.skipUnless( - cap.ok, f"selected backend {cap.backend!r} unavailable: {cap.detail}" + has_backend(backend), f"selected backend {backend!r} prerequisites unavailable" ) diff --git a/tests/backend_preflight.py b/tests/backend_preflight.py deleted file mode 100644 index 0659586..0000000 --- a/tests/backend_preflight.py +++ /dev/null @@ -1,27 +0,0 @@ -"""Backend capability preflight for the integration CI jobs. - -Run as ``python3 -m tests.backend_preflight`` (optionally with a backend name -argument; default: ``BOT_BOTTLE_BACKEND`` or ``docker``). Prints a single -clear PASS/FAIL line for the selected backend's capability and exits non-zero -on failure, so a job surfaces missing infrastructure at the job level instead -of hiding it among ``unittest.skip`` lines further down the output. -""" - -from __future__ import annotations - -import sys - -from tests._backend import backend_capability, selected_backend - - -def main(argv: list[str] | None = None) -> int: - argv = sys.argv[1:] if argv is None else argv - backend = argv[0] if argv else selected_backend() - cap = backend_capability(backend) - status = "PASS" if cap.ok else "FAIL" - print(f"[preflight] backend={backend} {status}: {cap.detail}") - return 0 if cap.ok else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/unit/test_backend_skip_guards.py b/tests/unit/test_backend_skip_guards.py index 54265a1..0695b85 100644 --- a/tests/unit/test_backend_skip_guards.py +++ b/tests/unit/test_backend_skip_guards.py @@ -1,9 +1,9 @@ -"""Tests for the backend-aware capability probes and skip guards -(``tests/_backend.py``) and the CI preflight (``tests/backend_preflight.py``). +"""Tests for the backend-aware skip guards in ``tests/_backend.py``. -The probes shell out to Docker / inspect ``/dev/kvm``; here they are driven -entirely through mocks so the unit job asserts the decision logic without -needing either backend present on the runner. +The guards delegate their readiness check to +``bot_bottle.backend.has_backend`` (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 @@ -12,12 +12,7 @@ import os import unittest from unittest.mock import patch -from tests import _backend, backend_preflight from tests._backend import ( - Capability, - backend_capability, - docker_capability, - firecracker_capability, selected_backend, skip_unless_backend, skip_unless_selected_backend_available, @@ -42,147 +37,42 @@ class TestSelectedBackend(unittest.TestCase): self.assertEqual("firecracker", selected_backend()) -class TestDockerCapability(unittest.TestCase): - def test_opt_out_via_skip_docker_tests(self): - with patch.dict(os.environ, {"SKIP_DOCKER_TESTS": "1"}, clear=True): - cap = docker_capability() - self.assertFalse(cap.ok) - self.assertIn("SKIP_DOCKER_TESTS", cap.detail) - - def test_not_on_path(self): - with patch.dict(os.environ, {}, clear=True), \ - patch("tests._backend.shutil.which", return_value=None): - cap = docker_capability() - self.assertFalse(cap.ok) - self.assertIn("PATH", cap.detail) - - def test_daemon_reachable(self): - with patch.dict(os.environ, {}, clear=True), \ - patch("tests._backend.shutil.which", return_value="/usr/bin/docker"), \ - patch("tests._backend.subprocess.run") as run: - run.return_value.returncode = 0 - cap = docker_capability() - self.assertTrue(cap.ok) - - def test_daemon_unreachable(self): - with patch.dict(os.environ, {}, clear=True), \ - patch("tests._backend.shutil.which", return_value="/usr/bin/docker"), \ - patch("tests._backend.subprocess.run") as run: - run.return_value.returncode = 1 - cap = docker_capability() - self.assertFalse(cap.ok) - self.assertIn("unreachable", cap.detail) - - -class TestFirecrackerCapability(unittest.TestCase): - def test_kvm_missing(self): - with patch("tests._backend.os.path.exists", return_value=False): - cap = firecracker_capability() - self.assertFalse(cap.ok) - self.assertIn("KVM", cap.detail) - - def test_kvm_inaccessible(self): - with patch("tests._backend.os.path.exists", return_value=True), \ - patch("tests._backend.os.access", return_value=False): - cap = firecracker_capability() - self.assertFalse(cap.ok) - self.assertIn("kvm group", cap.detail) - - def test_binary_missing(self): - with patch("tests._backend.os.path.exists", return_value=True), \ - patch("tests._backend.os.access", return_value=True), \ - patch("tests._backend.shutil.which", return_value=None): - cap = firecracker_capability() - self.assertFalse(cap.ok) - self.assertIn("firecracker not on PATH", cap.detail) - - def test_ready(self): - with patch("tests._backend.os.path.exists", return_value=True), \ - patch("tests._backend.os.access", return_value=True), \ - patch("tests._backend.shutil.which", return_value="/usr/bin/firecracker"): - cap = firecracker_capability() - self.assertTrue(cap.ok) - - -class TestBackendCapability(unittest.TestCase): - def test_unknown_backend(self): - cap = backend_capability("qemu") - self.assertFalse(cap.ok) - self.assertIn("no capability probe", cap.detail) - - def test_defaults_to_selected(self): - def probe() -> Capability: - return Capability("docker", True, "ok") - - with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ - patch.dict("tests._backend._PROBES", {"docker": probe}): - cap = backend_capability() - self.assertEqual("docker", cap.backend) - self.assertTrue(cap.ok) - - class TestSkipUnlessBackend(unittest.TestCase): def test_skips_when_other_backend_selected(self): - with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True): + # A different backend is selected — no host probe needed, skip. + with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \ + patch("tests._backend.has_backend") as has: decorated = skip_unless_backend("docker")(_new_case()) self.assertTrue(_skipped(decorated)) + has.assert_not_called() - def test_runs_when_selected_and_capable(self): + def test_runs_when_selected_and_available(self): with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ - patch("tests._backend.backend_capability", - return_value=Capability("docker", True, "ok")): + patch("tests._backend.has_backend", return_value=True): decorated = skip_unless_backend("docker")(_new_case()) self.assertFalse(_skipped(decorated)) - def test_skips_when_selected_but_incapable(self): + def test_skips_when_selected_but_unavailable(self): with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ - patch("tests._backend.backend_capability", - return_value=Capability("docker", False, "docker daemon unreachable")): + patch("tests._backend.has_backend", return_value=False): decorated = skip_unless_backend("docker")(_new_case()) self.assertTrue(_skipped(decorated)) class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase): - def test_runs_when_selected_backend_capable(self): + def test_runs_when_selected_backend_available(self): with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \ - patch("tests._backend.backend_capability", - return_value=Capability("firecracker", True, "ok")): + patch("tests._backend.has_backend", return_value=True) as has: decorated = skip_unless_selected_backend_available()(_new_case()) self.assertFalse(_skipped(decorated)) + has.assert_called_once_with("firecracker") - def test_skips_when_selected_backend_incapable(self): + def test_skips_when_selected_backend_unavailable(self): with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \ - patch("tests._backend.backend_capability", - return_value=Capability("firecracker", False, "/dev/kvm missing")): + patch("tests._backend.has_backend", return_value=False): decorated = skip_unless_selected_backend_available()(_new_case()) self.assertTrue(_skipped(decorated)) -class TestPreflightMain(unittest.TestCase): - def test_pass_exits_zero(self): - with patch.object(backend_preflight, "backend_capability", - return_value=Capability("docker", True, "docker daemon reachable")): - rc = backend_preflight.main(["docker"]) - self.assertEqual(0, rc) - - def test_fail_exits_nonzero(self): - with patch.object(backend_preflight, "backend_capability", - return_value=Capability("firecracker", False, "/dev/kvm missing")): - rc = backend_preflight.main(["firecracker"]) - self.assertEqual(1, rc) - - def test_defaults_to_selected_backend(self): - with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \ - patch.object(backend_preflight, "backend_capability", - return_value=Capability("docker", True, "ok")) as cap: - rc = backend_preflight.main([]) - self.assertEqual(0, rc) - cap.assert_called_once_with("docker") - - def test_module_exposes_selected_backend(self): - # Import-surface guard: the preflight re-exports the selector it uses. - self.assertIs(backend_preflight.selected_backend, _backend.selected_backend) - - if __name__ == "__main__": unittest.main() -- 2.52.0 From 0ca39cf4d58ef091941f947633934fe01075a07a Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 03:03:00 +0000 Subject: [PATCH 3/8] feat(backend): BackendStatus enum, quiet status(), is_backend_ready() Add a module-level BackendStatus(IntEnum) with READY=0 to replace magic 0 comparisons. Extend BottleBackend.status() with a quiet parameter: quiet=False (default) prints diagnostics; quiet=True returns the code silently for programmatic checks. Add is_backend_available() (cheap PATH check, alias for has_backend) and is_backend_ready(name, *, quiet=False) (full status() check) to the backend package for callers that need to distinguish availability from readiness. Update tests/_backend.py guards to use is_backend_ready(quiet=False) so diagnostic output is printed during test discovery, giving the operator a concrete reason for each skip rather than a bare "prerequisites unavailable" message. Co-Authored-By: Claude Sonnet 4.6 --- bot_bottle/backend/__init__.py | 47 +++++++++++++++++-- bot_bottle/backend/docker/backend.py | 8 +++- bot_bottle/backend/firecracker/backend.py | 8 +++- bot_bottle/backend/macos_container/backend.py | 8 +++- tests/_backend.py | 32 ++++++++----- 5 files changed, 82 insertions(+), 21 deletions(-) diff --git a/bot_bottle/backend/__init__.py b/bot_bottle/backend/__init__.py index ff5223e..cc5be62 100644 --- a/bot_bottle/backend/__init__.py +++ b/bot_bottle/backend/__init__.py @@ -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", ] diff --git a/bot_bottle/backend/docker/backend.py b/bot_bottle/backend/docker/backend.py index b2ae45b..5e0c735 100644 --- a/bot_bottle/backend/docker/backend.py +++ b/bot_bottle/backend/docker/backend.py @@ -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 diff --git a/bot_bottle/backend/firecracker/backend.py b/bot_bottle/backend/firecracker/backend.py index a8a9df9..89a851c 100644 --- a/bot_bottle/backend/firecracker/backend.py +++ b/bot_bottle/backend/firecracker/backend.py @@ -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 diff --git a/bot_bottle/backend/macos_container/backend.py b/bot_bottle/backend/macos_container/backend.py index 2448c00..6d59994 100644 --- a/bot_bottle/backend/macos_container/backend.py +++ b/bot_bottle/backend/macos_container/backend.py @@ -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 diff --git a/tests/_backend.py b/tests/_backend.py index 00f5f38..f40ca8c 100644 --- a/tests/_backend.py +++ b/tests/_backend.py @@ -1,9 +1,11 @@ -"""Backend selection + capability-aware skip guards for the integration suite. +"""Backend selection + readiness-aware skip guards for the integration suite. Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND`` -(default ``docker``) and skips on that backend's own readiness check — -``bot_bottle.backend.has_backend`` (the same probe behind -``./cli.py backend status``), not an unrelated "is Docker installed" test. +(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 @@ -11,7 +13,7 @@ from __future__ import annotations import os import unittest -from bot_bottle.backend import has_backend +from bot_bottle.backend import is_backend_available, 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``). @@ -30,13 +32,16 @@ def selected_backend() -> str: def skip_unless_backend(backend: str): """Skip a backend-specific test unless the selected backend matches AND - that backend is available on the host. + 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: @@ -44,18 +49,23 @@ def skip_unless_backend(backend: str): f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})" ) return unittest.skipUnless( - has_backend(backend), f"{backend} backend prerequisites unavailable" + 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 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 real prerequisites (e.g. Linux + KVM for - Firecracker) rather than unrelated Docker availability. + 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( - has_backend(backend), f"selected backend {backend!r} prerequisites unavailable" + is_backend_ready(backend, quiet=False), + f"selected backend {backend!r} not ready", ) -- 2.52.0 From 1f9a15dedef014692bdd474e3d853113ec8c1b69 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 04:25:04 +0000 Subject: [PATCH 4/8] fix: align skip-guard tests with is_backend_ready and drop unused import MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests/_backend.py imported is_backend_available but never called it, causing a pyright reportUnusedImport error. Remove the unused import. test_backend_skip_guards.py was patching tests._backend.has_backend but _backend.py calls is_backend_ready — the mock never intercepted the real call, producing AttributeError at test runtime. Update all patch targets to is_backend_ready and add quiet=False to the assert_called_once_with assertion to match the actual call signature. Co-Authored-By: Claude Sonnet 4.6 --- tests/_backend.py | 2 +- tests/unit/test_backend_skip_guards.py | 14 +++++++------- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/tests/_backend.py b/tests/_backend.py index f40ca8c..54f27f8 100644 --- a/tests/_backend.py +++ b/tests/_backend.py @@ -13,7 +13,7 @@ from __future__ import annotations import os import unittest -from bot_bottle.backend import is_backend_available, is_backend_ready +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``). diff --git a/tests/unit/test_backend_skip_guards.py b/tests/unit/test_backend_skip_guards.py index 0695b85..ecbc4a4 100644 --- a/tests/unit/test_backend_skip_guards.py +++ b/tests/unit/test_backend_skip_guards.py @@ -1,7 +1,7 @@ """Tests for the backend-aware skip guards in ``tests/_backend.py``. The guards delegate their readiness check to -``bot_bottle.backend.has_backend`` (the probe behind ``./cli.py backend +``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. """ @@ -41,20 +41,20 @@ 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.has_backend") as has: + 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.has_backend", return_value=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.has_backend", return_value=False): + patch("tests._backend.is_backend_ready", return_value=False): decorated = skip_unless_backend("docker")(_new_case()) self.assertTrue(_skipped(decorated)) @@ -62,14 +62,14 @@ class TestSkipUnlessBackend(unittest.TestCase): 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.has_backend", return_value=True) as has: + 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") + 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.has_backend", return_value=False): + patch("tests._backend.is_backend_ready", return_value=False): decorated = skip_unless_selected_backend_available()(_new_case()) self.assertTrue(_skipped(decorated)) -- 2.52.0 From 05b62ce80560f07ab622db7b16e552643220bd61 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 06:12:02 +0000 Subject: [PATCH 5/8] test: cover is_backend_available, is_backend_ready, and status(quiet=True) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add unit tests for the three new public symbols introduced by this PR: - TestIsBackendAvailable — delegates to has_backend; returns True/False correctly for available and unavailable backends - TestIsBackendReady — unknown names return False; known names return True/False based on status() return code; quiet flag is forwarded - TestBackendStatusQuiet — verifies the quiet=True branch in DockerBottleBackend, FirecrackerBottleBackend, and MacosContainerBottleBackend: return code is propagated and diagnostic stderr is suppressed Together these bring diff-coverage for the PR's changed lines to 100%. Co-Authored-By: Claude Sonnet 4.6 --- tests/unit/test_backend_selection.py | 54 ++++++++++++++++++++++++++++ tests/unit/test_backend_setup.py | 39 ++++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/tests/unit/test_backend_selection.py b/tests/unit/test_backend_selection.py index a5a4c18..05b6581 100644 --- a/tests/unit/test_backend_selection.py +++ b/tests/unit/test_backend_selection.py @@ -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=False): + 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=False): + 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=False): + 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; diff --git a/tests/unit/test_backend_setup.py b/tests/unit/test_backend_setup.py index a2871c2..12c3ca9 100644 --- a/tests/unit/test_backend_setup.py +++ b/tests/unit/test_backend_setup.py @@ -328,5 +328,44 @@ class TestNetpoolShellRenderers(unittest.TestCase): self.assertIn("BOT_BOTTLE_FC_POOL_SIZE=4", 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 + written = [] + real_status = dk.status + + 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() -- 2.52.0 From e074c6959f110de875db4c9a938850ca3b24ad61 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 06:18:51 +0000 Subject: [PATCH 6/8] fix: add missing type annotations and remove unused variables in new tests Pyright flagged three inner class `status` methods missing a type annotation on the `quiet` parameter, and two unused variables (`written`, `real_status`) left over from an earlier draft of test_docker_quiet_true_suppresses_stderr. Co-Authored-By: Claude Sonnet 4.6 --- tests/unit/test_backend_selection.py | 6 +++--- tests/unit/test_backend_setup.py | 2 -- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/tests/unit/test_backend_selection.py b/tests/unit/test_backend_selection.py index 05b6581..3d3b85d 100644 --- a/tests/unit/test_backend_selection.py +++ b/tests/unit/test_backend_selection.py @@ -409,7 +409,7 @@ class TestIsBackendReady(unittest.TestCase): def test_ready_when_status_returns_zero(self): class _ReadyBackend: - def status(self, *, quiet=False): + def status(self, *, quiet: bool = False) -> int: return 0 from bot_bottle.backend import is_backend_ready @@ -418,7 +418,7 @@ class TestIsBackendReady(unittest.TestCase): def test_not_ready_when_status_nonzero(self): class _BrokenBackend: - def status(self, *, quiet=False): + def status(self, *, quiet: bool = False) -> int: return 1 from bot_bottle.backend import is_backend_ready @@ -429,7 +429,7 @@ class TestIsBackendReady(unittest.TestCase): calls = [] class _SpyBackend: - def status(self, *, quiet=False): + def status(self, *, quiet: bool = False) -> int: calls.append(quiet) return 0 diff --git a/tests/unit/test_backend_setup.py b/tests/unit/test_backend_setup.py index 12c3ca9..4b616ee 100644 --- a/tests/unit/test_backend_setup.py +++ b/tests/unit/test_backend_setup.py @@ -342,8 +342,6 @@ class TestBackendStatusQuiet(unittest.TestCase): def test_docker_quiet_true_suppresses_stderr(self): import io as _io from bot_bottle.backend.docker.backend import DockerBottleBackend - written = [] - real_status = dk.status def _loud_status(): import sys -- 2.52.0 From 755a11a608d4a0b08bff9ce0f5cb8b8931ee5506 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 06:51:58 +0000 Subject: [PATCH 7/8] firecracker status(): add binary and KVM readiness checks Adds `_firecracker_binary_ok()` (runs `firecracker --version`) and `_kvm_accessible()` (opens /dev/kvm, issues KVM_GET_API_VERSION ioctl) and gates `status()` on both, so the launch preflight reports missing binary or inaccessible KVM before the caller ever attempts a boot. Regression tests in TestFirecrackerBinaryCheck, TestFirecrackerKvmCheck, and TestFirecrackerStatusRuntime cover all branches. Updated the pre-existing TestFirecrackerStatus fixture to also stub the two new helpers so it still exercises only the tap-pool/nft logic it was designed for. Co-Authored-By: Claude Sonnet 4.6 --- bot_bottle/backend/firecracker/setup.py | 66 ++++++++++++++-- tests/unit/test_backend_setup.py | 101 ++++++++++++++++++++++++ tests/unit/test_firecracker_backend.py | 4 +- 3 files changed, 164 insertions(+), 7 deletions(-) diff --git a/bot_bottle/backend/firecracker/setup.py b/bot_bottle/backend/firecracker/setup.py index 90a0427..0d5d00d 100644 --- a/bot_bottle/backend/firecracker/setup.py +++ b/bot_bottle/backend/firecracker/setup.py @@ -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,64 @@ 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 and responds to KVM_GET_API_VERSION. + + Uses an ioctl rather than os.access() so that a device with correct + permission bits but a non-functional KVM subsystem is caught here + rather than at VM boot time.""" + if not os.path.exists(util._KVM_DEVICE): + return False + try: + with open(util._KVM_DEVICE, "rb") as kvm: + fcntl.ioctl(kvm, _KVM_GET_API_VERSION) + 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 missing = netpool.missing_taps() total = netpool.pool_size() if missing: diff --git a/tests/unit/test_backend_setup.py b/tests/unit/test_backend_setup.py index 4b616ee..2b9b173 100644 --- a/tests/unit/test_backend_setup.py +++ b/tests/unit/test_backend_setup.py @@ -328,6 +328,107 @@ 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): + m = MagicMock() + m.__enter__ = MagicMock(return_value=m) + m.__exit__ = MagicMock(return_value=False) + with patch.object(fc.os.path, "exists", return_value=True), \ + patch("builtins.open", return_value=m), \ + patch.object(fc.fcntl, "ioctl", return_value=12): + self.assertTrue(fc._kvm_accessible()) + + def test_kvm_present_but_ioctl_fails(self): + m = MagicMock() + m.__enter__ = MagicMock(return_value=m) + m.__exit__ = MagicMock(return_value=False) + with patch.object(fc.os.path, "exists", return_value=True), \ + patch("builtins.open", return_value=m), \ + patch.object(fc.fcntl, "ioctl", side_effect=OSError("permission denied")): + self.assertFalse(fc._kvm_accessible()) + + +class TestFirecrackerStatusRuntime(unittest.TestCase): + """status() reports binary and KVM problems and returns non-zero.""" + + def _apply_pool_ok(self, stack: contextlib.ExitStack) -> None: + 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 diff --git a/tests/unit/test_firecracker_backend.py b/tests/unit/test_firecracker_backend.py index 3a1387b..7d4ccd9 100644 --- a/tests/unit/test_firecracker_backend.py +++ b/tests/unit/test_firecracker_backend.py @@ -136,7 +136,9 @@ class TestFirecrackerStatus(unittest.TestCase): from bot_bottle.backend.firecracker import setup as 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", return_value=None), \ + patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \ + patch.object(fc_setup, "_kvm_accessible", return_value=True): rc, out = self._run() self.assertEqual(0, rc) self.assertIn("unverified", out) -- 2.52.0 From 65a49a239effc9cbbb741b568905e079c0610a4c Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 16:03:51 +0000 Subject: [PATCH 8/8] firecracker status(): open /dev/kvm O_RDWR; check kernel, dropbear, mke2fs Two reviewer findings addressed: 1. _kvm_accessible() now opens /dev/kvm with O_RDWR|O_CLOEXEC instead of read-only. VM creation requires write access; a read-only descriptor can satisfy KVM_GET_API_VERSION but fails at boot time. 2. status() now checks every hard prerequisite that require_firecracker() checks at launch: guest kernel image, static dropbear binary, and mke2fs. Previously a host with a configured TAP pool but missing artifacts could pass status() and then fail during launch. Regression coverage: TestFirecrackerKvmCheck gains test_kvm_open_rdwr_fails; TestFirecrackerArtifactCheck covers each missing artifact; existing TestFirecrackerStatus fixtures updated to stub the new checks. Co-Authored-By: Claude Sonnet 4.6 --- bot_bottle/backend/firecracker/setup.py | 38 +++++++++++-- tests/unit/test_backend_setup.py | 76 ++++++++++++++++++++++--- tests/unit/test_firecracker_backend.py | 32 +++++++++-- 3 files changed, 127 insertions(+), 19 deletions(-) diff --git a/bot_bottle/backend/firecracker/setup.py b/bot_bottle/backend/firecracker/setup.py index 0d5d00d..d63dbb5 100644 --- a/bot_bottle/backend/firecracker/setup.py +++ b/bot_bottle/backend/firecracker/setup.py @@ -237,16 +237,18 @@ def _firecracker_binary_ok() -> bool: def _kvm_accessible() -> bool: - """True iff /dev/kvm can be opened and responds to KVM_GET_API_VERSION. + """True iff /dev/kvm can be opened read-write and responds to KVM_GET_API_VERSION. - Uses an ioctl rather than os.access() so that a device with correct - permission bits but a non-functional KVM subsystem is caught here - rather than at VM boot time.""" + 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: - with open(util._KVM_DEVICE, "rb") as kvm: - fcntl.ioctl(kvm, _KVM_GET_API_VERSION) + 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 @@ -281,6 +283,30 @@ def status() -> int: 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: diff --git a/tests/unit/test_backend_setup.py b/tests/unit/test_backend_setup.py index 2b9b173..f738032 100644 --- a/tests/unit/test_backend_setup.py +++ b/tests/unit/test_backend_setup.py @@ -357,28 +357,86 @@ class TestFirecrackerKvmCheck(unittest.TestCase): self.assertFalse(fc._kvm_accessible()) def test_kvm_accessible_when_ioctl_succeeds(self): - m = MagicMock() - m.__enter__ = MagicMock(return_value=m) - m.__exit__ = MagicMock(return_value=False) with patch.object(fc.os.path, "exists", return_value=True), \ - patch("builtins.open", return_value=m), \ + 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_present_but_ioctl_fails(self): - m = MagicMock() - m.__enter__ = MagicMock(return_value=m) - m.__exit__ = MagicMock(return_value=False) + def test_kvm_open_rdwr_fails(self): with patch.object(fc.os.path, "exists", return_value=True), \ - patch("builtins.open", return_value=m), \ + 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=[])) diff --git a/tests/unit/test_firecracker_backend.py b/tests/unit/test_firecracker_backend.py index 7d4ccd9..6ec358e 100644 --- a/tests/unit/test_firecracker_backend.py +++ b/tests/unit/test_firecracker_backend.py @@ -132,32 +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, "_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) -- 2.52.0