Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1972c8c6e9 | |||
| 5d109ea290 |
@@ -103,16 +103,14 @@ jobs:
|
||||
- name: Install coverage
|
||||
run: python3 -m pip install --break-system-packages coverage
|
||||
|
||||
# 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
|
||||
- name: Show environment
|
||||
run: |
|
||||
python3 --version
|
||||
python3 cli.py backend status --backend=docker
|
||||
if command -v docker >/dev/null 2>&1; then
|
||||
docker version || true
|
||||
else
|
||||
echo "docker not on PATH — integration tests will skip"
|
||||
fi
|
||||
|
||||
- name: Run integration tests (docker) with coverage
|
||||
env:
|
||||
|
||||
@@ -35,6 +35,7 @@ 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).
|
||||
|
||||
@@ -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 `list active`, dashboard
|
||||
enough metadata for callers (CLI `active`, dashboard
|
||||
agents pane) to render a row.
|
||||
|
||||
Selection is driven by `--backend` on `start` or BOT_BOTTLE_BACKEND
|
||||
@@ -33,7 +33,6 @@ backend field; the host picks.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import enum
|
||||
import os
|
||||
import shlex
|
||||
import sys
|
||||
@@ -60,12 +59,6 @@ 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
|
||||
@@ -207,7 +200,7 @@ class ExecResult:
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ActiveAgent:
|
||||
"""One currently-running agent, as the CLI `list active` and
|
||||
"""One currently-running agent, as the CLI `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.)
|
||||
@@ -600,7 +593,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 list active` walks past
|
||||
doesn't fail when `cli.py active` walks past
|
||||
firecracker."""
|
||||
|
||||
@classmethod
|
||||
@@ -618,18 +611,12 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def status(cls, *, quiet: bool = False) -> int:
|
||||
def status(cls) -> int:
|
||||
"""Report whether this backend's prerequisites are satisfied on
|
||||
the host — binaries, daemon reachability, network pool, range
|
||||
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)."""
|
||||
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=…]`."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
@@ -824,32 +811,9 @@ def has_backend(name: str) -> bool:
|
||||
return backends[name].is_available()
|
||||
|
||||
|
||||
def is_backend_available(name: str) -> bool:
|
||||
"""Cheap availability check: is the backend's binary on PATH?
|
||||
|
||||
Suitable for cleanup enumeration and auto-selection — does NOT probe
|
||||
the daemon or network pool. Use is_backend_ready() for a full
|
||||
readiness check before launching tests."""
|
||||
return has_backend(name)
|
||||
|
||||
|
||||
def is_backend_ready(name: str, *, quiet: bool = False) -> bool:
|
||||
"""Full readiness check: passes all of the backend's status() checks.
|
||||
|
||||
When quiet=False the backend prints diagnostic output explaining what
|
||||
is missing — intended for test-suite guards that run at discovery time
|
||||
so the operator sees a concrete failure reason for each skip.
|
||||
|
||||
Returns False for unknown backend names."""
|
||||
backends = _get_backends()
|
||||
if name not in backends:
|
||||
return False
|
||||
return backends[name].status(quiet=quiet) == BackendStatus.READY
|
||||
|
||||
|
||||
def enumerate_active_agents() -> list[ActiveAgent]:
|
||||
"""All currently-running agents, across every available
|
||||
backend. Used by CLI `list active` and the dashboard's agents
|
||||
backend. Used by CLI `active` and the dashboard's agents
|
||||
pane so neither has to know which backends exist. Skips
|
||||
backends whose `is_available()` reports False.
|
||||
|
||||
@@ -871,7 +835,6 @@ def enumerate_active_agents() -> list[ActiveAgent]:
|
||||
|
||||
__all__ = [
|
||||
"ActiveAgent",
|
||||
"BackendStatus",
|
||||
"Bottle",
|
||||
"BottleBackend",
|
||||
"BottleCleanupPlan",
|
||||
@@ -884,7 +847,5 @@ __all__ = [
|
||||
"get_bottle_backend",
|
||||
"get_freezer",
|
||||
"has_backend",
|
||||
"is_backend_available",
|
||||
"is_backend_ready",
|
||||
"known_backend_names",
|
||||
]
|
||||
|
||||
@@ -20,8 +20,7 @@ infrastructure: CA install and git copy-in.
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import io
|
||||
from contextlib import contextmanager, redirect_stderr
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Generator, Sequence
|
||||
|
||||
@@ -61,11 +60,8 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
|
||||
return _setup.setup()
|
||||
|
||||
@classmethod
|
||||
def status(cls, *, quiet: bool = False) -> int:
|
||||
def status(cls) -> int:
|
||||
from . import setup as _setup
|
||||
if quiet:
|
||||
with redirect_stderr(io.StringIO()):
|
||||
return _setup.status()
|
||||
return _setup.status()
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Active-agent enumeration for the docker backend.
|
||||
|
||||
Returns `ActiveAgent` records the CLI `list active` command and the
|
||||
Returns `ActiveAgent` records the CLI `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 `list active` and the dashboard's agents pane —
|
||||
by the CLI's `active` and the dashboard's agents pane —
|
||||
one subprocess per refresh tick, not one per bottle."""
|
||||
try:
|
||||
r = subprocess.run(
|
||||
|
||||
@@ -8,8 +8,7 @@ fail-closed nftables egress boundary. Selected by
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
from contextlib import contextmanager, redirect_stderr
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Generator, Sequence
|
||||
|
||||
@@ -53,11 +52,8 @@ class FirecrackerBottleBackend(
|
||||
return _setup.setup()
|
||||
|
||||
@classmethod
|
||||
def status(cls, *, quiet: bool = False) -> int:
|
||||
def status(cls) -> int:
|
||||
from . import setup as _setup
|
||||
if quiet:
|
||||
with redirect_stderr(io.StringIO()):
|
||||
return _setup.status()
|
||||
return _setup.status()
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -13,7 +13,6 @@ generic `./cli.py backend {setup,status}` command dispatches to.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
@@ -23,9 +22,6 @@ 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
|
||||
@@ -223,90 +219,14 @@ 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 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.
|
||||
# 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.
|
||||
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,8 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
from contextlib import contextmanager, redirect_stderr
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Generator, Sequence
|
||||
|
||||
@@ -44,11 +43,8 @@ class MacosContainerBottleBackend(
|
||||
return _setup.setup()
|
||||
|
||||
@classmethod
|
||||
def status(cls, *, quiet: bool = False) -> int:
|
||||
def status(cls) -> int:
|
||||
from . import setup as _setup
|
||||
if quiet:
|
||||
with redirect_stderr(io.StringIO()):
|
||||
return _setup.status()
|
||||
return _setup.status()
|
||||
|
||||
@classmethod
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Main CLI dispatcher.
|
||||
|
||||
Commands: backend, cleanup, commit, edit, info, init, list, resume, start, supervise
|
||||
Commands: active, backend, cleanup, commit, edit, 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 or active containers\n")
|
||||
sys.stderr.write(" list list available agents from bot-bottle.json\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 "
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""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
|
||||
@@ -27,7 +27,7 @@ def cmd_commit(argv: list[str]) -> int:
|
||||
nargs="?",
|
||||
default=None,
|
||||
help=(
|
||||
"bottle slug from `cli.py list active` "
|
||||
"bottle slug from `cli.py active` "
|
||||
"(omit to pick interactively)"
|
||||
),
|
||||
)
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
"""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
|
||||
+7
-49
@@ -1,62 +1,20 @@
|
||||
"""list: list available agents or active bottles."""
|
||||
"""list: list available agents."""
|
||||
|
||||
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:
|
||||
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)
|
||||
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")
|
||||
return 0
|
||||
|
||||
# `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}")
|
||||
manifest = ManifestIndex.resolve(USER_CWD)
|
||||
for name in manifest.all_agent_names:
|
||||
print(name)
|
||||
return 0
|
||||
|
||||
@@ -7,6 +7,7 @@ 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. |
|
||||
|
||||
+3
-13
@@ -1,23 +1,13 @@
|
||||
# CI
|
||||
|
||||
The test workflow lives at [`.gitea/workflows/test.yml`](../.gitea/workflows/test.yml).
|
||||
It runs the unit suite plus one integration job per backend
|
||||
(`integration-docker`, `integration-firecracker`) on:
|
||||
It runs `tests/run_tests.py` (full suite — unit + integration) on:
|
||||
|
||||
- every push to a branch with an open pull request, and
|
||||
- every push to `main`.
|
||||
|
||||
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.
|
||||
Integration tests need Docker on the runner; they skip cleanly via
|
||||
`tests/_docker.skip_unless_docker` when no daemon is reachable.
|
||||
|
||||
A small subset of integration tests skip when running specifically
|
||||
under Gitea Actions (`GITEA_ACTIONS=true`), because `act_runner` runs
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
# 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."
|
||||
+5
-20
@@ -2,15 +2,14 @@
|
||||
|
||||
Plain-Python test suite using stdlib `unittest`. No external
|
||||
dependencies. Unit tests run anywhere Python 3 is present; integration
|
||||
tests run through the backend named by `BOT_BOTTLE_BACKEND` (default
|
||||
`docker`) and skip cleanly when that backend isn't available on the host.
|
||||
tests need Docker and skip cleanly otherwise.
|
||||
|
||||
## Layout
|
||||
|
||||
```
|
||||
tests/
|
||||
fixtures.py # JSON manifest builders (shared)
|
||||
_backend.py # backend selection + skip guards (shared)
|
||||
_docker.py # docker-availability skip helper (shared)
|
||||
unit/
|
||||
test_egress.py
|
||||
test_egress_addon_core.py
|
||||
@@ -74,7 +73,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 a backend.
|
||||
`tests/integration/` for one that needs Docker.
|
||||
2. Filename: `test_<topic>.py`.
|
||||
3. Boilerplate:
|
||||
```python
|
||||
@@ -89,19 +88,5 @@ BOT_BOTTLE_RUN_CANARIES=1 python -m unittest discover -t . -s tests/canaries -v
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
```
|
||||
4. Skip guards live in `tests._backend` 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.
|
||||
4. For Docker-dependent tests, decorate the class with
|
||||
`@skip_unless_docker()` from `tests._docker`.
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
"""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",
|
||||
)
|
||||
@@ -0,0 +1,45 @@
|
||||
"""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)
|
||||
@@ -25,14 +25,14 @@ import os
|
||||
import subprocess
|
||||
import unittest
|
||||
|
||||
from tests._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
|
||||
_IMAGE = "bot-bottle-gateway-test:chunk1"
|
||||
_DOCKERFILE = "Dockerfile.gateway"
|
||||
|
||||
|
||||
@skip_unless_backend("docker")
|
||||
@skip_unless_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._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
# 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_backend("docker")
|
||||
@skip_unless_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._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
IMAGE = "busybox"
|
||||
|
||||
|
||||
@skip_unless_backend("docker")
|
||||
@skip_unless_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._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
# 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_backend("docker")
|
||||
@skip_unless_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._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
IMAGE = "busybox"
|
||||
|
||||
|
||||
@skip_unless_backend("docker")
|
||||
@skip_unless_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._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
IMAGE = "busybox"
|
||||
|
||||
|
||||
@skip_unless_backend("docker")
|
||||
@skip_unless_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
|
||||
|
||||
@@ -18,10 +18,10 @@ from bot_bottle.backend.docker.network import (
|
||||
network_create_internal,
|
||||
network_remove,
|
||||
)
|
||||
from tests._backend import skip_unless_backend
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
|
||||
@skip_unless_backend("docker")
|
||||
@skip_unless_docker()
|
||||
class TestOrphanCleanup(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.slug = f"cb-test-orphan-{os.getpid()}"
|
||||
|
||||
@@ -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._backend import skip_unless_selected_backend_available
|
||||
from tests._docker import skip_unless_docker_or_firecracker
|
||||
|
||||
|
||||
# Secrets planted in the bottle env as literals (agents substitute via
|
||||
@@ -67,7 +67,7 @@ _DUMMY_HOST_KEY = (
|
||||
)
|
||||
|
||||
|
||||
@skip_unless_selected_backend_available()
|
||||
@skip_unless_docker_or_firecracker()
|
||||
@unittest.skipIf(
|
||||
os.environ.get("GITEA_ACTIONS") == "true"
|
||||
and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
|
||||
|
||||
@@ -385,60 +385,6 @@ 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;
|
||||
|
||||
@@ -328,201 +328,5 @@ 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()
|
||||
|
||||
@@ -1,78 +0,0 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,35 @@
|
||||
"""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()
|
||||
@@ -132,56 +132,30 @@ 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", 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):
|
||||
patch.object(fc_setup.shutil, "which", return_value=None):
|
||||
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, "_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):
|
||||
patch.object(fc_setup.shutil, "which", return_value=None):
|
||||
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, "_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):
|
||||
patch.object(fc_setup.shutil, "which", return_value=None):
|
||||
rc, out = self._run()
|
||||
self.assertEqual(1, rc)
|
||||
self.assertIn("CLASHES", out)
|
||||
|
||||
Reference in New Issue
Block a user