didericis/bot-bottle
1
0
Fork 0
Code Issues 7 Pull Requests 15 Actions Packages Projects Releases Wiki Activity

feat(agent_provider): migrate tests, drop guest-home/skills-dir env knobs, activate PRD 0050

Browse Source 
- tests/unit/test_provision_apply.py covers the new shared
  apply helpers (apply_skills / apply_prompt / apply_provision)
  that replace the per-backend modules deleted in the prior
  commit.
- tests/unit/test_contrib_supervise_mcp.py covers both providers'
  provision_supervise_mcp behavior — confirms the codex bottle
  now runs `codex mcp add` symmetrically with claude.
- tests/unit/test_smolmachines_provision.py drops the four test
  classes whose subjects moved (TestProvisionPrompt /
  TestProvisionProviderAuth / TestProvisionSkills /
  TestProvisionSupervise); the backend-side CA / git / workspace
  classes stay.
- tests/unit/test_docker_provision_provider_auth.py removed; its
  coverage now lives in tests/unit/test_provision_apply.py
  (apply_provision is backend-agnostic, one test file suffices).

Drops the BOT_BOTTLE_CONTAINER_HOME, BOT_BOTTLE_GUEST_HOME,
BOT_BOTTLE_CONTAINER_SKILLS_DIR, and BOT_BOTTLE_GUEST_SKILLS_DIR
env knobs the deleted provision modules used to read. /home/node
is hardcoded everywhere the knobs lived; the values were
effectively constants today and removing them keeps the PRD-0050
surface area honest.

