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>
This commit is contained in:
2026-07-24 03:03:00 +00:00
parent d4d45f835e
commit 0ca39cf4d5
5 changed files with 82 additions and 21 deletions
+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
@@ -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
+21 -11
View File
@@ -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`` Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND``
(default ``docker``) and skips on that backend's own readiness check — (default ``docker``) and gates on that backend's full readiness check —
``bot_bottle.backend.has_backend`` (the same probe behind ``is_backend_ready()`` (equivalent to ``./cli.py backend status``), not just
``./cli.py backend status``), not an unrelated "is Docker installed" test. 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 from __future__ import annotations
@@ -11,7 +13,7 @@ from __future__ import annotations
import os import os
import unittest 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 # Default when ``BOT_BOTTLE_BACKEND`` is unset. Docker preserves the historical
# Docker-backed CI path (and mirrors the pin in ``test_sandbox_escape``). # 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): def skip_unless_backend(backend: str):
"""Skip a backend-specific test unless the selected backend matches AND """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``, Docker-implementation tests (``DockerBroker``, ``DockerGateway``,
``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op ``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op
under a run targeting a different backend instead of testing Docker under a run targeting a different backend instead of testing Docker
internals that run doesn't exercise — the guard reads internals that run doesn't exercise — the guard reads
``BOT_BOTTLE_BACKEND`` rather than "is Docker installed". ``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() sel = selected_backend()
if sel != backend: if sel != backend:
@@ -44,18 +49,23 @@ def skip_unless_backend(backend: str):
f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})" f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})"
) )
return unittest.skipUnless( 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(): 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, The test then runs through whichever backend ``BOT_BOTTLE_BACKEND`` names,
gated on that backend's real prerequisites (e.g. Linux + KVM for gated on that backend's full status() check (e.g. daemon reachable, TAP
Firecracker) rather than unrelated Docker availability. 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() backend = selected_backend()
return unittest.skipUnless( 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",
) )