didericis/bot-bottle
1
0
Fork 0
Code Issues 7 Pull Requests 15 Actions Packages Projects Releases Wiki Activity
Files
cefdc8c6e9800f99f455aa0b1a6540471383ff23
bot-bottle/tests/unit/test_compose.py
T
didericis cefdc8c6e9
test / unit (pull_request) Successful in 18s
Details
test / integration (pull_request) Successful in 1m5s
Details
feat(launch): switch start to docker compose project per bottle 
PRD 0018 chunk 3. Each instance is now one `docker compose` project:

  - launch.py renders the compose spec via chunk-1's
    bottle_plan_to_compose, writes it to state/<slug>/docker-compose.yml,
    `docker compose up -d`s, and (on teardown) dumps
    `docker compose logs --no-color --timestamps` to
    state/<slug>/compose.log before `docker compose down`.
  - Networks are pre-created (`docker network create --internal` +
    user-defined bridge) so pipelock yaml can know the internal CIDR
    before compose-up. Compose references them with `external: true`;
    the launch step's ExitStack still owns network removal.
  - Agent still runs `sleep infinity`; claude reaches it via
    `docker exec -it` exactly like before (per the PRD's resolved
    TTY question).
  - metadata.json grows a `compose_project` field so dashboard /
    cleanup tooling can derive compose invocations without
    re-deriving the slug.

Security follow-ups from chunk-2 review:

  (b) CA private keys: pipelock + egress ca-key.pem land at 0o600
      explicitly. The mitmproxy cert+key concat stays 0o644 because
      the egress container's uid-1000 user reads it through the
      bind mount; parent dir at 0o700 still restricts host-side
      reach.
  (c) Apply atomicity: egress_apply + pipelock_apply switch from
      `docker cp` to host-side write-temp-then-rename on the
      bind-mount source. POSIX rename is atomic on the same
      filesystem, so a sidecar SIGHUP racing the apply can't see
      a half-written routes.yaml / pipelock.yaml.

Per-sidecar Docker{Sidecar}.start/stop methods stay in place — the
integration test suite drives them directly to validate each image
in isolation, which is still useful. launch.py no longer calls
them; a follow-up chunk can prune if the integration tests move to
the compose lifecycle.

git-gate entrypoint's chmod 600 on the keyfile + known_hosts now
tolerates EROFS (`|| true`) — the host SSH key is already 0600
(SSH refuses to load otherwise), so the inside-container chmod
was already a no-op in the docker-cp path and now just needs to
not error on the read-only bind mount.

422 unit tests pass; supervise integration test passes; end-to-end
`./cli.py start implementer` brings up the project, attaches,
captures full merged logs on teardown, and reaps all containers +
networks.
2026-05-25 23:16:40 -04:00

