didericis-claude
5eb27cd9a8
Mirrors the fix already applied to the macos-container backend in
eb3e64e: bind-mount the parent egress directory instead of the
routes file itself, so the live routes update is visible inside the
running sidecar bundle when the host overwrites the file.
474 lines
18 KiB
Python
474 lines
18 KiB
Python
"""Unit: compose-spec renderer (PRD 0018 chunk 1).
|
||
|
||
Pure-function tests for `bottle_plan_to_compose`. Fixtures build a
|
||
fully-resolved DockerBottlePlan in memory; the renderer just
|
||
translates it to the compose dict. Conditional-service matrix is
|
||
covered via parameterized cases (git on/off × egress on/off ×
|
||
supervise on/off).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import subprocess
|
||
import unittest
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from unittest import mock
|
||
|
||
from bot_bottle.agent_provider import AgentProvisionPlan
|
||
from bot_bottle.backend import BottleSpec
|
||
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
|
||
from bot_bottle.backend.docker.compose import (
|
||
COMPOSE_PROJECT_PREFIX,
|
||
bottle_plan_to_compose,
|
||
compose_project_name,
|
||
list_active_slugs,
|
||
list_compose_projects,
|
||
slug_from_compose_project,
|
||
)
|
||
from bot_bottle.egress import (
|
||
EgressPlan,
|
||
EgressRoute,
|
||
)
|
||
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
|
||
from bot_bottle.manifest import ManifestIndex
|
||
from bot_bottle.supervise import SupervisePlan
|
||
|
||
|
||
SLUG = "demo-abc12"
|
||
STAGE = Path("/tmp/cb-stage")
|
||
STATE = Path("/tmp/cb-state")
|
||
|
||
|
||
def _manifest(*, supervise: bool, with_git: bool, with_egress: bool) -> ManifestIndex:
|
||
"""Minimal manifest with the toggles the chunk-1 matrix needs.
|
||
The renderer only reads from the plan, not the manifest, so this
|
||
is just here to back BottleSpec."""
|
||
bottle: dict[str, object] = {}
|
||
if supervise:
|
||
bottle["supervise"] = True
|
||
if with_git:
|
||
bottle["git-gate"] = {"repos": {
|
||
"upstream": {
|
||
"url": "ssh://git@example.com:22/x/y.git",
|
||
"key": {"provider": "static", "path": "/etc/hostname"},
|
||
},
|
||
}}
|
||
if with_egress:
|
||
bottle["egress"] = {
|
||
"routes": [{
|
||
"host": "api.example",
|
||
"auth": {"scheme": "Bearer", "token_ref": "TOK"},
|
||
}],
|
||
}
|
||
return ManifestIndex.from_json_obj({
|
||
"bottles": {"dev": bottle},
|
||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||
})
|
||
|
||
|
||
|
||
def _git_gate_plan(upstreams: tuple[GitGateUpstream, ...] = ()) -> GitGatePlan:
|
||
return GitGatePlan(
|
||
slug=SLUG,
|
||
entrypoint_script=STATE / "git-gate" / "entrypoint.sh",
|
||
hook_script=STATE / "git-gate" / "pre-receive",
|
||
access_hook_script=STATE / "git-gate" / "access-hook",
|
||
upstreams=upstreams,
|
||
internal_network=f"bot-bottle-net-{SLUG}",
|
||
egress_network=f"bot-bottle-egress-{SLUG}",
|
||
)
|
||
|
||
|
||
def _egress_plan(routes: tuple[EgressRoute, ...] = ()) -> EgressPlan:
|
||
token_env_map = {
|
||
r.token_env: r.token_ref
|
||
for r in routes
|
||
if r.token_env
|
||
}
|
||
return EgressPlan(
|
||
slug=SLUG,
|
||
routes_path=STATE / "egress" / "routes.yaml",
|
||
routes=routes,
|
||
token_env_map=token_env_map,
|
||
internal_network=f"bot-bottle-net-{SLUG}",
|
||
egress_network=f"bot-bottle-egress-{SLUG}",
|
||
mitmproxy_ca_host_path=STATE / "egress-ca" / "mitmproxy-ca.pem",
|
||
mitmproxy_ca_cert_only_host_path=STATE / "egress-ca" / "ca.pem",
|
||
)
|
||
|
||
|
||
def _supervise_plan() -> SupervisePlan:
|
||
return SupervisePlan(
|
||
slug=SLUG,
|
||
queue_dir=STATE / "supervise" / "queue",
|
||
current_config_dir=STATE / "supervise" / "current-config",
|
||
internal_network=f"bot-bottle-net-{SLUG}",
|
||
)
|
||
|
||
|
||
def _plan(
|
||
*,
|
||
with_git: bool = False,
|
||
with_egress: bool = False,
|
||
supervise: bool = False,
|
||
) -> DockerBottlePlan:
|
||
"""Build a fully-resolved DockerBottlePlan. Toggles cover the
|
||
matrix the renderer's conditional-service logic branches on."""
|
||
upstreams: tuple[GitGateUpstream, ...] = ()
|
||
if with_git:
|
||
upstreams = (GitGateUpstream(
|
||
name="upstream",
|
||
upstream_url="ssh://git@example.com:22/x/y.git",
|
||
upstream_host="example.com",
|
||
upstream_port="22",
|
||
identity_file="/etc/hostname",
|
||
known_host_key="",
|
||
known_hosts_file=STATE / "git-gate" / "upstream-known_hosts",
|
||
),)
|
||
routes: tuple[EgressRoute, ...] = ()
|
||
if with_egress:
|
||
routes = (EgressRoute(
|
||
host="api.example",
|
||
auth_scheme="Bearer",
|
||
token_env="EGRESS_TOKEN_0",
|
||
token_ref="TOK",
|
||
roles=(),
|
||
),)
|
||
|
||
index = _manifest(supervise=supervise, with_git=with_git, with_egress=with_egress)
|
||
spec = BottleSpec(
|
||
manifest=index,
|
||
agent_name="demo",
|
||
copy_cwd=False,
|
||
user_cwd="/tmp/x",
|
||
)
|
||
return DockerBottlePlan(
|
||
spec=spec,
|
||
manifest=index.load_for_agent("demo"),
|
||
stage_dir=STAGE,
|
||
slug=SLUG,
|
||
forwarded_env={"CLAUDE_CODE_OAUTH_TOKEN": "x"},
|
||
git_gate_plan=_git_gate_plan(upstreams),
|
||
egress_plan=_egress_plan(routes),
|
||
supervise_plan=_supervise_plan() if supervise else None,
|
||
use_runsc=False,
|
||
agent_provision=AgentProvisionPlan(
|
||
template="claude",
|
||
command="claude",
|
||
prompt_mode="append_file",
|
||
image="bot-bottle-claude:latest",
|
||
dockerfile="",
|
||
guest_home="/home/node",
|
||
instance_name=f"bot-bottle-{SLUG}",
|
||
prompt_file=STAGE / "prompt",
|
||
guest_env={},
|
||
),
|
||
)
|
||
|
||
|
||
class TestProjectAndNetworks(unittest.TestCase):
|
||
def test_project_name(self):
|
||
spec = bottle_plan_to_compose(_plan())
|
||
self.assertEqual(f"bot-bottle-{SLUG}", spec["name"])
|
||
|
||
def test_internal_network_is_internal(self):
|
||
spec = bottle_plan_to_compose(_plan())
|
||
net = spec["networks"]["internal"]
|
||
self.assertEqual(f"bot-bottle-net-{SLUG}", net["name"])
|
||
self.assertTrue(net["internal"])
|
||
|
||
def test_egress_network_is_external_bridge(self):
|
||
spec = bottle_plan_to_compose(_plan())
|
||
net = spec["networks"]["egress"]
|
||
self.assertEqual(f"bot-bottle-egress-{SLUG}", net["name"])
|
||
# No `internal:` key on the egress network — defaults to a
|
||
# normal user-defined bridge.
|
||
self.assertNotIn("internal", net)
|
||
|
||
|
||
class TestAgentAlwaysPresent(unittest.TestCase):
|
||
def test_agent_in_services(self):
|
||
s = bottle_plan_to_compose(_plan())["services"]
|
||
self.assertIn("agent", s)
|
||
|
||
def test_agent_command(self):
|
||
s = bottle_plan_to_compose(_plan())["services"]["agent"]
|
||
self.assertEqual(["sleep", "infinity"], s["command"])
|
||
|
||
def test_agent_image_uses_runtime_image(self):
|
||
plan = _plan()
|
||
s = bottle_plan_to_compose(plan)["services"]["agent"]
|
||
self.assertEqual(plan.image, s["image"])
|
||
|
||
def test_agent_only_on_internal_network(self):
|
||
s = bottle_plan_to_compose(_plan())["services"]["agent"]
|
||
self.assertEqual({"internal"}, set(s["networks"].keys()))
|
||
|
||
def test_agent_proxy_always_via_egress(self):
|
||
for with_egress in (False, True):
|
||
with self.subTest(with_egress=with_egress):
|
||
s = bottle_plan_to_compose(
|
||
_plan(with_egress=with_egress)
|
||
)["services"]["agent"]
|
||
proxy_lines = [e for e in s["environment"] if e.startswith("HTTPS_PROXY=")]
|
||
self.assertEqual(1, len(proxy_lines))
|
||
self.assertEqual("HTTPS_PROXY=http://egress:9099", proxy_lines[0])
|
||
|
||
def test_agent_proxy_via_egress_when_egress_present(self):
|
||
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["agent"]
|
||
proxy = [e for e in s["environment"] if e.startswith("HTTPS_PROXY=")][0]
|
||
self.assertEqual("HTTPS_PROXY=http://egress:9099", proxy)
|
||
|
||
def test_agent_no_proxy_adds_supervise_when_enabled(self):
|
||
s = bottle_plan_to_compose(
|
||
_plan(supervise=True)
|
||
)["services"]["agent"]
|
||
no_proxy = [e for e in s["environment"] if e.startswith("NO_PROXY=")][0]
|
||
self.assertIn("supervise", no_proxy)
|
||
|
||
def test_agent_forwarded_env_uses_bare_names(self):
|
||
# Bare NAME → compose inherits value from the up-process env,
|
||
# so secret token values stay out of the file.
|
||
s = bottle_plan_to_compose(_plan())["services"]["agent"]
|
||
self.assertIn("CLAUDE_CODE_OAUTH_TOKEN", s["environment"])
|
||
|
||
def test_agent_provider_env_uses_literal_values(self):
|
||
plan = _plan()
|
||
provision = AgentProvisionPlan(
|
||
template="codex",
|
||
command="codex",
|
||
prompt_mode="read_prompt_file",
|
||
image="bot-bottle-codex:latest",
|
||
dockerfile="",
|
||
guest_home="/home/node",
|
||
instance_name=f"bot-bottle-{SLUG}",
|
||
prompt_file=STAGE / "prompt",
|
||
guest_env={"CODEX_HOME": "/home/node/.codex"},
|
||
)
|
||
plan = type(plan)(**{**vars(plan), "agent_provision": provision}) # type: ignore
|
||
s = bottle_plan_to_compose(plan)["services"]["agent"]
|
||
self.assertIn("CODEX_HOME=/home/node/.codex", s["environment"])
|
||
|
||
def test_agent_runsc_runtime(self):
|
||
plan = _plan()
|
||
plan = type(plan)(**{**vars(plan), "use_runsc": True}) # type: ignore
|
||
s = bottle_plan_to_compose(plan)["services"]["agent"]
|
||
self.assertEqual("runsc", s["runtime"])
|
||
|
||
def test_agent_depends_only_on_sidecars(self):
|
||
# Bundle shape: the init supervisor owns intra-bundle daemon
|
||
# ordering, so the agent waits on the bundle container alone.
|
||
for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
|
||
with self.subTest(**kwargs):
|
||
s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
|
||
self.assertEqual(["sidecars"], s["depends_on"])
|
||
|
||
def test_agent_current_config_mount_only_with_supervise(self):
|
||
with_sv = bottle_plan_to_compose(_plan(supervise=True))["services"]["agent"]
|
||
self.assertTrue(any(
|
||
v["target"] == "/etc/bot-bottle/current-config"
|
||
for v in with_sv.get("volumes", [])
|
||
))
|
||
without_sv = bottle_plan_to_compose(_plan(supervise=False))["services"]["agent"]
|
||
# Either no volumes key at all, or no current-config target.
|
||
self.assertFalse(any(
|
||
v["target"] == "/etc/bot-bottle/current-config"
|
||
for v in without_sv.get("volumes", [])
|
||
))
|
||
|
||
|
||
class TestSidecarBundleShape(unittest.TestCase):
|
||
"""The compose renderer emits exactly one `sidecars` service in
|
||
place of the daemons it owns (egress + git-gate + supervise).
|
||
PRD 0024 chunk 5 dropped the legacy four-sidecar shape entirely,
|
||
so the bundle is the only thing exercised here."""
|
||
|
||
def _render(self, **plan_kwargs: object) -> Any: # type: ignore
|
||
return bottle_plan_to_compose(_plan(**plan_kwargs)) # type: ignore
|
||
|
||
def test_emits_two_services_minimal(self):
|
||
spec = self._render()
|
||
self.assertEqual({"sidecars", "agent"}, set(spec["services"].keys()))
|
||
|
||
def test_emits_two_services_full_matrix(self):
|
||
spec = self._render(with_git=True, with_egress=True, supervise=True)
|
||
# Still two services — the bundle absorbs git-gate/egress/supervise.
|
||
self.assertEqual({"sidecars", "agent"}, set(spec["services"].keys()))
|
||
|
||
def test_bundle_uses_bundle_image_and_dockerfile(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
|
||
self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
|
||
|
||
def test_bundle_container_name_uses_sidecars_prefix(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])
|
||
|
||
def test_bundle_joins_both_networks(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
self.assertEqual({"internal", "egress"}, set(sc["networks"].keys()))
|
||
|
||
def test_internal_aliases_include_egress_shortname(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
aliases = set(sc["networks"]["internal"]["aliases"])
|
||
self.assertIn("egress", aliases)
|
||
|
||
def test_internal_aliases_omit_inactive_sidecars(self):
|
||
# With no git-gate / supervise, those names are NOT aliased
|
||
# — keeps the alias list honest about what's actually
|
||
# listening inside the bundle.
|
||
sc = self._render()["services"]["sidecars"]
|
||
aliases = set(sc["networks"]["internal"]["aliases"])
|
||
self.assertNotIn("git-gate", aliases)
|
||
self.assertNotIn("supervise", aliases)
|
||
|
||
def test_internal_aliases_include_active_sidecars(self):
|
||
sc = self._render(with_git=True, supervise=True)["services"]["sidecars"]
|
||
aliases = set(sc["networks"]["internal"]["aliases"])
|
||
self.assertIn("git-gate", aliases)
|
||
self.assertIn("supervise", aliases)
|
||
|
||
def test_daemons_csv_lists_only_active(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
daemons = {
|
||
line.split("=", 1)[1]
|
||
for line in sc["environment"]
|
||
if line.startswith("BOT_BOTTLE_SIDECAR_DAEMONS=")
|
||
}
|
||
self.assertEqual({"egress"}, daemons)
|
||
|
||
def test_daemons_csv_expands_with_optional_sidecars(self):
|
||
sc = self._render(with_git=True, supervise=True)["services"]["sidecars"]
|
||
for line in sc["environment"]:
|
||
if line.startswith("BOT_BOTTLE_SIDECAR_DAEMONS="):
|
||
csv = line.split("=", 1)[1]
|
||
break
|
||
else:
|
||
self.fail("BOT_BOTTLE_SIDECAR_DAEMONS not in env")
|
||
self.assertEqual(
|
||
["egress", "git-gate", "supervise"],
|
||
csv.split(","),
|
||
)
|
||
|
||
def test_bundle_env_does_not_set_https_proxy(self):
|
||
# HTTPS_PROXY at the container level would route git-gate's
|
||
# git fetches through the proxy. Scoping it to mitmdump is
|
||
# the job of egress_entrypoint.sh; the bundle env must not
|
||
# leak it.
|
||
sc = self._render(with_egress=True)["services"]["sidecars"]
|
||
for line in sc["environment"]:
|
||
self.assertFalse(
|
||
line.startswith("HTTPS_PROXY=")
|
||
or line.startswith("HTTP_PROXY=")
|
||
or line.startswith("NO_PROXY="),
|
||
f"bundle env must not set {line!r}",
|
||
)
|
||
|
||
def test_egress_token_env_present_when_routes_declared(self):
|
||
sc = self._render(with_egress=True)["services"]["sidecars"]
|
||
env_strings = sc["environment"]
|
||
self.assertIn("EGRESS_TOKEN_0", env_strings)
|
||
|
||
def test_egress_token_env_omitted_when_no_routes(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
env_strings = sc["environment"]
|
||
self.assertNotIn("EGRESS_TOKEN_0", env_strings)
|
||
|
||
def test_supervise_env_present_when_active(self):
|
||
sc = self._render(supervise=True)["services"]["sidecars"]
|
||
env_strings = sc["environment"]
|
||
self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", env_strings)
|
||
self.assertTrue(any(e.startswith("SUPERVISE_QUEUE_DIR=") for e in env_strings))
|
||
self.assertTrue(any(e.startswith("SUPERVISE_PORT=") for e in env_strings))
|
||
|
||
def test_volumes_always_includes_egress_ca(self):
|
||
sc = self._render()["services"]["sidecars"]
|
||
targets = {v["target"] for v in sc["volumes"]}
|
||
self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
|
||
|
||
def test_volumes_union_full_matrix(self):
|
||
sc = self._render(with_git=True, with_egress=True, supervise=True)[
|
||
"services"]["sidecars"]
|
||
targets = {v["target"] for v in sc["volumes"]}
|
||
self.assertIn("/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", targets)
|
||
self.assertIn("/etc/egress", targets)
|
||
self.assertIn("/git-gate-entrypoint.sh", targets)
|
||
self.assertIn("/git-gate/creds/upstream-known_hosts", targets)
|
||
self.assertTrue(any("supervise/queue" in t or t.startswith("/run/supervise")
|
||
for t in targets))
|
||
|
||
def test_extra_hosts_omitted_for_git_upstreams(self):
|
||
sc = self._render(with_git=True)["services"]["sidecars"]
|
||
self.assertNotIn("extra_hosts", sc)
|
||
|
||
def test_agent_depends_on_bundle_only(self):
|
||
sc = self._render(with_git=True, with_egress=True, supervise=True)[
|
||
"services"]["agent"]
|
||
self.assertEqual(["sidecars"], sc["depends_on"])
|
||
|
||
def test_agent_proxy_url_resolves_via_bundle_alias(self):
|
||
# With egress active, the agent's HTTPS_PROXY points at
|
||
# `egress` shortname; bundle aliases `egress` to itself so
|
||
# the URL keeps working without an agent-side change.
|
||
spec = self._render(with_egress=True)
|
||
sc = spec["services"]["agent"]
|
||
proxy = next(e for e in sc["environment"] if e.startswith("HTTPS_PROXY="))
|
||
self.assertIn("egress", proxy)
|
||
self.assertIn("egress",
|
||
spec["services"]["sidecars"]["networks"]["internal"]["aliases"])
|
||
|
||
|
||
class TestProjectNaming(unittest.TestCase):
|
||
"""The slug ↔ compose-project mapping is the contract dashboard,
|
||
cleanup, and launch all rely on. Lock it down."""
|
||
|
||
def test_compose_project_name_is_prefix_plus_slug(self):
|
||
self.assertEqual(
|
||
f"{COMPOSE_PROJECT_PREFIX}myagent-abc12",
|
||
compose_project_name("myagent-abc12"),
|
||
)
|
||
|
||
def test_slug_from_compose_project_is_inverse(self):
|
||
self.assertEqual(
|
||
"myagent-abc12",
|
||
slug_from_compose_project(f"{COMPOSE_PROJECT_PREFIX}myagent-abc12"),
|
||
)
|
||
|
||
def test_slug_from_unrelated_project_returns_empty(self):
|
||
# Defends against `docker compose ls` including non-bottle
|
||
# projects on a host with other compose setups.
|
||
self.assertEqual("", slug_from_compose_project("other-project"))
|
||
|
||
|
||
class TestComposeProjectListing(unittest.TestCase):
|
||
def test_compose_ls_error_warns_by_default(self):
|
||
with (
|
||
mock.patch(
|
||
"bot_bottle.backend.docker.compose.subprocess.run",
|
||
return_value=subprocess.CompletedProcess(
|
||
args=["docker"], returncode=1, stdout="", stderr="no daemon",
|
||
),
|
||
),
|
||
mock.patch("bot_bottle.backend.docker.compose.warn") as warn,
|
||
):
|
||
self.assertEqual([], list_compose_projects())
|
||
warn.assert_called_once_with("docker compose ls failed: no daemon")
|
||
|
||
def test_compose_ls_error_can_be_quiet_for_dashboard_polling(self):
|
||
with (
|
||
mock.patch(
|
||
"bot_bottle.backend.docker.compose.subprocess.run",
|
||
return_value=subprocess.CompletedProcess(
|
||
args=["docker"], returncode=1, stdout="", stderr="no daemon",
|
||
),
|
||
),
|
||
mock.patch("bot_bottle.backend.docker.compose.warn") as warn,
|
||
):
|
||
self.assertEqual([], list_active_slugs(warn_on_error=False))
|
||
warn.assert_not_called()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|