Compare commits

..

19 Commits

Author SHA1 Message Date
didericis-claude a68ee778f1 firecracker status(): open /dev/kvm O_RDWR; check kernel, dropbear, mke2fs
test / integration-docker (push) Successful in 15s
test / unit (push) Successful in 40s
Update Quality Badges / update-badges (push) Successful in 44s
lint / lint (push) Successful in 58s
test / integration-firecracker (push) Successful in 5m5s
test / coverage (push) Successful in 49s
test / publish-infra (push) Successful in 2m4s
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 21:25:20 -04:00
didericis-claude de02d13ccf firecracker status(): add binary and KVM readiness checks
Adds `_firecracker_binary_ok()` (runs `firecracker --version`) and
`_kvm_accessible()` (opens /dev/kvm, issues KVM_GET_API_VERSION ioctl)
and gates `status()` on both, so the launch preflight reports missing
binary or inaccessible KVM before the caller ever attempts a boot.

Regression tests in TestFirecrackerBinaryCheck, TestFirecrackerKvmCheck,
and TestFirecrackerStatusRuntime cover all branches. Updated the
pre-existing TestFirecrackerStatus fixture to also stub the two new
helpers so it still exercises only the tap-pool/nft logic it was
designed for.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-24 21:25:20 -04:00
didericis-claude c740d1e145 fix: add missing type annotations and remove unused variables in new tests
Pyright flagged three inner class `status` methods missing a type
annotation on the `quiet` parameter, and two unused variables
(`written`, `real_status`) left over from an earlier draft of
test_docker_quiet_true_suppresses_stderr.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-24 21:25:20 -04:00
didericis-claude 177721c286 test: cover is_backend_available, is_backend_ready, and status(quiet=True)
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 21:25:20 -04:00
didericis-claude 5bcf3db1f8 fix: align skip-guard tests with is_backend_ready and drop unused import
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 21:25:20 -04:00
didericis-claude dcd658ece1 feat(backend): BackendStatus enum, quiet status(), is_backend_ready()
Add a module-level BackendStatus(IntEnum) with READY=0 to replace magic
0 comparisons. Extend BottleBackend.status() with a quiet parameter:
quiet=False (default) prints diagnostics; quiet=True returns the code
silently for programmatic checks.

Add is_backend_available() (cheap PATH check, alias for has_backend) and
is_backend_ready(name, *, quiet=False) (full status() check) to the
backend package for callers that need to distinguish availability from
readiness.

Update tests/_backend.py guards to use is_backend_ready(quiet=False) so
diagnostic output is printed during test discovery, giving the operator a
concrete reason for each skip rather than a bare "prerequisites
unavailable" message.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-24 21:25:20 -04:00
didericis-claude dfb56f8fe9 ci: reuse backend.has_backend / backend status for readiness
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 21:25:20 -04:00
didericis-claude 64adf23775 ci: backend-agnostic integration guards + per-backend preflight
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-24 21:25:20 -04:00
didericis-claude 1972c8c6e9 docs: add glossary of canonical bot-bottle terminology
lint / lint (push) Successful in 56s
tracker-policy-pr / check-pr (pull_request) Successful in 15s
Defines Agent Provider, Agent Runtime, Agent/Agent Definition,
Bottle/Bottle Definition, Sealed Bottle, Bottled Agent, and Active Bottle.
Links from docs/README.md and AGENTS.md for discoverability.