456 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 unittest
from pathlib import Path
from claude_bottle.backend import BottleSpec
from claude_bottle.backend.docker.bottle_plan import DockerBottlePlan
from claude_bottle.backend.docker.compose import bottle_plan_to_compose
from claude_bottle.egress import (
EgressPlan,
EgressRoute,
)
from claude_bottle.git_gate import GitGatePlan, GitGateUpstream
from claude_bottle.manifest import Manifest
from claude_bottle.pipelock import PipelockProxyPlan
from claude_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) -> Manifest:
"""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 = {}
if supervise:
bottle["supervise"] = True
if with_git:
bottle["git"] = [{
"Name": "upstream",
"Upstream": "ssh://git@example.com:22/x/y.git",
"IdentityFile": "/etc/hostname", # any existing file
}]
if with_egress:
bottle["egress"] = {
"routes": [{
"host": "api.example",
"auth": {"scheme": "Bearer", "token_ref": "TOK"},
}],
}
return Manifest.from_json_obj({
"bottles": {"dev": bottle},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
def _spec(*, supervise: bool, with_git: bool, with_egress: bool) -> BottleSpec:
return BottleSpec(
manifest=_manifest(
supervise=supervise, with_git=with_git, with_egress=with_egress,
),
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
)
def _proxy_plan() -> PipelockProxyPlan:
return PipelockProxyPlan(
yaml_path=STATE / "pipelock.yaml",
slug=SLUG,
internal_network=f"claude-bottle-net-{SLUG}",
internal_network_cidr="10.1.2.0/24",
egress_network=f"claude-bottle-egress-{SLUG}",
ca_cert_host_path=STATE / "pipelock-ca" / "ca.pem",
ca_key_host_path=STATE / "pipelock-ca" / "ca-key.pem",
)
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"claude-bottle-net-{SLUG}",
egress_network=f"claude-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"claude-bottle-net-{SLUG}",
egress_network=f"claude-bottle-egress-{SLUG}",
mitmproxy_ca_host_path=STATE / "egress-ca" / "mitmproxy-ca.pem",
mitmproxy_ca_cert_only_host_path=STATE / "egress-ca" / "ca.pem",
pipelock_ca_host_path=STATE / "pipelock-ca" / "ca.pem",
pipelock_proxy_url=f"http://claude-bottle-pipelock-{SLUG}:8888",
)
def _supervise_plan() -> SupervisePlan:
return SupervisePlan(
slug=SLUG,
queue_dir=STATE / "supervise" / "queue",
current_config_dir=STATE / "supervise" / "current-config",
internal_network=f"claude-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="",
extra_hosts={"example.com": "10.0.0.1"},
),)
routes: tuple[EgressRoute, ...] = ()
if with_egress:
routes = (EgressRoute(
host="api.example",
auth_scheme="Bearer",
token_env="EGRESS_TOKEN_0",
token_ref="TOK",
path_allowlist=(),
roles=(),
),)
return DockerBottlePlan(
spec=_spec(supervise=supervise, with_git=with_git, with_egress=with_egress),
stage_dir=STAGE,
slug=SLUG,
container_name=f"claude-bottle-{SLUG}",
container_name_pinned=False,
image="claude-bottle:latest",
derived_image="",
runtime_image="claude-bottle:latest",
dockerfile_path="",
env_file=Path("/dev/null"), # exists, size 0 → renderer skips env_file
forwarded_env={"CLAUDE_CODE_OAUTH_TOKEN": "x"},
prompt_file=STAGE / "prompt",
proxy_plan=_proxy_plan(),
git_gate_plan=_git_gate_plan(upstreams),
egress_plan=_egress_plan(routes),
supervise_plan=_supervise_plan() if supervise else None,
use_runsc=False,
)
class TestProjectAndNetworks(unittest.TestCase):
def test_project_name(self):
spec = bottle_plan_to_compose(_plan())
self.assertEqual(f"claude-bottle-{SLUG}", spec["name"])
def test_internal_network_marked_external(self):
# Chunk 3 pre-creates networks with `docker network create
# --internal` so pipelock can know the CIDR before compose-up.
# Compose references the network by name with `external: true`.
spec = bottle_plan_to_compose(_plan())
net = spec["networks"]["internal"]
self.assertEqual(f"claude-bottle-net-{SLUG}", net["name"])
self.assertTrue(net["external"])
def test_egress_network_marked_external(self):
spec = bottle_plan_to_compose(_plan())
net = spec["networks"]["egress"]
self.assertEqual(f"claude-bottle-egress-{SLUG}", net["name"])
self.assertTrue(net["external"])
class TestPipelockAlwaysPresent(unittest.TestCase):
"""Pipelock is unconditional — every bottle has the SSRF guard +
body scanner sitting on its upstream leg."""
def test_minimal_plan_has_pipelock(self):
spec = bottle_plan_to_compose(_plan())
self.assertIn("pipelock", spec["services"])
def test_pipelock_pinned_image_no_build(self):
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
self.assertTrue(s["image"].startswith("ghcr.io/luckypipewrench/pipelock"))
self.assertNotIn("build", s)
def test_pipelock_container_name(self):
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
self.assertEqual(f"claude-bottle-pipelock-{SLUG}", s["container_name"])
def test_pipelock_on_both_networks(self):
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
self.assertIn("internal", s["networks"])
self.assertIn("egress", s["networks"])
def test_pipelock_long_name_alias_on_internal(self):
# Backward compat: anything still dialing pipelock by
# `claude-bottle-pipelock-<slug>` resolves on the internal
# network.
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
aliases = s["networks"]["internal"]["aliases"]
self.assertIn(f"claude-bottle-pipelock-{SLUG}", aliases)
def test_pipelock_bind_mounts(self):
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
targets = {v["target"] for v in s["volumes"]}
self.assertEqual(
{"/etc/pipelock.yaml", "/etc/pipelock-ca.pem", "/etc/pipelock-ca-key.pem"},
targets,
)
for v in s["volumes"]:
self.assertEqual("bind", v["type"])
self.assertTrue(v["read_only"])
def test_pipelock_command(self):
s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
self.assertEqual(
["run", "--config", "/etc/pipelock.yaml", "--listen", "0.0.0.0:8888"],
s["command"],
)
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.runtime_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_via_pipelock_when_no_egress(self):
s = bottle_plan_to_compose(_plan(with_egress=False))["services"]["agent"]
env = s["environment"]
# Looking for HTTPS_PROXY pointing at pipelock's container name.
proxy_lines = [e for e in env if e.startswith("HTTPS_PROXY=")]
self.assertEqual(1, len(proxy_lines))
self.assertEqual(
f"HTTPS_PROXY=http://claude-bottle-pipelock-{SLUG}:8888",
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_runsc_runtime(self):
plan = _plan()
plan = type(plan)(**{**vars(plan), "use_runsc": True})
s = bottle_plan_to_compose(plan)["services"]["agent"]
self.assertEqual("runsc", s["runtime"])
def test_agent_depends_on_pipelock(self):
s = bottle_plan_to_compose(_plan())["services"]["agent"]
self.assertIn("pipelock", s["depends_on"])
def test_agent_depends_on_every_present_sidecar(self):
s = bottle_plan_to_compose(
_plan(with_git=True, with_egress=True, supervise=True)
)["services"]["agent"]
self.assertEqual(
{"pipelock", "git-gate", "egress", "supervise"},
set(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/claude-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/claude-bottle/current-config"
for v in without_sv.get("volumes", [])
))
class TestConditionalGitGate(unittest.TestCase):
def test_absent_when_no_upstreams(self):
s = bottle_plan_to_compose(_plan(with_git=False))["services"]
self.assertNotIn("git-gate", s)
def test_present_when_upstreams(self):
s = bottle_plan_to_compose(_plan(with_git=True))["services"]
self.assertIn("git-gate", s)
def test_git_gate_built_from_dockerfile(self):
s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
self.assertEqual("Dockerfile.git-gate", s["build"]["dockerfile"])
self.assertEqual("claude-bottle-git-gate:latest", s["image"])
def test_git_gate_extra_hosts(self):
s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
self.assertIn("example.com:10.0.0.1", s["extra_hosts"])
def test_git_gate_identity_file_bind_mount(self):
s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
# Per-upstream identity file is mounted at /git-gate/creds/<name>-key.
self.assertTrue(any(
v["target"] == "/git-gate/creds/upstream-key"
for v in s["volumes"]
))
class TestConditionalEgress(unittest.TestCase):
def test_absent_when_no_routes(self):
s = bottle_plan_to_compose(_plan(with_egress=False))["services"]
self.assertNotIn("egress", s)
def test_present_when_routes(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]
self.assertIn("egress", s)
def test_egress_alias_on_internal(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
self.assertIn("egress", s["networks"]["internal"]["aliases"])
def test_egress_upstream_envs(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
env = s["environment"]
self.assertIn(
f"EGRESS_UPSTREAM_PROXY=http://claude-bottle-pipelock-{SLUG}:8888",
env,
)
self.assertIn(
"EGRESS_UPSTREAM_CA=/home/mitmproxy/.mitmproxy/pipelock-ca.pem",
env,
)
def test_egress_token_slot_bare_name(self):
# Bare NAME entry in environment list → value inherits from
# compose process env, never lands in the rendered file.
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
self.assertIn("EGRESS_TOKEN_0", s["environment"])
def test_egress_depends_on_pipelock(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
self.assertIn("pipelock", s["depends_on"])
def test_egress_bind_mounts(self):
s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
targets = {v["target"] for v in s["volumes"]}
self.assertEqual(
{
"/etc/egress/routes.yaml",
"/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem",
"/home/mitmproxy/.mitmproxy/pipelock-ca.pem",
},
targets,
)
class TestConditionalSupervise(unittest.TestCase):
def test_absent_when_off(self):
s = bottle_plan_to_compose(_plan(supervise=False))["services"]
self.assertNotIn("supervise", s)
def test_present_when_on(self):
s = bottle_plan_to_compose(_plan(supervise=True))["services"]
self.assertIn("supervise", s)
def test_supervise_internal_only(self):
s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
self.assertEqual({"internal"}, set(s["networks"].keys()))
def test_supervise_alias_on_internal(self):
s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
self.assertIn("supervise", s["networks"]["internal"]["aliases"])
def test_supervise_queue_dir_mounted_rw(self):
s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
queue_mount = [v for v in s["volumes"] if v["target"] == "/run/supervise/queue"]
self.assertEqual(1, len(queue_mount))
self.assertFalse(queue_mount[0]["read_only"])
def test_supervise_env_vars(self):
s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", s["environment"])
class TestFullMatrix(unittest.TestCase):
"""The eight combinations of git/egress/supervise toggles. Just
asserts which services appear — content correctness is covered
per-service above."""
def test_matrix(self):
cases: list[tuple[bool, bool, bool, set[str]]] = []
for g in (False, True):
for e in (False, True):
for sv in (False, True):
expected = {"pipelock", "agent"}
if g:
expected.add("git-gate")
if e:
expected.add("egress")
if sv:
expected.add("supervise")
cases.append((g, e, sv, expected))
for g, e, sv, expected in cases:
with self.subTest(git=g, egress=e, supervise=sv):
s = bottle_plan_to_compose(
_plan(with_git=g, with_egress=e, supervise=sv)
)["services"]
self.assertEqual(expected, set(s.keys()))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink