Compare commits

...

7 Commits

Author SHA1 Message Date
didericis-claude 755a11a608 firecracker status(): add binary and KVM readiness checks
tracker-policy-pr / check-pr (pull_request) Successful in 11s
test / integration-docker (pull_request) Successful in 19s
lint / lint (push) Successful in 1m0s
test / unit (pull_request) Successful in 1m43s
test / integration-firecracker (pull_request) Successful in 3m22s
test / coverage (pull_request) Successful in 16s
test / publish-infra (pull_request) Has been skipped
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 <noreply@anthropic.com>
2026-07-24 06:51:58 +00:00
didericis-claude e074c6959f fix: add missing type annotations and remove unused variables in new tests
test / integration-docker (pull_request) Successful in 23s
test / unit (pull_request) Successful in 47s
lint / lint (push) Successful in 3m3s
test / integration-firecracker (pull_request) Successful in 3m24s
test / coverage (pull_request) Successful in 20s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 17s
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 <noreply@anthropic.com>
2026-07-24 06:18:51 +00:00
didericis-claude 05b62ce805 test: cover is_backend_available, is_backend_ready, and status(quiet=True)
test / integration-docker (pull_request) Successful in 21s
lint / lint (push) Failing after 59s
test / unit (pull_request) Successful in 1m47s
test / integration-firecracker (pull_request) Successful in 3m22s
test / coverage (pull_request) Successful in 18s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Failing after 13m0s
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 <noreply@anthropic.com>
2026-07-24 06:12:02 +00:00
didericis-claude 1f9a15dede fix: align skip-guard tests with is_backend_ready and drop unused import
tracker-policy-pr / check-pr (pull_request) Successful in 14s
test / integration-docker (pull_request) Successful in 16s
test / unit (pull_request) Successful in 37s
lint / lint (push) Successful in 53s
test / integration-firecracker (pull_request) Successful in 3m18s
test / coverage (pull_request) Failing after 35s
test / publish-infra (pull_request) Has been skipped
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 <noreply@anthropic.com>
2026-07-24 04:25:04 +00:00
didericis-claude 0ca39cf4d5 feat(backend): BackendStatus enum, quiet status(), is_backend_ready()
tracker-policy-pr / check-pr (pull_request) Successful in 11s
test / integration-docker (pull_request) Successful in 17s
lint / lint (push) Failing after 54s
test / unit (pull_request) Failing after 1m33s
test / integration-firecracker (pull_request) Successful in 3m17s
test / coverage (pull_request) Has been skipped
test / publish-infra (pull_request) Has been skipped
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 <noreply@anthropic.com>
2026-07-24 03:03:00 +00:00
didericis-claude d4d45f835e ci: reuse backend.has_backend / backend status for readiness
test / integration-docker (pull_request) Successful in 25s
test / unit (pull_request) Successful in 52s
lint / lint (push) Successful in 1m0s
test / integration-firecracker (pull_request) Successful in 2m5s
test / coverage (pull_request) Successful in 24s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 6s
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 <noreply@anthropic.com>
2026-07-24 01:42:04 +00:00
didericis-claude 9fdaba4bd4 ci: backend-agnostic integration guards + per-backend preflight
tracker-policy-pr / check-pr (pull_request) Successful in 15s
test / integration-docker (pull_request) Successful in 20s
lint / lint (push) Successful in 1m5s
test / unit (pull_request) Successful in 1m49s
test / integration-firecracker (pull_request) Successful in 2m0s
test / coverage (pull_request) Successful in 16s
test / publish-infra (pull_request) Has been skipped
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 <noreply@anthropic.com>
2026-07-23 21:33:14 -04:00
23 changed files with 522 additions and 127 deletions
+8 -6
View File
@@ -103,14 +103,16 @@ jobs:
- name: Install coverage - name: Install coverage
run: python3 -m pip install --break-system-packages coverage run: python3 -m pip install --break-system-packages coverage
- name: Show environment # Fail loudly if the backend this job promises isn't actually usable,
# rather than letting every test silently `unittest.skip` and the job
# go green on zero coverage. `backend status` prints a clear per-check
# summary (docker on PATH, daemon reachable) and exits non-zero when a
# prerequisite is missing — the same readiness check the skip guards
# gate on via `has_backend`.
- name: Preflight — Docker backend is ready
run: | run: |
python3 --version python3 --version
if command -v docker >/dev/null 2>&1; then python3 cli.py backend status --backend=docker
docker version || true
else
echo "docker not on PATH — integration tests will skip"
fi
- name: Run integration tests (docker) with coverage - name: Run integration tests (docker) with coverage
env: env:
+43 -4
View File
@@ -33,6 +33,7 @@ backend field; the host picks.
from __future__ import annotations from __future__ import annotations
import enum
import os import os
import shlex import shlex
import sys import sys
@@ -59,6 +60,12 @@ if TYPE_CHECKING:
from .freeze import CommitCancelled, Freezer, get_freezer 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) @dataclass(frozen=True)
class BottleSpec: class BottleSpec:
"""CLI-supplied intent. Backend-agnostic — each backend's prepare """CLI-supplied intent. Backend-agnostic — each backend's prepare
@@ -611,12 +618,18 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
@classmethod @classmethod
@abstractmethod @abstractmethod
def status(cls) -> int: def status(cls, *, quiet: bool = False) -> int:
"""Report whether this backend's prerequisites are satisfied on """Report whether this backend's prerequisites are satisfied on
the host — binaries, daemon reachability, network pool, range the host — binaries, daemon reachability, network pool, range
conflicts, etc. Prints a human-readable summary; returns 0 when conflicts, etc. Returns BackendStatus.READY (0) when the backend
the backend is ready to launch and non-zero when something is is ready to launch and non-zero when something is missing.
missing. Invoked by `./cli.py backend status [--backend=…]`."""
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 @classmethod
@abstractmethod @abstractmethod
@@ -811,6 +824,29 @@ def has_backend(name: str) -> bool:
return backends[name].is_available() 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]: def enumerate_active_agents() -> list[ActiveAgent]:
"""All currently-running agents, across every available """All currently-running agents, across every available
backend. Used by CLI `list active` and the dashboard's agents backend. Used by CLI `list active` and the dashboard's agents
@@ -835,6 +871,7 @@ def enumerate_active_agents() -> list[ActiveAgent]:
__all__ = [ __all__ = [
"ActiveAgent", "ActiveAgent",
"BackendStatus",
"Bottle", "Bottle",
"BottleBackend", "BottleBackend",
"BottleCleanupPlan", "BottleCleanupPlan",
@@ -847,5 +884,7 @@ __all__ = [
"get_bottle_backend", "get_bottle_backend",
"get_freezer", "get_freezer",
"has_backend", "has_backend",
"is_backend_available",
"is_backend_ready",
"known_backend_names", "known_backend_names",
] ]
+6 -2
View File
@@ -20,7 +20,8 @@ infrastructure: CA install and git copy-in.
from __future__ import annotations from __future__ import annotations
import shutil import shutil
from contextlib import contextmanager import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path from pathlib import Path
from typing import Generator, Sequence from typing import Generator, Sequence
@@ -60,8 +61,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
return _setup.setup() return _setup.setup()
@classmethod @classmethod
def status(cls) -> int: def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status() return _setup.status()
@classmethod @classmethod
+6 -2
View File
@@ -8,7 +8,8 @@ fail-closed nftables egress boundary. Selected by
from __future__ import annotations from __future__ import annotations
from contextlib import contextmanager import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path from pathlib import Path
from typing import Generator, Sequence from typing import Generator, Sequence
@@ -52,8 +53,11 @@ class FirecrackerBottleBackend(
return _setup.setup() return _setup.setup()
@classmethod @classmethod
def status(cls) -> int: def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status() return _setup.status()
@classmethod @classmethod
+60 -6
View File
@@ -13,6 +13,7 @@ generic `./cli.py backend {setup,status}` command dispatches to.
from __future__ import annotations from __future__ import annotations
import fcntl
import os import os
import shutil import shutil
import subprocess import subprocess
@@ -22,6 +23,9 @@ from pathlib import Path
from . import netpool from . import netpool
from . import util 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" _FC_RELEASES = "https://github.com/firecracker-microvm/firecracker/releases"
_UNIT_PATH = Path("/etc/systemd/system") / netpool.SYSTEMD_UNIT _UNIT_PATH = Path("/etc/systemd/system") / netpool.SYSTEMD_UNIT
@@ -219,14 +223,64 @@ def teardown() -> int:
return 0 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: def status() -> int:
# Readiness == what the launch preflight hard-requires: the TAP pool # Readiness == what the launch preflight hard-requires: the binary
# present (unprivileged, authoritative) and no range overlap. Listing # executable, /dev/kvm accessible, the TAP pool present, and no range
# the nft table usually needs root, so — like the preflight — an # overlap. Listing the nft table usually needs root, so — like the
# unconfirmable table is reported but NOT treated as not-ready; the # preflight — an unconfirmable table is reported but NOT treated as
# post-boot isolation probe is the authoritative check. This keeps an # not-ready; the post-boot isolation probe is the authoritative check.
# unprivileged `backend status` usable as a launch gate. # This keeps an unprivileged `backend status` usable as a launch gate.
ok = True 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() missing = netpool.missing_taps()
total = netpool.pool_size() total = netpool.pool_size()
if missing: if missing:
@@ -2,7 +2,8 @@
from __future__ import annotations from __future__ import annotations
from contextlib import contextmanager import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path from pathlib import Path
from typing import Generator, Sequence from typing import Generator, Sequence
@@ -43,8 +44,11 @@ class MacosContainerBottleBackend(
return _setup.setup() return _setup.setup()
@classmethod @classmethod
def status(cls) -> int: def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status() return _setup.status()
@classmethod @classmethod
+13 -3
View File
@@ -1,13 +1,23 @@
# CI # CI
The test workflow lives at [`.gitea/workflows/test.yml`](../.gitea/workflows/test.yml). 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 a branch with an open pull request, and
- every push to `main`. - every push to `main`.
Integration tests need Docker on the runner; they skip cleanly via Each integration job selects its backend via `BOT_BOTTLE_BACKEND` and
`tests/_docker.skip_unless_docker` when no daemon is reachable. runs a **preflight** (`./cli.py backend status --backend=<name>`) that
prints a clear per-check readiness summary and fails the job when the
backend is missing — so absent infrastructure is visible at the job level
rather than hidden among per-test `unittest.skip` lines. The skip guards in
[`tests/_backend.py`](../tests/_backend.py) gate on the same readiness
check (`bot_bottle.backend.has_backend`): backend-agnostic tests use
`skip_unless_selected_backend_available()` and run through whichever
backend is selected (checking, e.g., Linux + `/dev/kvm` for Firecracker
rather than unrelated Docker availability); Docker-implementation tests use
`skip_unless_backend("docker")` and no-op under a non-Docker run.
A small subset of integration tests skip when running specifically A small subset of integration tests skip when running specifically
under Gitea Actions (`GITEA_ACTIONS=true`), because `act_runner` runs under Gitea Actions (`GITEA_ACTIONS=true`), because `act_runner` runs
+20 -5
View File
@@ -2,14 +2,15 @@
Plain-Python test suite using stdlib `unittest`. No external Plain-Python test suite using stdlib `unittest`. No external
dependencies. Unit tests run anywhere Python 3 is present; integration 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 ## Layout
``` ```
tests/ tests/
fixtures.py # JSON manifest builders (shared) fixtures.py # JSON manifest builders (shared)
_docker.py # docker-availability skip helper (shared) _backend.py # backend selection + skip guards (shared)
unit/ unit/
test_egress.py test_egress.py
test_egress_addon_core.py test_egress_addon_core.py
@@ -73,7 +74,7 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
## Adding a test ## Adding a test
1. Pick the directory: `tests/unit/` for a pure unit test, 1. Pick the directory: `tests/unit/` for a pure unit test,
`tests/integration/` for one that needs Docker. `tests/integration/` for one that needs a backend.
2. Filename: `test_<topic>.py`. 2. Filename: `test_<topic>.py`.
3. Boilerplate: 3. Boilerplate:
```python ```python
@@ -88,5 +89,19 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
``` ```
4. For Docker-dependent tests, decorate the class with 4. Skip guards live in `tests._backend` and gate on the backend's own
`@skip_unless_docker()` from `tests._docker`. readiness check, `bot_bottle.backend.has_backend` — the same probe
behind `./cli.py backend status`:
- Backend-agnostic tests (go through `get_bottle_backend()`) decorate
the class with `@skip_unless_selected_backend_available()` — the test
runs against whichever backend `BOT_BOTTLE_BACKEND` selects and skips
unless that backend is available (checking, e.g., Linux + `/dev/kvm`
for Firecracker rather than unrelated Docker availability).
- Backend-specific tests (exercise `DockerBroker`, `DockerGateway`,
`backend.docker.*`, …) decorate with `@skip_unless_backend("docker")`
so they no-op under a run targeting a different backend.
Each CI integration job runs `./cli.py backend status --backend=<name>`
as a preflight, which prints a clear per-check summary and exits non-zero
when the backend is missing — so absent infrastructure fails the job
instead of hiding among per-test `unittest.skip` lines.
+71
View File
@@ -0,0 +1,71 @@
"""Backend selection + readiness-aware skip guards for the integration suite.
Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND``
(default ``docker``) and gates on that backend's full readiness check —
``is_backend_ready()`` (equivalent to ``./cli.py backend status``), not just
a binary-on-PATH probe. When the backend is not ready, diagnostic output is
printed during test discovery so the operator sees a concrete reason for each
skip.
"""
from __future__ import annotations
import os
import unittest
from bot_bottle.backend import is_backend_ready
# Default when ``BOT_BOTTLE_BACKEND`` is unset. Docker preserves the historical
# Docker-backed CI path (and mirrors the pin in ``test_sandbox_escape``).
DEFAULT_BACKEND = "docker"
def selected_backend() -> str:
"""The backend this test run targets, from ``BOT_BOTTLE_BACKEND``.
Mirrors the CLI's env selector; unset means ``docker`` so an
unconfigured run behaves exactly as the suite did before backends were
pluggable.
"""
return os.environ.get("BOT_BOTTLE_BACKEND") or DEFAULT_BACKEND
def skip_unless_backend(backend: str):
"""Skip a backend-specific test unless the selected backend matches AND
that backend is fully ready.
Docker-implementation tests (``DockerBroker``, ``DockerGateway``,
``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op
under a run targeting a different backend instead of testing Docker
internals that run doesn't exercise — the guard reads
``BOT_BOTTLE_BACKEND`` rather than "is Docker installed".
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
sel = selected_backend()
if sel != backend:
return unittest.skip(
f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})"
)
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"{backend} backend not ready",
)
def skip_unless_selected_backend_available():
"""Skip a backend-agnostic test unless the *selected* backend is fully ready.
The test then runs through whichever backend ``BOT_BOTTLE_BACKEND`` names,
gated on that backend's full status() check (e.g. daemon reachable, TAP
pool present for Firecracker) rather than just a binary-on-PATH probe.
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
backend = selected_backend()
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"selected backend {backend!r} not ready",
)
-45
View File
@@ -1,45 +0,0 @@
"""Docker availability check used by integration tests."""
from __future__ import annotations
import os
import shutil
import subprocess
import unittest
def docker_available() -> bool:
if os.environ.get("SKIP_DOCKER_TESTS"):
return False
if shutil.which("docker") is None:
return False
try:
return (
subprocess.run(
["docker", "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=5,
).returncode
== 0
)
except subprocess.TimeoutExpired:
return False
def skip_unless_docker(reason: str = "docker unreachable"):
return unittest.skipUnless(docker_available(), reason)
def skip_unless_docker_or_firecracker(
reason: str = "neither Docker nor Firecracker selected",
):
"""Skip a backend-agnostic test unless one supported backend can run.
Firecracker does not require the host Docker daemon. The KVM coverage job
deliberately sets ``SKIP_DOCKER_TESTS`` to exclude Docker-only integration
classes while still exercising this path.
"""
firecracker_selected = os.environ.get("BOT_BOTTLE_BACKEND") == "firecracker"
return unittest.skipUnless(firecracker_selected or docker_available(), reason)
+2 -2
View File
@@ -25,14 +25,14 @@ import os
import subprocess import subprocess
import unittest import unittest
from tests._docker import skip_unless_docker from tests._backend import skip_unless_backend
_IMAGE = "bot-bottle-gateway-test:chunk1" _IMAGE = "bot-bottle-gateway-test:chunk1"
_DOCKERFILE = "Dockerfile.gateway" _DOCKERFILE = "Dockerfile.gateway"
@skip_unless_docker() @skip_unless_backend("docker")
@unittest.skipIf( @unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true", os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: multi-stage build pulls a 200+MB " "skipped under act_runner: multi-stage build pulls a 200+MB "
@@ -31,7 +31,7 @@ from bot_bottle.backend.docker.gateway_net import next_free_ip
from bot_bottle.orchestrator.client import OrchestratorClient from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK
from bot_bottle.orchestrator.lifecycle import OrchestratorService 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 # One upstream reachable under two names; reflects the Authorization header so
# the probe can see exactly what the gateway injected (or didn't). # 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( @unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true", os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo " "skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -13,12 +13,12 @@ import unittest
from bot_bottle.orchestrator.broker import LaunchRequest, sign_request from bot_bottle.orchestrator.broker import LaunchRequest, sign_request
from bot_bottle.orchestrator.docker_broker import DockerBroker, container_name 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" IMAGE = "busybox"
@skip_unless_docker() @skip_unless_backend("docker")
class TestDockerBrokerIntegration(unittest.TestCase): class TestDockerBrokerIntegration(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.secret = secrets.token_bytes(16) self.secret = secrets.token_bytes(16)
@@ -28,7 +28,7 @@ from pathlib import Path
from bot_bottle.orchestrator.client import OrchestratorClient from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.lifecycle import OrchestratorService from bot_bottle.orchestrator.lifecycle import OrchestratorService
from bot_bottle.paths import host_control_plane_token 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 # Fixed (not per-run-suffixed) so repeated runs reuse the same layer-cached
# image instead of leaking a new dangling tag on every invocation. # 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" _TEST_INFRA_IMAGE = "bot-bottle-infra:itest"
@skip_unless_docker() @skip_unless_backend("docker")
@unittest.skipIf( @unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true", os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo " "skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -11,12 +11,12 @@ import subprocess
import unittest import unittest
from bot_bottle.orchestrator.gateway import DockerGateway from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker from tests._backend import skip_unless_backend
IMAGE = "busybox" IMAGE = "busybox"
@skip_unless_docker() @skip_unless_backend("docker")
class TestDockerGatewayIntegration(unittest.TestCase): class TestDockerGatewayIntegration(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.name = "bot-bottle-orch-gateway-itest-" + secrets.token_hex(4) self.name = "bot-bottle-orch-gateway-itest-" + secrets.token_hex(4)
@@ -12,12 +12,12 @@ import subprocess
import unittest import unittest
from bot_bottle.orchestrator.gateway import DockerGateway from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker from tests._backend import skip_unless_backend
IMAGE = "busybox" IMAGE = "busybox"
@skip_unless_docker() @skip_unless_backend("docker")
class TestDockerGatewayImageExists(unittest.TestCase): class TestDockerGatewayImageExists(unittest.TestCase):
def test_image_exists_true_for_present_false_for_absent(self) -> None: def test_image_exists_true_for_present_false_for_absent(self) -> None:
# Ensure the tiny image is present (build_if_missing is disabled here # Ensure the tiny image is present (build_if_missing is disabled here
+2 -2
View File
@@ -18,10 +18,10 @@ from bot_bottle.backend.docker.network import (
network_create_internal, network_create_internal,
network_remove, 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): class TestOrphanCleanup(unittest.TestCase):
def setUp(self): def setUp(self):
self.slug = f"cb-test-orphan-{os.getpid()}" self.slug = f"cb-test-orphan-{os.getpid()}"
+2 -2
View File
@@ -31,7 +31,7 @@ from pathlib import Path
from bot_bottle.backend import BottleSpec, get_bottle_backend from bot_bottle.backend import BottleSpec, get_bottle_backend
from bot_bottle.bottle_state import cleanup_state from bot_bottle.bottle_state import cleanup_state
from bot_bottle.manifest import ManifestIndex 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 # 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( @unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true" os.environ.get("GITEA_ACTIONS") == "true"
and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker", and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
+54
View File
@@ -385,6 +385,60 @@ class TestHasBackend(unittest.TestCase):
self.assertFalse(has_backend("nonexistent")) self.assertFalse(has_backend("nonexistent"))
class TestIsBackendAvailable(unittest.TestCase):
def test_delegates_to_has_backend(self):
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_available("docker"))
def test_known_and_available(self):
class _Ready:
def is_available(self):
return True
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {"docker": _Ready()}):
self.assertTrue(is_backend_available("docker"))
class TestIsBackendReady(unittest.TestCase):
def test_unknown_backend_returns_false(self):
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_ready("docker"))
def test_ready_when_status_returns_zero(self):
class _ReadyBackend:
def status(self, *, quiet: bool = False) -> int:
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _ReadyBackend()}):
self.assertTrue(is_backend_ready("docker"))
def test_not_ready_when_status_nonzero(self):
class _BrokenBackend:
def status(self, *, quiet: bool = False) -> int:
return 1
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _BrokenBackend()}):
self.assertFalse(is_backend_ready("docker"))
def test_quiet_flag_forwarded(self):
calls = []
class _SpyBackend:
def status(self, *, quiet: bool = False) -> int:
calls.append(quiet)
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _SpyBackend()}):
is_backend_ready("docker", quiet=True)
self.assertEqual([True], calls)
class TestEnsureOrchestrator(unittest.TestCase): class TestEnsureOrchestrator(unittest.TestCase):
"""The backend-agnostic orchestrator bring-up entry point. Docker starts """The backend-agnostic orchestrator bring-up entry point. Docker starts
the orchestrator + gateway containers; firecracker boots the infra VM; the orchestrator + gateway containers; firecracker boots the infra VM;
+138
View File
@@ -328,5 +328,143 @@ class TestNetpoolShellRenderers(unittest.TestCase):
self.assertIn("BOT_BOTTLE_FC_POOL_SIZE=4", out) 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
code is propagated and that calling with quiet=False leaves the non-quiet
path active (covered by the other TestDockerSetupStatus tests)."""
def test_docker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.docker.backend import DockerBottleBackend
with patch.object(dk, "status", return_value=0):
self.assertEqual(0, DockerBottleBackend.status(quiet=True))
def test_docker_quiet_true_suppresses_stderr(self):
import io as _io
from bot_bottle.backend.docker.backend import DockerBottleBackend
def _loud_status():
import sys
print("should be suppressed", file=sys.stderr)
return 0
with patch.object(dk, "status", side_effect=_loud_status):
buf = _io.StringIO()
with contextlib.redirect_stderr(buf):
DockerBottleBackend.status(quiet=True)
self.assertEqual("", buf.getvalue())
def test_firecracker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.firecracker.backend import FirecrackerBottleBackend
with patch.object(fc, "status", return_value=1):
self.assertEqual(1, FirecrackerBottleBackend.status(quiet=True))
def test_macos_quiet_true_propagates_return_code(self):
from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
with patch.object(mc, "status", return_value=0):
self.assertEqual(0, MacosContainerBottleBackend.status(quiet=True))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the backend-aware skip guards in ``tests/_backend.py``.
The guards delegate their readiness check to
``bot_bottle.backend.is_backend_ready`` (the probe behind ``./cli.py backend
status``); here that probe is mocked so the unit job asserts the
selection/skip logic without either backend present on the runner.
"""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._backend import (
selected_backend,
skip_unless_backend,
skip_unless_selected_backend_available,
)
def _skipped(decorated: type) -> bool:
return getattr(decorated, "__unittest_skip__", False)
def _new_case() -> type:
return type("Case", (unittest.TestCase,), {})
class TestSelectedBackend(unittest.TestCase):
def test_defaults_to_docker_when_unset(self):
with patch.dict(os.environ, {}, clear=True):
self.assertEqual("docker", selected_backend())
def test_reads_env(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True):
self.assertEqual("firecracker", selected_backend())
class TestSkipUnlessBackend(unittest.TestCase):
def test_skips_when_other_backend_selected(self):
# A different backend is selected — no host probe needed, skip.
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready") as has:
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
has.assert_not_called()
def test_runs_when_selected_and_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True):
decorated = skip_unless_backend("docker")(_new_case())
self.assertFalse(_skipped(decorated))
def test_skips_when_selected_but_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase):
def test_runs_when_selected_backend_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True) as has:
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertFalse(_skipped(decorated))
has.assert_called_once_with("firecracker", quiet=False)
def test_skips_when_selected_backend_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertTrue(_skipped(decorated))
if __name__ == "__main__":
unittest.main()
-35
View File
@@ -1,35 +0,0 @@
"""Tests for integration-test backend selection helpers."""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._docker import skip_unless_docker_or_firecracker
class TestSkipUnlessDockerOrFirecracker(unittest.TestCase):
def test_firecracker_runs_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "firecracker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertFalse(getattr(decorated, "__unittest_skip__", False))
def test_non_firecracker_still_skips_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "docker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertTrue(getattr(decorated, "__unittest_skip__", False))
if __name__ == "__main__":
unittest.main()
+3 -1
View File
@@ -136,7 +136,9 @@ class TestFirecrackerStatus(unittest.TestCase):
from bot_bottle.backend.firecracker import setup as fc_setup from bot_bottle.backend.firecracker import setup as fc_setup
with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \ with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
patch.object(fc_setup.netpool, "overlapping_routes", 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() rc, out = self._run()
self.assertEqual(0, rc) self.assertEqual(0, rc)
self.assertIn("unverified", out) self.assertIn("unverified", out)