"""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(*, 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 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, ) -> DockerBottlePlan: """Build a fully-resolved DockerBottlePlan. Toggles cover the matrix the renderer's conditional-service logic branches on. Every bottle is supervised (issue #249), so the supervise plan is always present.""" 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(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(), 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_includes_supervise(self): s = bottle_plan_to_compose(_plan())["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}]: with self.subTest(**kwargs): s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"] self.assertEqual(["sidecars"], s["depends_on"]) def test_agent_current_config_always_mounted(self): # Every bottle is supervised (issue #249), so the read-only # current-config mount is always present in the agent. agent = bottle_plan_to_compose(_plan())["services"]["agent"] self.assertTrue(any( v["target"] == "/etc/bot-bottle/current-config" for v in agent.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) # 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, that name is NOT aliased — keeps the alias # list honest about what's actually listening inside the bundle. # supervise is always present (issue #249). sc = self._render()["services"]["sidecars"] aliases = set(sc["networks"]["internal"]["aliases"]) self.assertNotIn("git-gate", aliases) self.assertIn("supervise", aliases) def test_internal_aliases_include_active_sidecars(self): sc = self._render(with_git=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=") } # egress + supervise are always present (issue #249). self.assertEqual({"egress,supervise"}, daemons) def test_daemons_csv_expands_with_optional_sidecars(self): sc = self._render(with_git=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", "supervise", "git-gate"], 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()["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)[ "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)[ "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()