Compare commits

..

8 Commits

Author SHA1 Message Date
didericis-claude 65a49a239e firecracker status(): open /dev/kvm O_RDWR; check kernel, dropbear, mke2fs
test / integration-docker (pull_request) Successful in 19s
tracker-policy-pr / check-pr (pull_request) Successful in 19s
test / unit (pull_request) Successful in 40s
lint / lint (push) Successful in 58s
test / integration-firecracker (pull_request) Successful in 3m20s
test / coverage (pull_request) Successful in 22s
test / publish-infra (pull_request) Has been skipped
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 <noreply@anthropic.com>
2026-07-24 16:03:51 +00:00
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
32 changed files with 742 additions and 260 deletions
+8 -6
View File
@@ -103,14 +103,16 @@ jobs:
- name: Install coverage
run: python3 -m pip install --break-system-packages coverage
- name: Show environment
# Fail loudly if the backend this job promises isn't actually usable,
# rather than letting every test silently `unittest.skip` and the job
# go green on zero coverage. `backend status` prints a clear per-check
# summary (docker on PATH, daemon reachable) and exits non-zero when a
# prerequisite is missing — the same readiness check the skip guards
# gate on via `has_backend`.
- name: Preflight — Docker backend is ready
run: |
python3 --version
if command -v docker >/dev/null 2>&1; then
docker version || true
else
echo "docker not on PATH — integration tests will skip"
fi
python3 cli.py backend status --backend=docker
- name: Run integration tests (docker) with coverage
env:
-1
View File
@@ -35,7 +35,6 @@ backend remains available with `BOT_BOTTLE_BACKEND=docker` or
- `.bot-bottle/` — per-repo agent and bottle manifests (YAML markdown format).
- `examples/` — example bottles and agents showing the manifest format.
- `docs/README.md` — docs overview; when to write which document.
- `docs/glossary.md` — canonical term definitions (Agent Provider, Bottle, Sealed Bottle, etc.).
- `docs/prds/` — product requirement docs (see `docs/prds/README.md` for format).
- `docs/research/` — research notes (see `docs/research/README.md`).
- `docs/decisions/` — decision records (ADR-lite).
+47 -8
View File
@@ -21,7 +21,7 @@ backend exposes five methods:
enumerate_active() -> Sequence[ActiveAgent]
Return every currently-running bottle on this backend, with
enough metadata for callers (CLI `active`, dashboard
enough metadata for callers (CLI `list active`, dashboard
agents pane) to render a row.
Selection is driven by `--backend` on `start` or BOT_BOTTLE_BACKEND
@@ -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
@@ -200,7 +207,7 @@ class ExecResult:
@dataclass(frozen=True)
class ActiveAgent:
"""One currently-running agent, as the CLI `active` and
"""One currently-running agent, as the CLI `list active` and
dashboard agents pane render it. ("Agent" is the project's
consistent name for the thing running inside a bottle — the
bottle is the container, the agent is what runs in it.)
@@ -593,7 +600,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
Linux + KVM. Used by the cross-backend
`enumerate_active_agents` / `cmd_cleanup` to skip backends
the operator hasn't installed, so a docker-only host
doesn't fail when `cli.py active` walks past
doesn't fail when `cli.py list active` walks past
firecracker."""
@classmethod
@@ -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,9 +824,32 @@ 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 `active` and the dashboard's agents
backend. Used by CLI `list active` and the dashboard's agents
pane so neither has to know which backends exist. Skips
backends whose `is_available()` reports False.
@@ -835,6 +871,7 @@ def enumerate_active_agents() -> list[ActiveAgent]:
__all__ = [
"ActiveAgent",
"BackendStatus",
"Bottle",
"BottleBackend",
"BottleCleanupPlan",
@@ -847,5 +884,7 @@ __all__ = [
"get_bottle_backend",
"get_freezer",
"has_backend",
"is_backend_available",
"is_backend_ready",
"known_backend_names",
]
+6 -2
View File
@@ -20,7 +20,8 @@ infrastructure: CA install and git copy-in.
from __future__ import annotations
import shutil
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -60,8 +61,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+2 -2
View File
@@ -1,6 +1,6 @@
"""Active-agent enumeration for the docker backend.
Returns `ActiveAgent` records the CLI `active` command and the
Returns `ActiveAgent` records the CLI `list active` command and the
dashboard agents pane consume. Empty when docker isn't reachable
— gated by `has_backend('docker')` at the cross-backend caller
so this module trusts that docker is available when called.
@@ -60,7 +60,7 @@ def _parse_services_by_project(stdout: str) -> dict[str, set[str]]:
def _query_services_by_project() -> dict[str, set[str]]:
"""One `docker ps` call → `{project: {service, ...}}`. Used
by the CLI's `active` and the dashboard's agents pane —
by the CLI's `list active` and the dashboard's agents pane —
one subprocess per refresh tick, not one per bottle."""
try:
r = subprocess.run(
+6 -2
View File
@@ -8,7 +8,8 @@ fail-closed nftables egress boundary. Selected by
from __future__ import annotations
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -52,8 +53,11 @@ class FirecrackerBottleBackend(
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+86 -6
View File
@@ -13,6 +13,7 @@ generic `./cli.py backend {setup,status}` command dispatches to.
from __future__ import annotations
import fcntl
import os
import shutil
import subprocess
@@ -22,6 +23,9 @@ from pathlib import Path
from . import netpool
from . import util
# KVM_GET_API_VERSION = _IO(KVMIO=0xAE, 0x00): cheapest proof of KVM access.
_KVM_GET_API_VERSION = 0xAE00
_FC_RELEASES = "https://github.com/firecracker-microvm/firecracker/releases"
_UNIT_PATH = Path("/etc/systemd/system") / netpool.SYSTEMD_UNIT
@@ -219,14 +223,90 @@ def teardown() -> int:
return 0
def _firecracker_binary_ok() -> bool:
"""True iff the firecracker binary is on PATH and `--version` exits 0."""
if shutil.which("firecracker") is None:
return False
try:
return subprocess.run(
["firecracker", "--version"],
capture_output=True, check=False, timeout=5,
).returncode == 0
except (OSError, subprocess.TimeoutExpired):
return False
def _kvm_accessible() -> bool:
"""True iff /dev/kvm can be opened read-write and responds to KVM_GET_API_VERSION.
VM creation requires write access; opening read-only may satisfy the
ioctl but fails at boot time, so O_RDWR is the permission check."""
if not os.path.exists(util._KVM_DEVICE):
return False
try:
fd = os.open(util._KVM_DEVICE, os.O_RDWR | os.O_CLOEXEC)
try:
fcntl.ioctl(fd, _KVM_GET_API_VERSION)
finally:
os.close(fd)
return True
except OSError:
return False
def status() -> int:
# Readiness == what the launch preflight hard-requires: the TAP pool
# present (unprivileged, authoritative) and no range overlap. Listing
# the nft table usually needs root, so — like the preflight — an
# unconfirmable table is reported but NOT treated as not-ready; the
# post-boot isolation probe is the authoritative check. This keeps an
# unprivileged `backend status` usable as a launch gate.
# Readiness == what the launch preflight hard-requires: the binary
# executable, /dev/kvm accessible, the TAP pool present, and no range
# overlap. Listing the nft table usually needs root, so — like the
# preflight — an unconfirmable table is reported but NOT treated as
# not-ready; the post-boot isolation probe is the authoritative check.
# This keeps an unprivileged `backend status` usable as a launch gate.
ok = True
if _firecracker_binary_ok():
sys.stderr.write(f"firecracker binary: ok ({shutil.which('firecracker')})\n")
else:
fc_path = shutil.which("firecracker")
if fc_path is None:
sys.stderr.write("firecracker binary: NOT found on PATH\n")
else:
sys.stderr.write(
f"firecracker binary: found ({fc_path}) but `--version` failed\n"
)
ok = False
if _kvm_accessible():
sys.stderr.write(f"KVM: {util._KVM_DEVICE} accessible\n")
else:
if not os.path.exists(util._KVM_DEVICE):
sys.stderr.write(f"KVM: {util._KVM_DEVICE} not present\n")
else:
sys.stderr.write(
f"KVM: {util._KVM_DEVICE} not accessible (open/ioctl failed)\n"
)
ok = False
kernel = util.kernel_path()
if kernel.is_file():
sys.stderr.write(f"guest kernel: {kernel}\n")
else:
sys.stderr.write(
f"guest kernel: NOT found at {kernel} "
f"(set BOT_BOTTLE_FC_KERNEL or cache a vmlinux there)\n"
)
ok = False
dropbear = util.dropbear_path()
if dropbear.is_file():
sys.stderr.write(f"dropbear: {dropbear}\n")
else:
sys.stderr.write(
f"dropbear: NOT found at {dropbear} "
f"(set BOT_BOTTLE_FC_DROPBEAR or cache a static binary)\n"
)
ok = False
mke2fs = shutil.which("mke2fs")
if mke2fs is not None:
sys.stderr.write(f"mke2fs: {mke2fs}\n")
else:
sys.stderr.write("mke2fs: NOT found on PATH (install e2fsprogs)\n")
ok = False
missing = netpool.missing_taps()
total = netpool.pool_size()
if missing:
@@ -2,7 +2,8 @@
from __future__ import annotations
from contextlib import contextmanager
import io
from contextlib import contextmanager, redirect_stderr
from pathlib import Path
from typing import Generator, Sequence
@@ -43,8 +44,11 @@ class MacosContainerBottleBackend(
return _setup.setup()
@classmethod
def status(cls) -> int:
def status(cls, *, quiet: bool = False) -> int:
from . import setup as _setup
if quiet:
with redirect_stderr(io.StringIO()):
return _setup.status()
return _setup.status()
@classmethod
+5 -5
View File
@@ -1,6 +1,6 @@
"""Main CLI dispatcher.
Commands: active, backend, cleanup, commit, edit, init, list, resume, start, supervise
Commands: backend, cleanup, commit, edit, info, init, list, resume, start, supervise
"""
from __future__ import annotations
@@ -13,11 +13,11 @@ from ..manifest import ManifestError
from ..store_manager import StoreManager
from ._common import PROG
from . import list as _list_mod
from .active import cmd_active
from .backend import cmd_backend
from .cleanup import cmd_cleanup
from .commit import cmd_commit
from .edit import cmd_edit
from .info import cmd_info
from .init import cmd_init
from .login import cmd_login
from .resume import cmd_resume
@@ -27,11 +27,11 @@ from .supervise import cmd_supervise
cmd_list = _list_mod.cmd_list
COMMANDS = {
"active": cmd_active,
"backend": cmd_backend,
"cleanup": cmd_cleanup,
"commit": cmd_commit,
"edit": cmd_edit,
"info": cmd_info,
"init": cmd_init,
"list": cmd_list,
"login": cmd_login,
@@ -51,13 +51,13 @@ NO_MIGRATION_COMMANDS = frozenset({"backend", "login"})
def usage() -> None:
sys.stderr.write(f"usage: {PROG} <command> [args...]\n\n")
sys.stderr.write("Commands:\n")
sys.stderr.write(" active list currently-running bot-bottle bottles\n")
sys.stderr.write(" backend set up / check / undo a backend's host prerequisites (setup|status|teardown)\n")
sys.stderr.write(" cleanup stop and remove all active bot-bottle containers\n")
sys.stderr.write(" commit snapshot a running bottle's container state to a Docker image\n")
sys.stderr.write(" edit open an agent in vim for editing\n")
sys.stderr.write(" info print env, skills, and prompt details for a named agent\n")
sys.stderr.write(" init interactively create a new agent and add it to bot-bottle.json\n")
sys.stderr.write(" list list available agents from bot-bottle.json\n")
sys.stderr.write(" list list available agents or active containers\n")
sys.stderr.write(" login register this host with a bot-bottle console\n")
sys.stderr.write(
" resume re-launch a bottle by its identity "
-53
View File
@@ -1,53 +0,0 @@
"""active: list currently-running bot-bottle bottles."""
from __future__ import annotations
import os
import sys
from ..backend import enumerate_active_agents
from ._common import PROG
_ANSI_COLOR_CODES: dict[str, str] = {
"red": "\033[91m",
"green": "\033[92m",
"yellow": "\033[93m",
"blue": "\033[94m",
"magenta": "\033[95m",
}
_ANSI_RESET = "\033[0m"
def _ansi_label(text: str, color: str) -> str:
if not color:
return text
if not sys.stdout.isatty():
return text
term = os.environ.get("TERM", "")
if term in ("dumb", ""):
return text
code = _ANSI_COLOR_CODES.get(color)
if not code:
return text
return f"{code}{text}{_ANSI_RESET}"
def cmd_active(argv: list[str]) -> int:
if argv and argv[0] in ("-h", "--help"):
sys.stderr.write(f"usage: {PROG} active\n")
sys.stderr.write("\nList all currently-running bot-bottle bottles.\n")
sys.stderr.write("Output: <backend>\\t<slug>\\t<label>\\t<services>\n")
return 0
active = enumerate_active_agents()
if not active:
print("no active bot-bottle bottles", file=sys.stderr)
return 0
# One line per bottle: `<backend>\t<slug>\t<label>\t<services>`.
# Tab-separated keeps the format stable for shell pipelines.
for b in active:
services = ",".join(b.services) if b.services else "-"
display_name = f"{b.label} ({b.agent_name})" if b.label else b.agent_name
colored_name = _ansi_label(display_name, b.color)
print(f"{b.backend_name}\t{b.slug}\t{colored_name}\t{services}")
return 0
+1 -1
View File
@@ -27,7 +27,7 @@ def cmd_commit(argv: list[str]) -> int:
nargs="?",
default=None,
help=(
"bottle slug from `cli.py active` "
"bottle slug from `cli.py list active` "
"(omit to pick interactively)"
),
)
+49
View File
@@ -0,0 +1,49 @@
"""info: print env, skills, and prompt details for a named agent."""
from __future__ import annotations
import argparse
from ..log import info
from ..manifest import ManifestIndex
from ._common import PROG, USER_CWD
def cmd_info(argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog=f"{PROG} info", add_help=True)
parser.add_argument("name", help="agent name defined in bot-bottle.json")
args = parser.parse_args(argv)
names = ManifestIndex.resolve(USER_CWD)
names.require_agent(args.name)
manifest = names.load_for_agent(args.name)
agent = manifest.agent
bottle = manifest.bottle
env_names = list(bottle.env.keys())
prompt_first_line = agent.prompt.splitlines()[0] if agent.prompt else ""
print()
info(f"agent : {args.name}")
info(f"env (names only): {', '.join(env_names) if env_names else '(none)'}")
info(f"skills : {' '.join(agent.skills) if agent.skills else '(none)'}")
info(
f"prompt : {len(agent.prompt)} chars; "
f"first line: {prompt_first_line or '(empty)'}"
)
info(f"bottle : {agent.bottle}")
identity = manifest.git_identity_summary()
if identity:
info(f" git identity : {identity}")
if bottle.git:
for e in bottle.git:
info(
f" git remote : {e.Name} -> {e.Upstream} "
f"(IdentityFile={e.IdentityFile})"
)
if e.KnownHostKey:
info(f" KnownHostKey: {e.KnownHostKey}")
else:
info(" git remotes : (none)")
print()
return 0
+49 -7
View File
@@ -1,20 +1,62 @@
"""list: list available agents."""
"""list: list available agents or active bottles."""
from __future__ import annotations
import argparse
import os
import sys
from ..backend import enumerate_active_agents
from ..manifest import ManifestIndex
from ._common import PROG, USER_CWD
_ANSI_COLOR_CODES: dict[str, str] = {
"red": "\033[91m",
"green": "\033[92m",
"yellow": "\033[93m",
"blue": "\033[94m",
"magenta": "\033[95m",
}
_ANSI_RESET = "\033[0m"
def _ansi_label(text: str, color: str) -> str:
if not color:
return text
if not sys.stdout.isatty():
return text
term = os.environ.get("TERM", "")
if term in ("dumb", ""):
return text
code = _ANSI_COLOR_CODES.get(color)
if not code:
return text
return f"{code}{text}{_ANSI_RESET}"
def cmd_list(argv: list[str]) -> int:
if argv and argv[0] in ("-h", "--help"):
sys.stderr.write(f"usage: {PROG} list\n")
sys.stderr.write("\nList all available agents from bot-bottle.json.\n")
parser = argparse.ArgumentParser(prog=f"{PROG} list", add_help=True)
parser.add_argument("scope", choices=["available", "active"])
args = parser.parse_args(argv)
if args.scope == "available":
manifest = ManifestIndex.resolve(USER_CWD)
for name in manifest.all_agent_names:
print(name)
return 0
manifest = ManifestIndex.resolve(USER_CWD)
for name in manifest.all_agent_names:
print(name)
# `active` enumerates every backend (docker, firecracker,
# macos-container) so non-docker bottles aren't hidden behind
# the env var.
active = enumerate_active_agents()
if not active:
print("no active bot-bottle bottles", file=sys.stderr)
return 0
# One line per bottle: `<backend>\t<slug>\t<label>\t<services>`.
# Tab-separated keeps the format stable for shell pipelines.
for b in active:
services = ",".join(b.services) if b.services else "-"
display_name = f"{b.label} ({b.agent_name})" if b.label else b.agent_name
colored_name = _ansi_label(display_name, b.color)
print(f"{b.backend_name}\t{b.slug}\t{colored_name}\t{services}")
return 0
-1
View File
@@ -7,7 +7,6 @@ picking the right document for what you're capturing.
| Artifact | For |
|---|---|
| **Glossary** (`docs/glossary.md`) | Canonical term definitions — what words mean in this project. |
| **PRD** (`docs/prds/`) | A feature: what to build, scope, success criteria. |
| **Research note** (`docs/research/`) | A landscape/tradeoff investigation. |
| **Decision record** (`docs/decisions/`) | A decision that isn't itself a feature — a policy, a convention, a "we will / won't do this," or a load-bearing choice made inside a larger PRD that deserves to be discoverable on its own. |
+13 -3
View File
@@ -1,13 +1,23 @@
# CI
The test workflow lives at [`.gitea/workflows/test.yml`](../.gitea/workflows/test.yml).
It runs `tests/run_tests.py` (full suite — unit + integration) on:
It runs the unit suite plus one integration job per backend
(`integration-docker`, `integration-firecracker`) on:
- every push to a branch with an open pull request, and
- every push to `main`.
Integration tests need Docker on the runner; they skip cleanly via
`tests/_docker.skip_unless_docker` when no daemon is reachable.
Each integration job selects its backend via `BOT_BOTTLE_BACKEND` and
runs a **preflight** (`./cli.py backend status --backend=<name>`) that
prints a clear per-check readiness summary and fails the job when the
backend is missing — so absent infrastructure is visible at the job level
rather than hidden among per-test `unittest.skip` lines. The skip guards in
[`tests/_backend.py`](../tests/_backend.py) gate on the same readiness
check (`bot_bottle.backend.has_backend`): backend-agnostic tests use
`skip_unless_selected_backend_available()` and run through whichever
backend is selected (checking, e.g., Linux + `/dev/kvm` for Firecracker
rather than unrelated Docker availability); Docker-implementation tests use
`skip_unless_backend("docker")` and no-op under a non-Docker run.
A small subset of integration tests skip when running specifically
under Gitea Actions (`GITEA_ACTIONS=true`), because `act_runner` runs
-57
View File
@@ -1,57 +0,0 @@
# Glossary
Canonical terminology for bot-bottle. Prefer these names in docs, comments, and UI.
---
## Agent Provider
The component that connects to an external model provider and sets up the model
harness inside the agent runtime. Configured in a bottle manifest under
`agent_provider:`; built-in templates are `claude` and `codex`. Responsible for
provider-specific auth, startup args, and egress routes.
## Agent Runtime
The OCI image (and the container or VM it runs in) that houses the model
harness. On the macOS-container backend this is an Apple Container; on
Firecracker it is a microVM; on the legacy Docker backend it is a Docker
container. The agent runtime is built from the agent provider's Dockerfile
(built-in or custom).
## Agent / Agent Definition
A Markdown file with YAML frontmatter that declares the system prompt and
identity of the model harness. Lives under `~/.bot-bottle/agents/` or a repo's
`.bot-bottle/agents/`. Specifies which bottle to run under (`bottle:`) and
which skills to load. Agent definitions are safe to commit; they contain no
secrets or egress policy.
## Bottle / Bottle Definition
A Markdown file with YAML frontmatter that declares the security and runtime
boundaries for one agent runtime: egress allowlist, git remotes, env vars,
nested-container flag, and agent provider config. Lives under
`~/.bot-bottle/bottles/`. Bottles are scoped to `$HOME` so a cloned repo
cannot override host egress policy.
## Sealed Bottle
The fully-resolved bottle after all `extends:` inheritance is applied and every
field has been validated. The sealed bottle is the immutable boundary spec that
the launcher enforces — no further overrides are possible once it is sealed.
## Bottled Agent
The combination of an Agent Definition and a Sealed Bottle, representing a
single deployable agent with a fixed identity and fixed boundaries. A bottled
agent has two observable states:
- **Active** — the agent runtime is running and the agent is executing.
- **Frozen** — the agent runtime has been snapshotted (Firecracker committed
image); the agent is not running but can be resumed from the snapshot.
## Active Bottle
Shorthand for an active (running) Bottled Agent. Used in the supervisor TUI
and discovery layer to mean "a bottle whose agent runtime is currently up."
+20 -5
View File
@@ -2,14 +2,15 @@
Plain-Python test suite using stdlib `unittest`. No external
dependencies. Unit tests run anywhere Python 3 is present; integration
tests need Docker and skip cleanly otherwise.
tests run through the backend named by `BOT_BOTTLE_BACKEND` (default
`docker`) and skip cleanly when that backend isn't available on the host.
## Layout
```
tests/
fixtures.py # JSON manifest builders (shared)
_docker.py # docker-availability skip helper (shared)
_backend.py # backend selection + skip guards (shared)
unit/
test_egress.py
test_egress_addon_core.py
@@ -73,7 +74,7 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
## Adding a test
1. Pick the directory: `tests/unit/` for a pure unit test,
`tests/integration/` for one that needs Docker.
`tests/integration/` for one that needs a backend.
2. Filename: `test_<topic>.py`.
3. Boilerplate:
```python
@@ -88,5 +89,19 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
if __name__ == "__main__":
unittest.main()
```
4. For Docker-dependent tests, decorate the class with
`@skip_unless_docker()` from `tests._docker`.
4. Skip guards live in `tests._backend` and gate on the backend's own
readiness check, `bot_bottle.backend.has_backend` — the same probe
behind `./cli.py backend status`:
- Backend-agnostic tests (go through `get_bottle_backend()`) decorate
the class with `@skip_unless_selected_backend_available()` — the test
runs against whichever backend `BOT_BOTTLE_BACKEND` selects and skips
unless that backend is available (checking, e.g., Linux + `/dev/kvm`
for Firecracker rather than unrelated Docker availability).
- Backend-specific tests (exercise `DockerBroker`, `DockerGateway`,
`backend.docker.*`, …) decorate with `@skip_unless_backend("docker")`
so they no-op under a run targeting a different backend.
Each CI integration job runs `./cli.py backend status --backend=<name>`
as a preflight, which prints a clear per-check summary and exits non-zero
when the backend is missing — so absent infrastructure fails the job
instead of hiding among per-test `unittest.skip` lines.
+71
View File
@@ -0,0 +1,71 @@
"""Backend selection + readiness-aware skip guards for the integration suite.
Each integration test targets the backend named by ``BOT_BOTTLE_BACKEND``
(default ``docker``) and gates on that backend's full readiness check —
``is_backend_ready()`` (equivalent to ``./cli.py backend status``), not just
a binary-on-PATH probe. When the backend is not ready, diagnostic output is
printed during test discovery so the operator sees a concrete reason for each
skip.
"""
from __future__ import annotations
import os
import unittest
from bot_bottle.backend import is_backend_ready
# Default when ``BOT_BOTTLE_BACKEND`` is unset. Docker preserves the historical
# Docker-backed CI path (and mirrors the pin in ``test_sandbox_escape``).
DEFAULT_BACKEND = "docker"
def selected_backend() -> str:
"""The backend this test run targets, from ``BOT_BOTTLE_BACKEND``.
Mirrors the CLI's env selector; unset means ``docker`` so an
unconfigured run behaves exactly as the suite did before backends were
pluggable.
"""
return os.environ.get("BOT_BOTTLE_BACKEND") or DEFAULT_BACKEND
def skip_unless_backend(backend: str):
"""Skip a backend-specific test unless the selected backend matches AND
that backend is fully ready.
Docker-implementation tests (``DockerBroker``, ``DockerGateway``,
``backend.docker.*``) use ``skip_unless_backend("docker")`` so they no-op
under a run targeting a different backend instead of testing Docker
internals that run doesn't exercise — the guard reads
``BOT_BOTTLE_BACKEND`` rather than "is Docker installed".
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
sel = selected_backend()
if sel != backend:
return unittest.skip(
f"backend {backend!r} not selected (BOT_BOTTLE_BACKEND={sel})"
)
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"{backend} backend not ready",
)
def skip_unless_selected_backend_available():
"""Skip a backend-agnostic test unless the *selected* backend is fully ready.
The test then runs through whichever backend ``BOT_BOTTLE_BACKEND`` names,
gated on that backend's full status() check (e.g. daemon reachable, TAP
pool present for Firecracker) rather than just a binary-on-PATH probe.
When the backend is not ready, ``status()`` output is printed so the
operator sees a concrete diagnostic for each skipped test module.
"""
backend = selected_backend()
return unittest.skipUnless(
is_backend_ready(backend, quiet=False),
f"selected backend {backend!r} not ready",
)
-45
View File
@@ -1,45 +0,0 @@
"""Docker availability check used by integration tests."""
from __future__ import annotations
import os
import shutil
import subprocess
import unittest
def docker_available() -> bool:
if os.environ.get("SKIP_DOCKER_TESTS"):
return False
if shutil.which("docker") is None:
return False
try:
return (
subprocess.run(
["docker", "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=5,
).returncode
== 0
)
except subprocess.TimeoutExpired:
return False
def skip_unless_docker(reason: str = "docker unreachable"):
return unittest.skipUnless(docker_available(), reason)
def skip_unless_docker_or_firecracker(
reason: str = "neither Docker nor Firecracker selected",
):
"""Skip a backend-agnostic test unless one supported backend can run.
Firecracker does not require the host Docker daemon. The KVM coverage job
deliberately sets ``SKIP_DOCKER_TESTS`` to exclude Docker-only integration
classes while still exercising this path.
"""
firecracker_selected = os.environ.get("BOT_BOTTLE_BACKEND") == "firecracker"
return unittest.skipUnless(firecracker_selected or docker_available(), reason)
+2 -2
View File
@@ -25,14 +25,14 @@ import os
import subprocess
import unittest
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
_IMAGE = "bot-bottle-gateway-test:chunk1"
_DOCKERFILE = "Dockerfile.gateway"
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: multi-stage build pulls a 200+MB "
@@ -31,7 +31,7 @@ from bot_bottle.backend.docker.gateway_net import next_free_ip
from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK
from bot_bottle.orchestrator.lifecycle import OrchestratorService
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
# One upstream reachable under two names; reflects the Authorization header so
# the probe can see exactly what the gateway injected (or didn't).
@@ -72,7 +72,7 @@ _PROBE_SRC = (
)
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -13,12 +13,12 @@ import unittest
from bot_bottle.orchestrator.broker import LaunchRequest, sign_request
from bot_bottle.orchestrator.docker_broker import DockerBroker, container_name
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerBrokerIntegration(unittest.TestCase):
def setUp(self) -> None:
self.secret = secrets.token_bytes(16)
@@ -28,7 +28,7 @@ from pathlib import Path
from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.lifecycle import OrchestratorService
from bot_bottle.paths import host_control_plane_token
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
# Fixed (not per-run-suffixed) so repeated runs reuse the same layer-cached
# image instead of leaking a new dangling tag on every invocation.
@@ -37,7 +37,7 @@ _TEST_GATEWAY_IMAGE = "bot-bottle-gateway:itest"
_TEST_INFRA_IMAGE = "bot-bottle-infra:itest"
@skip_unless_docker()
@skip_unless_backend("docker")
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo "
@@ -11,12 +11,12 @@ import subprocess
import unittest
from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerGatewayIntegration(unittest.TestCase):
def setUp(self) -> None:
self.name = "bot-bottle-orch-gateway-itest-" + secrets.token_hex(4)
@@ -12,12 +12,12 @@ import subprocess
import unittest
from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
IMAGE = "busybox"
@skip_unless_docker()
@skip_unless_backend("docker")
class TestDockerGatewayImageExists(unittest.TestCase):
def test_image_exists_true_for_present_false_for_absent(self) -> None:
# Ensure the tiny image is present (build_if_missing is disabled here
+2 -2
View File
@@ -18,10 +18,10 @@ from bot_bottle.backend.docker.network import (
network_create_internal,
network_remove,
)
from tests._docker import skip_unless_docker
from tests._backend import skip_unless_backend
@skip_unless_docker()
@skip_unless_backend("docker")
class TestOrphanCleanup(unittest.TestCase):
def setUp(self):
self.slug = f"cb-test-orphan-{os.getpid()}"
+2 -2
View File
@@ -31,7 +31,7 @@ from pathlib import Path
from bot_bottle.backend import BottleSpec, get_bottle_backend
from bot_bottle.bottle_state import cleanup_state
from bot_bottle.manifest import ManifestIndex
from tests._docker import skip_unless_docker_or_firecracker
from tests._backend import skip_unless_selected_backend_available
# Secrets planted in the bottle env as literals (agents substitute via
@@ -67,7 +67,7 @@ _DUMMY_HOST_KEY = (
)
@skip_unless_docker_or_firecracker()
@skip_unless_selected_backend_available()
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true"
and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
+54
View File
@@ -385,6 +385,60 @@ class TestHasBackend(unittest.TestCase):
self.assertFalse(has_backend("nonexistent"))
class TestIsBackendAvailable(unittest.TestCase):
def test_delegates_to_has_backend(self):
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_available("docker"))
def test_known_and_available(self):
class _Ready:
def is_available(self):
return True
from bot_bottle.backend import is_backend_available
with patch.object(backend_mod, "_backends", {"docker": _Ready()}):
self.assertTrue(is_backend_available("docker"))
class TestIsBackendReady(unittest.TestCase):
def test_unknown_backend_returns_false(self):
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {}):
self.assertFalse(is_backend_ready("docker"))
def test_ready_when_status_returns_zero(self):
class _ReadyBackend:
def status(self, *, quiet: bool = False) -> int:
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _ReadyBackend()}):
self.assertTrue(is_backend_ready("docker"))
def test_not_ready_when_status_nonzero(self):
class _BrokenBackend:
def status(self, *, quiet: bool = False) -> int:
return 1
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _BrokenBackend()}):
self.assertFalse(is_backend_ready("docker"))
def test_quiet_flag_forwarded(self):
calls = []
class _SpyBackend:
def status(self, *, quiet: bool = False) -> int:
calls.append(quiet)
return 0
from bot_bottle.backend import is_backend_ready
with patch.object(backend_mod, "_backends", {"docker": _SpyBackend()}):
is_backend_ready("docker", quiet=True)
self.assertEqual([True], calls)
class TestEnsureOrchestrator(unittest.TestCase):
"""The backend-agnostic orchestrator bring-up entry point. Docker starts
the orchestrator + gateway containers; firecracker boots the infra VM;
+196
View File
@@ -328,5 +328,201 @@ class TestNetpoolShellRenderers(unittest.TestCase):
self.assertIn("BOT_BOTTLE_FC_POOL_SIZE=4", out)
class TestFirecrackerBinaryCheck(unittest.TestCase):
def test_binary_missing_returns_false(self):
with patch.object(fc.shutil, "which", return_value=None):
self.assertFalse(fc._firecracker_binary_ok())
def test_binary_present_and_runs_ok(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run",
return_value=subprocess.CompletedProcess([], 0)):
self.assertTrue(fc._firecracker_binary_ok())
def test_binary_found_but_exits_nonzero(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run",
return_value=subprocess.CompletedProcess([], 1)):
self.assertFalse(fc._firecracker_binary_ok())
def test_binary_found_but_oserror(self):
with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
patch.object(fc.subprocess, "run", side_effect=OSError("exec failed")):
self.assertFalse(fc._firecracker_binary_ok())
class TestFirecrackerKvmCheck(unittest.TestCase):
def test_kvm_device_absent_returns_false(self):
with patch.object(fc.os.path, "exists", return_value=False):
self.assertFalse(fc._kvm_accessible())
def test_kvm_accessible_when_ioctl_succeeds(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", return_value=42), \
patch.object(fc.os, "close"), \
patch.object(fc.fcntl, "ioctl", return_value=12):
self.assertTrue(fc._kvm_accessible())
def test_kvm_open_rdwr_fails(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", side_effect=OSError("Permission denied")):
self.assertFalse(fc._kvm_accessible())
def test_kvm_present_but_ioctl_fails(self):
with patch.object(fc.os.path, "exists", return_value=True), \
patch.object(fc.os, "open", return_value=42), \
patch.object(fc.os, "close"), \
patch.object(fc.fcntl, "ioctl", side_effect=OSError("permission denied")):
self.assertFalse(fc._kvm_accessible())
class TestFirecrackerArtifactCheck(unittest.TestCase):
"""status() reports missing guest kernel, dropbear binary, and mke2fs."""
def _apply_all_ok(self, stack: contextlib.ExitStack) -> None:
"""Stub every status() check to pass except what the test overrides."""
stack.enter_context(patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(patch.object(fc, "_kvm_accessible", return_value=True))
k: MagicMock = MagicMock()
k.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
d: MagicMock = MagicMock()
d.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
stack.enter_context(patch.object(fc.shutil, "which", return_value="/usr/bin/x"))
stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
def test_status_fails_when_kernel_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
k: MagicMock = MagicMock()
k.is_file.return_value = False
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("guest kernel: NOT found", out)
def test_status_fails_when_dropbear_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
d: MagicMock = MagicMock()
d.is_file.return_value = False
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("dropbear: NOT found", out)
def test_status_fails_when_mke2fs_missing(self):
with contextlib.ExitStack() as stack:
self._apply_all_ok(stack)
stack.enter_context(patch.object(
fc.shutil, "which",
side_effect=lambda cmd: (None if cmd == "mke2fs" else "/usr/bin/x"), # type: ignore[misc]
))
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("mke2fs: NOT found", out)
class TestFirecrackerStatusRuntime(unittest.TestCase):
"""status() reports binary and KVM problems and returns non-zero."""
def _apply_pool_ok(self, stack: contextlib.ExitStack) -> None:
kernel_mock = MagicMock()
kernel_mock.is_file.return_value = True
dropbear_mock = MagicMock()
dropbear_mock.is_file.return_value = True
stack.enter_context(patch.object(fc.util, "kernel_path", return_value=kernel_mock))
stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=dropbear_mock))
stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
def test_status_fails_when_binary_missing(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=False))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value=None))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("NOT found on PATH", out)
def test_status_fails_when_kvm_not_accessible(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=False))
stack.enter_context(
patch.object(fc.os.path, "exists", return_value=True))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(1, rc)
self.assertIn("not accessible", out)
def test_status_ok_when_binary_and_kvm_ready(self):
with contextlib.ExitStack() as stack:
stack.enter_context(
patch.object(fc, "_firecracker_binary_ok", return_value=True))
stack.enter_context(
patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
stack.enter_context(
patch.object(fc, "_kvm_accessible", return_value=True))
stack.enter_context(
patch.object(netpool, "nft_table_present", return_value=True))
self._apply_pool_ok(stack)
rc, out = _cap(fc.status)
self.assertEqual(0, rc)
self.assertIn("firecracker binary: ok", out)
self.assertIn("KVM:", out)
class TestBackendStatusQuiet(unittest.TestCase):
"""status(quiet=True) routes the underlying status() call through
redirect_stderr so diagnostic output is suppressed. Verify the return
code is propagated and that calling with quiet=False leaves the non-quiet
path active (covered by the other TestDockerSetupStatus tests)."""
def test_docker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.docker.backend import DockerBottleBackend
with patch.object(dk, "status", return_value=0):
self.assertEqual(0, DockerBottleBackend.status(quiet=True))
def test_docker_quiet_true_suppresses_stderr(self):
import io as _io
from bot_bottle.backend.docker.backend import DockerBottleBackend
def _loud_status():
import sys
print("should be suppressed", file=sys.stderr)
return 0
with patch.object(dk, "status", side_effect=_loud_status):
buf = _io.StringIO()
with contextlib.redirect_stderr(buf):
DockerBottleBackend.status(quiet=True)
self.assertEqual("", buf.getvalue())
def test_firecracker_quiet_true_propagates_return_code(self):
from bot_bottle.backend.firecracker.backend import FirecrackerBottleBackend
with patch.object(fc, "status", return_value=1):
self.assertEqual(1, FirecrackerBottleBackend.status(quiet=True))
def test_macos_quiet_true_propagates_return_code(self):
from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
with patch.object(mc, "status", return_value=0):
self.assertEqual(0, MacosContainerBottleBackend.status(quiet=True))
if __name__ == "__main__":
unittest.main()
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the backend-aware skip guards in ``tests/_backend.py``.
The guards delegate their readiness check to
``bot_bottle.backend.is_backend_ready`` (the probe behind ``./cli.py backend
status``); here that probe is mocked so the unit job asserts the
selection/skip logic without either backend present on the runner.
"""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._backend import (
selected_backend,
skip_unless_backend,
skip_unless_selected_backend_available,
)
def _skipped(decorated: type) -> bool:
return getattr(decorated, "__unittest_skip__", False)
def _new_case() -> type:
return type("Case", (unittest.TestCase,), {})
class TestSelectedBackend(unittest.TestCase):
def test_defaults_to_docker_when_unset(self):
with patch.dict(os.environ, {}, clear=True):
self.assertEqual("docker", selected_backend())
def test_reads_env(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True):
self.assertEqual("firecracker", selected_backend())
class TestSkipUnlessBackend(unittest.TestCase):
def test_skips_when_other_backend_selected(self):
# A different backend is selected — no host probe needed, skip.
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready") as has:
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
has.assert_not_called()
def test_runs_when_selected_and_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True):
decorated = skip_unless_backend("docker")(_new_case())
self.assertFalse(_skipped(decorated))
def test_skips_when_selected_but_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_backend("docker")(_new_case())
self.assertTrue(_skipped(decorated))
class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase):
def test_runs_when_selected_backend_available(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=True) as has:
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertFalse(_skipped(decorated))
has.assert_called_once_with("firecracker", quiet=False)
def test_skips_when_selected_backend_unavailable(self):
with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
patch("tests._backend.is_backend_ready", return_value=False):
decorated = skip_unless_selected_backend_available()(_new_case())
self.assertTrue(_skipped(decorated))
if __name__ == "__main__":
unittest.main()
-35
View File
@@ -1,35 +0,0 @@
"""Tests for integration-test backend selection helpers."""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from tests._docker import skip_unless_docker_or_firecracker
class TestSkipUnlessDockerOrFirecracker(unittest.TestCase):
def test_firecracker_runs_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "firecracker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertFalse(getattr(decorated, "__unittest_skip__", False))
def test_non_firecracker_still_skips_when_docker_tests_are_disabled(self):
with patch.dict(
os.environ,
{"BOT_BOTTLE_BACKEND": "docker", "SKIP_DOCKER_TESTS": "1"},
clear=True,
):
decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
self.assertTrue(getattr(decorated, "__unittest_skip__", False))
if __name__ == "__main__":
unittest.main()
+29 -3
View File
@@ -132,30 +132,56 @@ class TestFirecrackerStatus(unittest.TestCase):
rc = fc_setup.status()
return rc, buf.getvalue()
def _stub_artifacts(self, fc_setup: object) -> tuple[MagicMock, MagicMock]:
k: MagicMock = MagicMock()
k.is_file.return_value = True
d: MagicMock = MagicMock()
d.is_file.return_value = True
return k, d
def test_ready_when_taps_present_even_if_nft_unverifiable(self):
from bot_bottle.backend.firecracker import setup as fc_setup
def _which(cmd: str) -> str | None:
return None if cmd == "nft" else f"/usr/bin/{cmd}"
k, d = self._stub_artifacts(fc_setup)
with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", side_effect=_which), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, out = self._run()
self.assertEqual(0, rc)
self.assertIn("unverified", out)
def test_not_ready_when_taps_missing(self):
from bot_bottle.backend.firecracker import setup as fc_setup
k, d = self._stub_artifacts(fc_setup)
with patch.object(fc_setup.netpool, "missing_taps", return_value=["bbfc0"]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", return_value=None), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, _ = self._run()
self.assertEqual(1, rc)
def test_not_ready_on_range_overlap(self):
from bot_bottle.backend.firecracker import netpool
from bot_bottle.backend.firecracker import setup as fc_setup
k, d = self._stub_artifacts(fc_setup)
conflict = netpool.RouteConflict(dst="10.243.0.0/24", dev="eth0")
with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
patch.object(fc_setup.netpool, "overlapping_routes", return_value=[conflict]), \
patch.object(fc_setup.shutil, "which", return_value=None):
patch.object(fc_setup.shutil, "which", return_value=None), \
patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
patch.object(fc_setup, "_kvm_accessible", return_value=True), \
patch.object(fc_setup.util, "kernel_path", return_value=k), \
patch.object(fc_setup.util, "dropbear_path", return_value=d):
rc, out = self._run()
self.assertEqual(1, rc)
self.assertIn("CLASHES", out)