Closes #474

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-24 00:12:32 -04:00
didericis-claude 5d109ea290 CLI cleanup: remove info, rename list subcommands (#472)
test / unit (push) Successful in 44s
test / integration-docker (push) Successful in 47s
lint / lint (push) Successful in 54s
Update Quality Badges / update-badges (push) Successful in 55s
test / integration-firecracker (push) Successful in 4m47s
test / coverage (push) Successful in 15s
test / publish-infra (push) Successful in 1m49s
- Remove `info` command
- Rename `list active` → `active` (new top-level command)
- Rename `list available` → `list` (no subcommand argument)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-24 00:02:19 -04:00
didericis-codex f24ae45d13 test(gateway): cover single and simultaneous daemon crashes
test / integration-docker (push) Successful in 21s
lint / lint (push) Successful in 1m4s
test / unit (push) Successful in 1m44s
test / integration-firecracker (push) Successful in 4m55s
test / coverage (push) Successful in 21s
test / publish-infra (push) Successful in 1m50s
Update Quality Badges / update-badges (push) Failing after 14m42s
2026-07-23 20:15:18 -04:00
didericis-codex 594d07410a fix(gateway): restart dead children before completion check 2026-07-23 20:15:18 -04:00
didericis-claude c9c62f256d fix(egress): restart dead daemons and cap inbound scan body size (#455)
Two complementary fixes for the egress gateway OOM:

1. gateway_init: auto-restart any daemon that dies unexpectedly. The
   supervisor already had restart_daemon()/request_restart() logic; this
   wires it into tick() so an OOM-killed mitmdump is respawned without
   operator intervention. Implements the "eventual" failure policy
   described in the original module docstring.

2. egress_addon: cap response body bytes passed to the DLP inbound scan
   at EGRESS_INBOUND_SCAN_LIMIT_BYTES (default 1 MiB). mitmproxy buffers
   the full response before the hook fires; capping at scan time limits
   the additional amplification from decoded text and regex strings. Bodies
   over the limit emit an egress_scan_truncated log event. Set to 0 to
   disable the cap.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-23 20:15:18 -04:00
didericis 10150ae9f5 docs(paths): correct root docstring after SQLite queue migration
lint / lint (push) Successful in 2m37s
test / unit (push) Successful in 1m42s
test / integration-firecracker (push) Successful in 4m43s
test / integration-docker (push) Successful in 33s
test / coverage (push) Successful in 19s
Update Quality Badges / update-badges (push) Successful in 1m45s
test / publish-infra (push) Successful in 2m1s
The module docstring still listed "queue" and "audit logs" as things
living under the app data root, which read as directories. Both have
been tables in the shared SQLite DB since PRD 0067; the legacy `queue/`
directory was unplumbed in 29904609 and nothing has written there since.

Lists what the root actually holds and points at the stores that own the
queue/audit rows, so the next reader doesn't go looking for a directory
that isn't there.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-23 19:40:20 -04:00
Quality Badge Bot 86c7ac1843 chore: update quality badges
- Coverage: 84%
- Core coverage: 94%

[skip ci]
2026-07-23 23:29:52 +00:00
didericis-claude 2e0414f969 fix(test): restore headless unit tests broken by PTY check in e719022
test / integration-docker (push) Successful in 15s
test / unit (push) Successful in 39s
Update Quality Badges / update-badges (push) Successful in 45s
lint / lint (push) Successful in 2m36s
test / integration-firecracker (push) Successful in 5m15s
test / coverage (push) Successful in 16s
test / publish-infra (push) Successful in 1m39s
The PTY check added in e719022 calls sys.stdin.fileno() which raises
io.UnsupportedOperation under pytest's stdin capture. Guard the fileno()
call with a try/except so the check degrades cleanly to "not a tty"
instead of crashing, then stub os.isatty in the test setUp so headless
unit tests aren't blocked on a real TTY.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-23 23:28:55 +00:00
didericis-claude 12b071833d test: annotate _git_repo helper to satisfy pyright strict mode
test / integration-docker (push) Successful in 13s
Update Quality Badges / update-badges (push) Successful in 40s
lint / lint (push) Successful in 53s
test / integration-firecracker (push) Successful in 4m56s
test / unit (push) Failing after 33s
test / coverage (push) Has been skipped
test / publish-infra (push) Has been skipped
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-23 18:32:59 -04:00
didericis-claude bf72282f8e fix(manifest): avoid KeyError when override bottle has repos absent from base
merge_bottles_runtime used .get(n, base_repos_by_name[n]) which eagerly
evaluates the default, crashing when a repo exists only in the override
bottle. Replaced with a conditional expression so the base lookup only
runs when needed.

Adds three regression tests covering override-only, base-only, and
name-collision cases.

Closes #457

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-23 18:32:59 -04:00
Quality Badge Bot f2c3710d0d chore: update quality badges
- Coverage: 83%
- Core coverage: 94%

[skip ci]
2026-07-23 22:13:28 +00:00
37 changed files with 807 additions and 244 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,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).
+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 `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,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 `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.)
@@ -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 list active` walks past
doesn't fail when `cli.py 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 `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.
@@ -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 `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(
+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: 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 "
+53
View File
@@ -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
+1 -1
View File
@@ -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)"
),
)
-49
View File
@@ -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
View File
@@ -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
+6 -1
View File
@@ -14,6 +14,7 @@ the private orchestrator `_launch_bottle`.
from __future__ import annotations
import argparse
import io
import os
import shutil
import sys
@@ -195,7 +196,11 @@ def _start_headless(
path, so the agent still execs on the inherited stdio/PTY — an
orchestrator allocates that PTY and relays it to its
desktop/mobile clients."""
if not os.isatty(sys.stdin.fileno()):
try:
stdin_fd = sys.stdin.fileno()
except io.UnsupportedOperation:
stdin_fd = -1
if not os.isatty(stdin_fd):
die(
"--headless requires a PTY on stdin; run via:\n"
" script -q /dev/null ./cli.py start ..."
+1 -1
View File
@@ -57,7 +57,7 @@ def _merge_two_bottles_runtime(base: "ManifestBottle", override: "ManifestBottle
n for n in override_repos_by_name if n not in base_repos_by_name
]
merged_git = tuple(
override_repos_by_name.get(n, base_repos_by_name[n])
override_repos_by_name[n] if n in override_repos_by_name else base_repos_by_name[n]
for n in merged_repos_names
)
+9 -3
View File
@@ -1,8 +1,14 @@
"""Foundational filesystem paths for bot-bottle.
`bot_bottle_root()` is the app data root state, queue, audit logs,
git-gate keys, and the shared DB all live under it. It defaults to
`~/.bot-bottle` and is overridable with the **`BOT_BOTTLE_ROOT`** env var.
`bot_bottle_root()` is the app data root per-bottle state, git-gate
keys, the gateway CA, and the shared SQLite DB all live under it. It
defaults to `~/.bot-bottle` and is overridable with the
**`BOT_BOTTLE_ROOT`** env var.
Note that the supervise queue and the audit log are *tables in the shared
DB*, not directories under the root see `queue_store.py` / `audit_store.py`.
The root held a `queue/` directory before the SQLite migration (PRD 0067);
nothing writes there now.
The env override is the single knob for redirecting the root: the test
suite points it at a throwaway dir instead of monkey-patching the function
+1
View File
@@ -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. |
+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
@@ -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."
+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()
+3
View File
@@ -69,6 +69,9 @@ class TestCmdStartHeadless(unittest.TestCase):
self._modal = patch.object(tui_mod, "name_color_modal").start()
patch.dict(os.environ, {}, clear=False).start()
os.environ.pop("BOT_BOTTLE_BACKEND", None)
# PTY check uses os.isatty(sys.stdin.fileno()); stub both so
# headless unit tests aren't blocked on a real TTY.
patch("bot_bottle.cli.start.os.isatty", return_value=True).start()
self.addCleanup(patch.stopall)
def _spec(self):
-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)
+25
View File
@@ -25,6 +25,10 @@ def _bottle(**kwargs: object) -> ManifestBottle:
return ManifestBottle.from_dict("test", kwargs)
def _git_repo(url: str) -> dict[str, object]:
return {"url": url, "key": {"provider": "gitea", "forge_token_env": "TOK"}}
class TestMergeBottlesRuntime(unittest.TestCase):
def test_single_bottle_returns_as_is(self):
b = _bottle(env={"FOO": "1"})
@@ -115,6 +119,27 @@ class TestMergeBottlesRuntime(unittest.TestCase):
self.assertEqual("2", result.env["B"])
self.assertEqual("3", result.env["C"])
def test_git_repo_only_in_override_does_not_raise(self):
# Regression for issue #457: override bottle declares a repo that the
# base doesn't have → KeyError on base_repos_by_name[n].
base = _bottle(env={"X": "base"})
override = _bottle(**{"git-gate": {"repos": {"myrepo": _git_repo("ssh://git@example.com/repo.git")}}})
result = merge_bottles_runtime([base, override])
self.assertIn("myrepo", [e.Name for e in result.git])
def test_git_repo_only_in_base_survives_override(self):
base = _bottle(**{"git-gate": {"repos": {"myrepo": _git_repo("ssh://git@example.com/repo.git")}}})
override = _bottle(env={"X": "override"})
result = merge_bottles_runtime([base, override])
self.assertIn("myrepo", [e.Name for e in result.git])
def test_git_repo_override_wins_by_name(self):
base = _bottle(**{"git-gate": {"repos": {"myrepo": _git_repo("ssh://git@base.example.com/repo.git")}}})
override = _bottle(**{"git-gate": {"repos": {"myrepo": _git_repo("ssh://git@override.example.com/repo.git")}}})
result = merge_bottles_runtime([base, override])
self.assertEqual(1, len(result.git))
self.assertEqual("ssh://git@override.example.com/repo.git", result.git[0].Upstream)
def test_empty_list_raises(self):
with self.assertRaises(ValueError):
merge_bottles_runtime([])