Flips PRD 0050 Status: Draft → Active. Closes #177 on merge.
This commit is contained in:
didericis (claude)
2026-06-03 21:27:42 +00:00
committed by didericis
parent 8c45016aa2
commit 970c5066d7
9 changed files with 447 additions and 495 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot_bottle/backend/docker/prepare.py
+1 -1
View File
@@ -63,7 +63,7 @@ def resolve_plan(
bottle = manifest.bottle_for(spec.agent_name)
provider = bottle.agent_provider
provider_runtime = runtime_for(provider.template)
guest_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
guest_home = "/home/node"
workspace_plan = resolve_workspace_plan(spec, guest_home=guest_home)
# PRD 0016 follow-up: identity, not bare slug. A fresh `start`
bot_bottle/backend/docker/provision/git.py
+1 -3
View File
@@ -18,7 +18,6 @@ Three concerns, all about git in the agent:
from __future__ import annotations
import os
import shlex
from ....git_gate import GIT_GATE_HOSTNAME, git_gate_render_gitconfig
@@ -58,8 +57,7 @@ def _provision_git_gate_config(plan: DockerBottlePlan, bottle: Bottle) -> None:
manifest_bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if not manifest_bottle.git:
return
container_home = os.environ.get("BOT_BOTTLE_CONTAINER_HOME", "/home/node")
container_gitconfig = f"{container_home}/.gitconfig"
container_gitconfig = "/home/node/.gitconfig"
content = git_gate_render_gitconfig(manifest_bottle.git, GIT_GATE_HOSTNAME)
config_file = plan.stage_dir / "agent_gitconfig"
bot_bottle/backend/smolmachines/prepare.py
+1 -1
View File
@@ -61,7 +61,7 @@ def resolve_plan(
bottle = manifest.bottle_for(spec.agent_name)
provider = bottle.agent_provider
provider_runtime = runtime_for(provider.template)
guest_home = os.environ.get("BOT_BOTTLE_GUEST_HOME", "/home/node")
guest_home = "/home/node"
workspace_plan = resolve_workspace_plan(spec, guest_home=guest_home)
slug = spec.identity or bottle_identity(spec.agent_name)
bot_bottle/backend/smolmachines/provision/git.py
+3 -6
View File
@@ -36,15 +36,12 @@ from ... import Bottle
from ..bottle_plan import SmolmachinesBottlePlan
# `node` is the agent user from the repo Dockerfile. Override via
# BOT_BOTTLE_GUEST_HOME mirrors the docker backend's
# BOT_BOTTLE_CONTAINER_HOME knob — same purpose, different
# transport.
_DEFAULT_GUEST_HOME = "/home/node"
# `node` is the agent user from the repo Dockerfile.
_GUEST_HOME = "/home/node"
def _guest_home() -> str:
return os.environ.get("BOT_BOTTLE_GUEST_HOME", _DEFAULT_GUEST_HOME)
return _GUEST_HOME
def provision_git(plan: SmolmachinesBottlePlan, bottle: Bottle) -> None:
docs/prds/0050-agent-provider-contrib.md
+18 -20
View File
@@ -1,6 +1,6 @@
# PRD 0050: Move provider-specific agent logic into contrib
- **Status:** Draft
- **Status:** Active
- **Author:** claude
- **Created:** 2026-06-03
- **Issue:** #177
@@ -363,26 +363,24 @@ Each chunk is one commit on the PR; the PR ships as one cut.
6. **Activate.** Flip Status: Draft → Active in this PRD; close
#177 on merge.
## Open questions
## Open questions (resolved)
1. **Does Codex's `claude mcp add` equivalent exist as a CLI verb,
or is direct TOML editing the only option?** If a `codex mcp
add` (or similar) is in the binary, Option A in the MCP section
becomes more attractive. The implementation chunk for Codex MCP
should check `codex --help` against the version pinned in
`Dockerfile.codex` before committing to TOML editing.
2. **Shared apply helper vs. default methods on `AgentProvider`?**
Both work. Default methods read cleaner at call sites; a free
function in `contrib/_provision_apply.py` is easier to test in
isolation. Defer to the implementer; not load-bearing for the
PRD.
3. **Should `BOT_BOTTLE_CONTAINER_HOME` / `BOT_BOTTLE_GUEST_HOME`
/`BOT_BOTTLE_CONTAINER_SKILLS_DIR` env knobs survive the move?**
They're only read by the backend-side provision modules being
deleted. The provider plugin can take them too, but the values
are effectively constants today. Leaving the env-read fall-back
in place for compat is one line per knob; this PRD says yes,
carry them over.
1. **`codex mcp add` exists.** Implementation calls
`codex mcp add --transport http supervise <url>` as `node` —
symmetric with `claude mcp add` (no `--scope user`; Codex writes
`~/.codex/config.toml` by default). Failure logs a warning; the
bottle still works without the entry.
2. **Default methods on `AgentProvider`.** The base ABC's
`provision_skills` / `provision_prompt` / `provision` delegate
to a small `bot_bottle/_provision_apply.py` helper. Concrete
subclasses don't override the defaults today; the helper exists
so a future provider that legitimately needs a different layout
can stay declarative.
3. **Env knobs removed.** `BOT_BOTTLE_CONTAINER_HOME`,
`BOT_BOTTLE_GUEST_HOME`, `BOT_BOTTLE_CONTAINER_SKILLS_DIR`, and
`BOT_BOTTLE_GUEST_SKILLS_DIR` are gone; `/home/node` is hardcoded
everywhere it was read. The values were effectively constants;
the knobs added surface area for no real flexibility.
## References
tests/unit/test_contrib_supervise_mcp.py
+163
View File
@@ -0,0 +1,163 @@
"""Unit: contrib supervise MCP registration (PRD 0050).
Each provider plugin's `provision_supervise_mcp` runs the
provider's own CLI (`claude mcp add` / `codex mcp add`) inside the
agent guest to register the per-bottle supervise sidecar in the
provider's user config. The previous claude-only `provision_supervise`
modules under backend/{docker,smolmachines}/provision/supervise.py
covered this behavior pre-PRD-0050."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock
from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.contrib.claude.agent_provider import ClaudeAgentProvider
from bot_bottle.contrib.codex.agent_provider import CodexAgentProvider
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
bottle = MagicMock(spec=Bottle)
bottle.name = "bot-bottle-demo-abc12"
bottle.exec.return_value = (
exec_result if exec_result is not None
else ExecResult(returncode=0, stdout="", stderr="")
)
return bottle
def _plan(*, supervise: bool, template: str = "claude") -> DockerBottlePlan:
manifest = Manifest.from_json_obj({
"bottles": {
"dev": {"agent_provider": {"template": template},
**({"supervise": True} if supervise else {})},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
spec = BottleSpec(
manifest=manifest, agent_name="demo",
copy_cwd=False, user_cwd="/tmp/x",
)
supervise_plan = None
if supervise:
supervise_plan = SupervisePlan(
slug="demo-abc12",
queue_dir=Path("/tmp/queue"),
current_config_dir=Path("/tmp/current-config"),
)
return DockerBottlePlan(
spec=spec,
stage_dir=Path("/tmp/stage"),
slug="demo-abc12",
container_name="bot-bottle-demo-abc12",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
runtime_image="bot-bottle-claude:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"), slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
hook_script=Path("/tmp/git-gate-hook"),
access_hook_script=Path("/tmp/git-gate-access-hook"),
upstreams=(),
),
egress_plan=EgressPlan(
slug="demo-abc12",
routes_path=Path("/tmp/routes.yaml"),
routes=(),
token_env_map={},
),
supervise_plan=supervise_plan,
use_runsc=False,
agent_provision=AgentProvisionPlan(
template=template, command=template,
prompt_mode="append_file" if template == "claude" else "read_prompt_file",
image="", dockerfile="", guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
_URL = "http://supervise:9100/"
class TestClaudeSuperviseMcp(unittest.TestCase):
def test_noop_when_supervise_disabled(self):
bottle = _make_bottle()
ClaudeAgentProvider().provision_supervise_mcp(
_plan(supervise=False), bottle, _URL,
)
bottle.exec.assert_not_called()
def test_runs_claude_mcp_add_as_node(self):
bottle = _make_bottle()
ClaudeAgentProvider().provision_supervise_mcp(
_plan(supervise=True), bottle, _URL,
)
bottle.exec.assert_called_once()
script = bottle.exec.call_args.args[0]
self.assertEqual("node", bottle.exec.call_args.kwargs.get("user"))
self.assertIn("claude mcp add", script)
self.assertIn("--scope user", script)
self.assertIn("--transport http", script)
self.assertIn("supervise", script)
self.assertIn(_URL, script)
def test_logs_warning_on_failure_but_does_not_raise(self):
bottle = _make_bottle(
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
)
ClaudeAgentProvider().provision_supervise_mcp(
_plan(supervise=True), bottle, _URL,
)
class TestCodexSuperviseMcp(unittest.TestCase):
def test_noop_when_supervise_disabled(self):
bottle = _make_bottle()
CodexAgentProvider().provision_supervise_mcp(
_plan(supervise=False, template="codex"), bottle, _URL,
)
bottle.exec.assert_not_called()
def test_runs_codex_mcp_add_as_node(self):
bottle = _make_bottle()
CodexAgentProvider().provision_supervise_mcp(
_plan(supervise=True, template="codex"), bottle, _URL,
)
bottle.exec.assert_called_once()
script = bottle.exec.call_args.args[0]
self.assertEqual("node", bottle.exec.call_args.kwargs.get("user"))
self.assertIn("codex mcp add", script)
self.assertIn("--transport http", script)
self.assertIn("supervise", script)
self.assertIn(_URL, script)
def test_logs_warning_on_failure_but_does_not_raise(self):
bottle = _make_bottle(
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
)
CodexAgentProvider().provision_supervise_mcp(
_plan(supervise=True, template="codex"), bottle, _URL,
)
if __name__ == "__main__":
unittest.main()
tests/unit/test_docker_provision_provider_auth.py
-173
View File
@@ -1,173 +0,0 @@
"""Unit: docker provider auth marker provisioning."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock
from bot_bottle.agent_provider import (
AgentProvisionDir,
AgentProvisionFile,
AgentProvisionPlan,
)
from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.docker.provision import provider_auth as _provider_auth
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.workspace import workspace_plan
def _plan(
*,
codex_auth_file: Path | None = None,
agent_provider_template: str = "codex",
) -> DockerBottlePlan:
manifest = Manifest.from_json_obj({
"bottles": {"dev": {"agent_provider": {"template": "codex"}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
spec = BottleSpec(
manifest=manifest,
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
)
return DockerBottlePlan(
spec=spec,
stage_dir=Path("/tmp/stage"),
slug="demo-abc12",
container_name="bot-bottle-demo-abc12",
container_name_pinned=False,
image="bot-bottle-codex:latest",
derived_image="",
runtime_image="bot-bottle-codex:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"),
slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
hook_script=Path("/tmp/git-gate-hook"),
access_hook_script=Path("/tmp/git-gate-access-hook"),
upstreams=(),
),
egress_plan=EgressPlan(
slug="demo-abc12",
routes_path=Path("/tmp/routes.yaml"),
routes=(),
token_env_map={},
),
supervise_plan=None,
use_runsc=False,
agent_provision=_agent_provision(
agent_provider_template, codex_auth_file=codex_auth_file,
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
def _agent_provision(
template: str, *, codex_auth_file: Path | None = None,
) -> AgentProvisionPlan:
if template != "codex":
return AgentProvisionPlan(
template=template,
command=template,
prompt_mode="append_file",
image="",
dockerfile="",
guest_env={},
)
files = [
AgentProvisionFile(
Path("/tmp/codex-config.toml"),
"/home/node/.codex/config.toml",
),
]
if codex_auth_file is not None:
files.append(AgentProvisionFile(
codex_auth_file,
"/home/node/.codex/auth.json",
))
return AgentProvisionPlan(
template="codex",
command="codex",
prompt_mode="read_prompt_file",
image="bot-bottle-codex:latest",
dockerfile="",
guest_env={},
dirs=(AgentProvisionDir("/home/node/.codex"),),
files=tuple(files),
)
def _make_bottle(name: str = "bot-bottle-demo-abc12") -> MagicMock:
bottle = MagicMock(spec=Bottle)
bottle.name = name
bottle.exec.return_value = ExecResult(returncode=0, stdout="", stderr="")
return bottle
class TestProvisionProviderAuth(unittest.TestCase):
def test_noop_for_non_codex_provider(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(
_plan(agent_provider_template="claude"), bottle,
)
self.assertEqual(0, bottle.cp_in.call_count)
self.assertEqual(0, bottle.exec.call_count)
def test_codex_provider_trusts_launch_dir_without_auth_file(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(_plan(), bottle)
scripts = [c.args[0] for c in bottle.exec.call_args_list]
self.assertTrue(
any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts)
)
cp_calls = [c.args for c in bottle.cp_in.call_args_list]
self.assertIn(
("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
cp_calls,
)
self.assertTrue(
any("chown" in s and "/home/node/.codex/config.toml" in s for s in scripts)
)
self.assertTrue(
any("chmod" in s and "/home/node/.codex/config.toml" in s for s in scripts)
)
def test_copies_dummy_auth_json_to_codex_home(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(
_plan(codex_auth_file=Path("/tmp/codex-auth.json")),
bottle,
)
cp_calls = [c.args for c in bottle.cp_in.call_args_list]
self.assertIn(
("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
cp_calls,
)
self.assertIn(
("/tmp/codex-auth.json", "/home/node/.codex/auth.json"),
cp_calls,
)
scripts = [c.args[0] for c in bottle.exec.call_args_list]
self.assertTrue(
any("chown" in s and "/home/node/.codex/auth.json" in s for s in scripts)
)
self.assertTrue(
any("chmod" in s and "/home/node/.codex/auth.json" in s for s in scripts)
)
if __name__ == "__main__":
unittest.main()
tests/unit/test_provision_apply.py
+260
View File
@@ -0,0 +1,260 @@
"""Unit: shared provision-apply helpers (PRD 0050).
Covers `bot_bottle._provision_apply.apply_skills` /
`apply_prompt` / `apply_provision` — the backend-agnostic helpers
that AgentProvider's default `provision_skills` / `provision_prompt`
/ `provision` dispatch through. The same suite covered the
docker / smolmachines `provision/{skills,prompt,provider_auth}.py`
modules before they were deleted."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle import _provision_apply
from bot_bottle._provision_apply import (
PROMPT_PATH,
SKILLS_DIR,
apply_prompt,
apply_provision,
apply_skills,
)
from bot_bottle.agent_provider import (
AgentProvisionCommand,
AgentProvisionDir,
AgentProvisionFile,
AgentProvisionPlan,
)
from bot_bottle.backend import Bottle, BottleSpec, ExecResult
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.pipelock import PipelockProxyPlan
from bot_bottle.workspace import workspace_plan
def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
bottle = MagicMock(spec=Bottle)
bottle.name = "bot-bottle-demo-abc12"
bottle.exec.return_value = (
exec_result if exec_result is not None
else ExecResult(returncode=0, stdout="", stderr="")
)
return bottle
def _exec_scripts(bottle: MagicMock) -> list[str]:
return [c.args[0] for c in bottle.exec.call_args_list]
def _plan(
*,
agent_prompt: str = "",
skills: list[str] | None = None,
agent_provision: AgentProvisionPlan | None = None,
) -> DockerBottlePlan:
manifest = Manifest.from_json_obj({
"bottles": {"dev": {}},
"agents": {
"demo": {
"skills": list(skills or []),
"prompt": agent_prompt,
"bottle": "dev",
},
},
})
spec = BottleSpec(
manifest=manifest,
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
)
return DockerBottlePlan(
spec=spec,
stage_dir=Path("/tmp/stage"),
slug="demo-abc12",
container_name="bot-bottle-demo-abc12",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
runtime_image="bot-bottle-claude:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
proxy_plan=PipelockProxyPlan(
yaml_path=Path("/tmp/pipelock.yaml"),
slug="demo-abc12",
),
git_gate_plan=GitGatePlan(
slug="demo-abc12",
entrypoint_script=Path("/tmp/git-gate-entrypoint.sh"),
hook_script=Path("/tmp/git-gate-hook"),
access_hook_script=Path("/tmp/git-gate-access-hook"),
upstreams=(),
),
egress_plan=EgressPlan(
slug="demo-abc12",
routes_path=Path("/tmp/routes.yaml"),
routes=(),
token_env_map={},
),
supervise_plan=None,
use_runsc=False,
agent_provision=agent_provision or AgentProvisionPlan(
template="claude",
command="claude",
prompt_mode="append_file",
image="",
dockerfile="",
guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
class TestApplyPrompt(unittest.TestCase):
def test_cp_uses_bottle_cp_in(self):
bottle = _make_bottle()
apply_prompt(_plan(), bottle)
bottle.cp_in.assert_called_once_with(
"/tmp/state/demo-abc12/agent/prompt.txt",
PROMPT_PATH,
)
def test_returns_path_when_agent_has_prompt(self):
bottle = _make_bottle()
r = apply_prompt(_plan(agent_prompt="You are a helpful assistant."), bottle)
self.assertEqual(PROMPT_PATH, r)
def test_returns_none_when_agent_has_no_prompt(self):
bottle = _make_bottle()
r = apply_prompt(_plan(agent_prompt=""), bottle)
self.assertIsNone(r)
bottle.cp_in.assert_called_once()
def test_chowns_to_node_after_copy(self):
bottle = _make_bottle()
apply_prompt(_plan(), bottle)
scripts = _exec_scripts(bottle)
self.assertTrue(any("chown node:node" in s and PROMPT_PATH in s for s in scripts))
self.assertTrue(any("chmod 600" in s and PROMPT_PATH in s for s in scripts))
class TestApplySkills(unittest.TestCase):
def test_noop_when_agent_has_no_skills(self):
bottle = _make_bottle()
apply_skills(_plan(skills=[]), bottle)
bottle.cp_in.assert_not_called()
bottle.exec.assert_not_called()
def test_mkdir_plus_cp_per_skill(self):
bottle = _make_bottle()
with patch.object(
_provision_apply, "host_skill_dir", create=True,
side_effect=lambda n: f"/host/skills/{n}",
) if False else patch(
"bot_bottle.backend.util.host_skill_dir",
side_effect=lambda n: f"/host/skills/{n}",
), patch("bot_bottle._provision_apply.os.path.isdir", return_value=True):
apply_skills(_plan(skills=["init-prd", "verify"]), bottle)
scripts = _exec_scripts(bottle)
self.assertTrue(any("mkdir -p" in s and SKILLS_DIR in s for s in scripts))
cp_targets = {c.args[1] for c in bottle.cp_in.call_args_list}
self.assertEqual({
f"{SKILLS_DIR}/init-prd/",
f"{SKILLS_DIR}/verify/",
}, cp_targets)
self.assertEqual(
2,
sum(1 for s in scripts if "chown -R node:node" in s),
)
def test_missing_skill_dies(self):
bottle = _make_bottle()
with patch(
"bot_bottle.backend.util.host_skill_dir",
side_effect=lambda n: f"/host/skills/{n}",
), patch("bot_bottle._provision_apply.os.path.isdir", return_value=False):
with self.assertRaises(SystemExit):
apply_skills(_plan(skills=["init-prd"]), bottle)
class TestApplyProvision(unittest.TestCase):
"""The `dirs` / `pre_copy` / `files` / `verify` apply loop that
used to live in `provision_provider_auth`."""
def test_noop_on_empty_provision_plan(self):
bottle = _make_bottle()
apply_provision(_plan(), bottle)
bottle.cp_in.assert_not_called()
bottle.exec.assert_not_called()
def test_codex_provision_creates_dir_and_copies_config(self):
provision = AgentProvisionPlan(
template="codex",
command="codex",
prompt_mode="read_prompt_file",
image="bot-bottle-codex:latest",
dockerfile="",
guest_env={},
dirs=(AgentProvisionDir("/home/node/.codex"),),
files=(AgentProvisionFile(
Path("/tmp/codex-config.toml"),
"/home/node/.codex/config.toml",
),),
)
bottle = _make_bottle()
apply_provision(_plan(agent_provision=provision), bottle)
bottle.cp_in.assert_called_once_with(
"/tmp/codex-config.toml",
"/home/node/.codex/config.toml",
)
scripts = _exec_scripts(bottle)
self.assertTrue(any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts))
self.assertTrue(any("chown" in s and "/home/node/.codex/config.toml" in s for s in scripts))
self.assertTrue(any("chmod" in s and "/home/node/.codex/config.toml" in s for s in scripts))
def test_runs_pre_copy_then_verify(self):
provision = AgentProvisionPlan(
template="codex",
command="codex",
prompt_mode="read_prompt_file",
image="",
dockerfile="",
guest_env={},
pre_copy=(AgentProvisionCommand(
("find", "/home/node/.codex", "-name", "*.sqlite", "-delete"),
"could not reset runtime db files",
),),
verify=(AgentProvisionCommand(
("runuser", "-u", "node", "--", "codex", "login", "status"),
"codex rejected the dummy auth",
),),
)
bottle = _make_bottle()
apply_provision(_plan(agent_provision=provision), bottle)
scripts = _exec_scripts(bottle)
self.assertTrue(any("find" in s and "-delete" in s for s in scripts))
self.assertTrue(any("runuser" in s and "codex login status" in s for s in scripts))
def test_dies_when_dir_creation_fails(self):
provision = AgentProvisionPlan(
template="codex",
command="codex",
prompt_mode="read_prompt_file",
image="",
dockerfile="",
guest_env={},
dirs=(AgentProvisionDir("/home/node/.codex"),),
)
bottle = _make_bottle(exec_result=ExecResult(1, "", "mkdir: nope\n"))
with self.assertRaises(SystemExit):
apply_provision(_plan(agent_provision=provision), bottle)
if __name__ == "__main__":
unittest.main()
tests/unit/test_smolmachines_provision.py
-291
View File
@@ -26,10 +26,6 @@ from bot_bottle.backend.smolmachines.bottle_plan import (
from bot_bottle.backend.smolmachines.provision import (
ca as _ca,
git as _git,
prompt as _prompt,
provider_auth as _provider_auth,
skills as _skills,
supervise as _supervise,
workspace as _workspace,
)
from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
@@ -223,257 +219,6 @@ def _agent_provision(
)
class TestProvisionPrompt(unittest.TestCase):
def test_cp_uses_bottle_cp_in(self):
bottle = _make_bottle()
_prompt.provision_prompt(_plan(), bottle)
bottle.cp_in.assert_called_once_with(
"/tmp/state/demo-abc12/agent/prompt.txt",
"/home/node/.bot-bottle-prompt.txt",
)
def test_returns_path_when_agent_has_prompt(self):
bottle = _make_bottle()
r = _prompt.provision_prompt(
_plan(agent_prompt="You are a helpful assistant."),
bottle,
)
self.assertEqual("/home/node/.bot-bottle-prompt.txt", r)
def test_returns_none_when_agent_has_no_prompt(self):
# The file is still copied (path-must-exist contract);
# only the return value differs.
bottle = _make_bottle()
r = _prompt.provision_prompt(_plan(agent_prompt=""), bottle)
self.assertIsNone(r)
bottle.cp_in.assert_called_once()
def test_chowns_to_node_after_copy(self):
# cp_in lands as root; without the chown, the node user
# can't read its own mode-600 prompt.
bottle = _make_bottle()
_prompt.provision_prompt(_plan(), bottle)
scripts = _exec_scripts(bottle)
self.assertTrue(
any("chown node:node" in s and "/home/node/.bot-bottle-prompt.txt" in s
for s in scripts)
)
self.assertTrue(
any("chmod 600" in s and "/home/node/.bot-bottle-prompt.txt" in s
for s in scripts)
)
class TestProvisionProviderAuth(unittest.TestCase):
def test_noop_for_non_codex_provider(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(_plan(), bottle)
self.assertEqual(0, bottle.cp_in.call_count)
self.assertEqual(0, bottle.exec.call_count)
def test_codex_provider_trusts_launch_dir_without_auth_file(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(
_plan(agent_provider_template="codex"),
bottle,
)
bottle.cp_in.assert_called_once_with(
"/tmp/codex-config.toml",
"/home/node/.codex/config.toml",
)
scripts = _exec_scripts(bottle)
self.assertTrue(any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts))
self.assertTrue(
any("chown" in s and "node:node" in s and "/home/node/.codex/config.toml" in s
for s in scripts)
)
self.assertTrue(
any("chmod" in s and "600" in s and "/home/node/.codex/config.toml" in s
for s in scripts)
)
def test_copies_dummy_auth_json_to_codex_home(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(
_plan(
agent_provider_template="codex",
codex_auth_file=Path("/tmp/codex-auth.json"),
),
bottle,
)
cp_calls = [c.args for c in bottle.cp_in.call_args_list]
self.assertIn(
("/tmp/codex-config.toml", "/home/node/.codex/config.toml"),
cp_calls,
)
self.assertIn(
("/tmp/codex-auth.json", "/home/node/.codex/auth.json"),
cp_calls,
)
scripts = _exec_scripts(bottle)
self.assertTrue(any("mkdir -p" in s and "/home/node/.codex" in s for s in scripts))
self.assertTrue(
any("chown" in s and "node:node" in s and s.rstrip().endswith("/home/node/.codex")
for s in scripts)
)
self.assertTrue(
any("chmod" in s and "700" in s and s.rstrip().endswith("/home/node/.codex")
for s in scripts)
)
# The pre_copy `find ... -delete` script should be present
# (shlex.join properly quotes the `(`/`)`/`*.sqlite`).
self.assertTrue(
any("find" in s and "-delete" in s and "*.sqlite" in s for s in scripts)
)
self.assertTrue(
any("chown" in s and "/home/node/.codex/auth.json" in s for s in scripts)
)
self.assertTrue(
any("chmod" in s and "600" in s and "/home/node/.codex/auth.json" in s
for s in scripts)
)
# Verify command runs `codex login status` via runuser node.
self.assertTrue(
any("runuser" in s and "codex login status" in s for s in scripts)
)
def test_honors_codex_home_from_guest_env(self):
bottle = _make_bottle()
_provider_auth.provision_provider_auth(
_plan(
agent_provider_template="codex",
codex_auth_file=Path("/tmp/codex-auth.json"),
guest_env={"CODEX_HOME": "/run/codex-home"},
),
bottle,
)
cp_calls = [c.args for c in bottle.cp_in.call_args_list]
self.assertIn(
("/tmp/codex-config.toml", "/run/codex-home/config.toml"),
cp_calls,
)
self.assertIn(
("/tmp/codex-auth.json", "/run/codex-home/auth.json"),
cp_calls,
)
scripts = _exec_scripts(bottle)
self.assertTrue(
any("runuser" in s and "CODEX_HOME=/run/codex-home" in s and "codex login status" in s
for s in scripts)
)
def test_dies_when_codex_home_cannot_be_created(self):
bottle = _make_bottle(
exec_result=ExecResult(1, "", "mkdir: nope\n"),
)
with self.assertRaises(SystemExit):
_provider_auth.provision_provider_auth(
_plan(
agent_provider_template="codex",
codex_auth_file=Path("/tmp/codex-auth.json"),
),
bottle,
)
self.assertEqual(0, bottle.cp_in.call_count)
self.assertEqual(1, bottle.exec.call_count)
def test_dies_when_codex_rejects_dummy_auth(self):
# CODEX_HOME setup ok, but codex login status fails (last exec).
bottle = _make_bottle()
bottle.exec.side_effect = [
ExecResult(0, "", ""), # mkdir CODEX_HOME
ExecResult(0, "", ""), # chown CODEX_HOME
ExecResult(0, "", ""), # chmod CODEX_HOME
ExecResult(0, "", ""), # find ... -delete (pre_copy)
ExecResult(0, "", ""), # chown config.toml
ExecResult(0, "", ""), # chmod config.toml
ExecResult(0, "", ""), # chown auth.json
ExecResult(0, "", ""), # chmod auth.json
ExecResult(1, "Not logged in\n", ""), # login status (verify)
]
with self.assertRaises(SystemExit):
_provider_auth.provision_provider_auth(
_plan(
agent_provider_template="codex",
codex_auth_file=Path("/tmp/codex-auth.json"),
),
bottle,
)
class TestProvisionSkills(unittest.TestCase):
def _patch_host_skill_dir(self, returns: dict[str, str]):
return patch(
"bot_bottle.backend.smolmachines.provision.skills.host_skill_dir",
side_effect=lambda n: returns.get(n, f"/nope/{n}"),
)
def test_no_op_when_agent_has_no_skills(self):
bottle = _make_bottle()
_skills.provision_skills(_plan(skills=[]), bottle)
self.assertEqual(0, bottle.cp_in.call_count)
self.assertEqual(0, bottle.exec.call_count)
def test_mkdir_plus_cp_per_skill(self):
bottle = _make_bottle()
with self._patch_host_skill_dir({
"init-prd": "/host/skills/init-prd",
"verify": "/host/skills/verify",
}), patch(
"bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
return_value=True,
):
_skills.provision_skills(
_plan(skills=["init-prd", "verify"]),
bottle,
)
# mkdir skills_dir once + (rm -rf + chown) per skill = 5 exec calls.
self.assertEqual(5, bottle.exec.call_count)
scripts = _exec_scripts(bottle)
self.assertTrue(
any("mkdir -p" in s and "/home/node/.claude/skills" in s for s in scripts)
)
# Two cp calls, one per skill, into the per-skill subdir.
self.assertEqual(2, bottle.cp_in.call_count)
cp_targets = {call.args[1] for call in bottle.cp_in.call_args_list}
self.assertEqual(
{
"/home/node/.claude/skills/init-prd",
"/home/node/.claude/skills/verify",
},
cp_targets,
)
# Each skill gets a chown -R node:node so claude can read it.
chown_scripts = [s for s in scripts if "chown -R node:node" in s]
self.assertEqual(2, len(chown_scripts))
def test_skills_dir_overridable_via_env(self):
import os
bottle = _make_bottle()
with self._patch_host_skill_dir({"init-prd": "/host/skills/init-prd"}), \
patch(
"bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
return_value=True,
), \
patch.dict(os.environ, {"BOT_BOTTLE_GUEST_SKILLS_DIR": "/home/node/.claude/skills"}):
_skills.provision_skills(_plan(skills=["init-prd"]), bottle)
self.assertEqual(
"/home/node/.claude/skills/init-prd",
bottle.cp_in.call_args.args[1],
)
def test_missing_skill_dies(self):
bottle = _make_bottle()
with self._patch_host_skill_dir({"init-prd": "/host/skills/init-prd"}), \
patch(
"bot_bottle.backend.smolmachines.provision.skills.os.path.isdir",
return_value=False,
):
with self.assertRaises(SystemExit):
_skills.provision_skills(_plan(skills=["init-prd"]), bottle)
def _write_self_signed_cert(path: Path) -> None:
"""Drop a real self-signed PEM at `path` so provision_ca's
fingerprint computation (PEM_cert_to_DER_cert + sha256) has
@@ -774,41 +519,5 @@ class TestProvisionWorkspace(unittest.TestCase):
)
class TestProvisionSupervise(unittest.TestCase):
def test_noop_when_supervise_not_enabled(self):
bottle = _make_bottle()
_supervise.provision_supervise(_plan(), bottle)
bottle.exec.assert_not_called()
def test_calls_claude_mcp_add_when_supervise_enabled(self):
plan = _plan(
supervise=True,
agent_supervise_url="http://127.0.0.1:9100/",
)
bottle = _make_bottle()
_supervise.provision_supervise(plan, bottle)
bottle.exec.assert_called_once()
script = bottle.exec.call_args.args[0]
user = bottle.exec.call_args.kwargs.get("user")
self.assertEqual("node", user)
# SmolmachinesBottle.exec(user="node") handles uid switch +
# HOME setup automatically — the script itself is just the
# claude command.
self.assertIn("claude mcp add", script)
self.assertIn("--scope user", script)
self.assertIn("--transport http", script)
self.assertIn("supervise", script)
self.assertIn("http://127.0.0.1:9100/", script)
def test_non_zero_exit_logs_warning_but_does_not_raise(self):
plan = _plan(supervise=True)
bottle = _make_bottle(
exec_result=ExecResult(returncode=1, stdout="", stderr="boom"),
)
# No raise — the bottle still works without the MCP
# entry, so we log and move on.
_supervise.provision_supervise(plan, bottle)
if __name__ == "__main__":
unittest.main()