"""Unit: smolmachines provisioning helpers (PRD 0023 chunks 4a + 4d). Tests mock `bottle.exec` / `bottle.cp_in` and assert on the dispatched script shape. The real round-trip lives in the chunk-4 integration smoke.""" from __future__ import annotations import subprocess import tempfile import unittest from dataclasses import replace from pathlib import Path from unittest.mock import MagicMock, patch from bot_bottle.agent_provider import ( AgentProvisionCommand, AgentProvisionDir, AgentProvisionFile, AgentProvisionPlan, ) from bot_bottle.backend import Bottle, BottleSpec, ExecResult from bot_bottle.backend.smolmachines.bottle_plan import ( SmolmachinesBottlePlan, ) 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 from bot_bottle.egress import EgressPlan, EgressRoute from bot_bottle.git_gate import GitGatePlan, GitGateUpstream from bot_bottle.manifest import GitEntry, Manifest from bot_bottle.pipelock import PipelockProxyPlan from bot_bottle.supervise import SupervisePlan from bot_bottle.workspace import workspace_plan def _make_bottle( name: str = "bot-bottle-demo-abc12", exec_result: ExecResult | None = None, ) -> MagicMock: bottle = MagicMock(spec=Bottle) bottle.name = name 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]: """All script strings passed to bottle.exec, in call order.""" return [c.args[0] for c in bottle.exec.call_args_list] def _exec_users(bottle: MagicMock) -> list[str]: """user= kwarg from each bottle.exec call, in order.""" return [c.kwargs.get("user", "node") for c in bottle.exec.call_args_list] def _plan( *, agent_prompt: str = "", skills: list[str] | None = None, git: list[GitEntry] = (), git_user: dict | None = None, copy_cwd: bool = False, user_cwd: str = "/tmp/x", stage_dir: Path | None = None, egress_routes: tuple[EgressRoute, ...] = (), egress_ca_path: Path = Path(), pipelock_ca_path: Path = Path(), supervise: bool = False, bundle_ip: str = "192.168.50.2", agent_git_gate_host: str = "127.0.0.1:55555", agent_supervise_url: str = "http://127.0.0.1:55556/", codex_auth_file: Path | None = None, agent_provider_template: str = "claude", guest_env: dict[str, str] | None = None, ) -> SmolmachinesBottlePlan: bottle_json: dict = {} git_gate_json: dict = {} if git: git_gate_json["repos"] = { g.Name: { "url": g.Upstream, "identity": g.IdentityFile, } for g in git } if git_user is not None: git_gate_json["user"] = git_user if git_gate_json: bottle_json["git-gate"] = git_gate_json if supervise: bottle_json["supervise"] = True manifest = Manifest.from_json_obj({ "bottles": {"dev": bottle_json}, "agents": { "demo": { "skills": list(skills or []), "prompt": agent_prompt, "bottle": "dev", }, }, }) spec = BottleSpec( manifest=manifest, agent_name="demo", copy_cwd=copy_cwd, user_cwd=user_cwd, ) supervise_plan = None if supervise: supervise_plan = SupervisePlan( slug="demo-abc12", queue_dir=Path("/tmp/queue"), current_config_dir=Path("/tmp/current-config"), ) return SmolmachinesBottlePlan( spec=spec, stage_dir=stage_dir or Path("/tmp/stage"), slug="demo-abc12", bundle_subnet="192.168.50.0/24", bundle_gateway="192.168.50.1", bundle_ip=bundle_ip, machine_name="bot-bottle-demo-abc12", agent_image_ref="bot-bottle-claude:latest", guest_env=dict(guest_env or {}), prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"), proxy_plan=PipelockProxyPlan( yaml_path=Path("/tmp/pipelock.yaml"), slug="demo-abc12", ca_cert_host_path=pipelock_ca_path, ), 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=egress_routes, token_env_map={}, mitmproxy_ca_cert_only_host_path=egress_ca_path, ), supervise_plan=supervise_plan, agent_git_gate_host=agent_git_gate_host, agent_supervise_url=agent_supervise_url, agent_provision=_agent_provision( agent_provider_template, codex_auth_file=codex_auth_file, guest_env=dict(guest_env or {}), ), workspace_plan=workspace_plan(spec, guest_home="/home/node"), ) def _agent_provision( template: str, *, codex_auth_file: Path | None = None, guest_env: dict[str, str] | None = None, ) -> AgentProvisionPlan: if template != "codex": return AgentProvisionPlan( template=template, command=template, prompt_mode="append_file", image="", dockerfile="", guest_env=dict(guest_env or {}), ) auth_dir = (guest_env or {}).get("CODEX_HOME", "/home/node/.codex") files = [ AgentProvisionFile( Path("/tmp/codex-config.toml"), f"{auth_dir}/config.toml", ), ] pre_copy: tuple[AgentProvisionCommand, ...] = () verify: tuple[AgentProvisionCommand, ...] = () if codex_auth_file is not None: files.append(AgentProvisionFile(codex_auth_file, f"{auth_dir}/auth.json")) pre_copy = (AgentProvisionCommand(( "find", auth_dir, "-maxdepth", "1", "-type", "f", "(", "-name", "*.sqlite", "-o", "-name", "*.sqlite-*", "-o", "-name", "*.codex-repair-*.bak", ")", "-delete", ), "codex host credentials: could not reset runtime db files"),) verify = (AgentProvisionCommand(( "runuser", "-u", "node", "--", "env", "HOME=/home/node", f"CODEX_HOME={auth_dir}", "codex", "login", "status", ), "codex host credentials: dummy auth was copied into the guest"),) return AgentProvisionPlan( template="codex", command="codex", prompt_mode="read_prompt_file", image="bot-bottle-codex:latest", dockerfile="", guest_env=dict(guest_env or {}), dirs=(AgentProvisionDir(auth_dir),), files=tuple(files), pre_copy=pre_copy, verify=verify, ) 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 actual bytes to chew on. Generated once per test via openssl.""" subprocess.run( ["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "/dev/null", "-out", str(path), "-days", "1", "-subj", "/CN=test"], check=True, capture_output=True, ) class TestProvisionCA(unittest.TestCase): """provision_ca selects the right CA cert (egress when the bottle has routes, else pipelock) and dispatches cp_in + exec in the right order.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-ca.") self.tmp = Path(self._tmp.name) self.pipelock_ca = self.tmp / "pipelock-ca.pem" self.egress_ca = self.tmp / "egress-ca.pem" _write_self_signed_cert(self.pipelock_ca) _write_self_signed_cert(self.egress_ca) def tearDown(self): self._tmp.cleanup() # provision_ca dies hard if update-ca-certificates' exit # is non-zero; supply a stock success return so the bulk of # the tests below exercise the happy path. _UPDATE_OK = ExecResult( returncode=0, stdout="Updating certificates in /etc/ssl/certs...\n1 added, 0 removed; done.\n", stderr="", ) def test_pipelock_path_when_no_routes(self): plan = _plan(pipelock_ca_path=self.pipelock_ca) bottle = _make_bottle(exec_result=self._UPDATE_OK) _ca.provision_ca(plan, bottle) bottle.cp_in.assert_called_once_with( str(self.pipelock_ca), _ca.AGENT_CA_PATH, ) # chmod + chown + update-ca-certificates are folded into # one exec invocation; look at the single exec's script # rather than expecting separate calls. bottle.exec.assert_called_once() script = bottle.exec.call_args.args[0] self.assertIn("chmod 644", script) self.assertIn("update-ca-certificates", script) self.assertEqual("root", bottle.exec.call_args.kwargs.get("user")) def test_egress_path_when_routes_declared(self): plan = _plan( egress_routes=(EgressRoute(host="api.anthropic.com"),), egress_ca_path=self.egress_ca, pipelock_ca_path=self.pipelock_ca, ) bottle = _make_bottle(exec_result=self._UPDATE_OK) _ca.provision_ca(plan, bottle) # When routes are declared, egress is the agent's first hop, # so egress's CA is the one that gets installed. bottle.cp_in.assert_called_once_with( str(self.egress_ca), _ca.AGENT_CA_PATH, ) def test_retries_smolvm_sigkill_during_update_ca(self): plan = _plan(pipelock_ca_path=self.pipelock_ca) killed = ExecResult( returncode=137, stdout="Updating certificates in /etc/ssl/certs...\n", stderr="", ) bottle = _make_bottle() bottle.exec.side_effect = [killed, self._UPDATE_OK] with patch( "bot_bottle.backend.smolmachines.provision.ca.time.sleep" ) as sleep: _ca.provision_ca(plan, bottle) self.assertEqual(2, bottle.exec.call_count) sleep.assert_called_once_with(1.0) def test_dies_when_selected_cert_missing(self): # Plan claims a pipelock cert at a path that doesn't exist — # something went wrong in launch's pipelock_tls_init. plan = _plan(pipelock_ca_path=self.tmp / "does-not-exist.pem") bottle = _make_bottle() with self.assertRaises(SystemExit): _ca.provision_ca(plan, bottle) class TestProvisionGit(unittest.TestCase): """provision_git dispatches two independent passes (cwd .git copy + gitconfig insteadOf write); each no-ops on its own when its condition doesn't hold.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-git.") self.stage = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def test_noop_when_no_cwd_and_no_git_entries(self): bottle = _make_bottle() _git.provision_git(_plan(stage_dir=self.stage), bottle) bottle.cp_in.assert_not_called() bottle.exec.assert_not_called() def test_copies_cwd_git_when_copy_cwd_and_git_present(self): # Stage a fake host .git dir under user_cwd so the path- # check in _provision_cwd_git fires. cwd = self.stage / "cwd" (cwd / ".git").mkdir(parents=True) plan = _plan( copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage, ) bottle = _make_bottle() _git.provision_git(plan, bottle) bottle.cp_in.assert_called_once_with( f"{cwd}/.git", "/home/node/workspace/.git", ) scripts = _exec_scripts(bottle) self.assertTrue(any("mkdir -p" in s and "/home/node/workspace" in s for s in scripts)) # chown the workspace tree so the agent (node) owns it. self.assertTrue( any("chown -R" in s and "node:node" in s and "/home/node/workspace/.git" in s for s in scripts) ) def test_skips_cwd_when_copy_cwd_false(self): plan = _plan(copy_cwd=False, stage_dir=self.stage) bottle = _make_bottle() _git.provision_git(plan, bottle) bottle.cp_in.assert_not_called() def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self): # Smolmachines's TSI-allowlisted guest dials git-gate via # smart HTTP at `127.0.0.1:` — the bundle's # git HTTP port is published on host loopback at launch # time, and the plan carries the discovered host port. plan = _plan( git=[GitEntry( Name="bot-bottle", Upstream="ssh://git@host/repo.git", IdentityFile="~/.ssh/id_ed25519", )], stage_dir=self.stage, agent_git_gate_host="127.0.0.1:9418", ) bottle = _make_bottle() _git.provision_git(plan, bottle) # The staged gitconfig path is whatever NamedTemporaryFile # picked; we read its contents. cp_call = bottle.cp_in.call_args staged_path = Path(cp_call.args[0]) self.assertEqual(self.stage, staged_path.parent) content = staged_path.read_text() self.assertIn( '[url "http://127.0.0.1:9418/bot-bottle.git"]', content, ) self.assertIn( "\tinsteadOf = ssh://git@host/repo.git", content, ) class TestBundleLaunchSpec(unittest.TestCase): def test_git_gate_uses_http_daemon_for_smolmachines(self): plan = _plan() plan = replace( plan, git_gate_plan=replace( plan.git_gate_plan, upstreams=(GitGateUpstream( name="bot-bottle", upstream_url="ssh://git@host/repo.git", upstream_host="host", upstream_port="22", identity_file="/tmp/key", known_host_key="", ),), ), ) spec = _bundle_launch_spec(plan, "net", "127.0.0.16") self.assertEqual( "egress,pipelock,git-gate,git-http", spec.daemons_csv, ) self.assertIn(9420, spec.ports_to_publish) self.assertNotIn(9418, spec.ports_to_publish) class TestProvisionGitUser(unittest.TestCase): """`_provision_git_user` runs `git config --global` inside the guest as the node user. SmolmachinesBottle.exec sets HOME and USER automatically for the requested user, so --global lands in /home/node/.gitconfig. No-op when the bottle didn't declare git_user (issue #86).""" def _git_config_calls(self, bottle: MagicMock) -> list[tuple[str, str]]: """Filter bottle.exec calls down to git-config invocations, return list of (script, user) tuples.""" out = [] for c in bottle.exec.call_args_list: script = c.args[0] if c.args else "" user = c.kwargs.get("user", "node") if "git config" in script: out.append((script, user)) return out def test_noop_when_no_git_user(self): bottle = _make_bottle() _git._provision_git_user(_plan(), bottle) self.assertEqual([], self._git_config_calls(bottle)) def test_sets_name_and_email_as_node(self): plan = _plan(git_user={ "name": "Eric Bauerfeld", "email": "eric@dideric.is", }) bottle = _make_bottle() _git._provision_git_user(plan, bottle) calls = self._git_config_calls(bottle) self.assertEqual(2, len(calls)) # Both run as node so SmolmachinesBottle.exec sets HOME=/home/node # automatically, ensuring --global writes to /home/node/.gitconfig. for script, user in calls: self.assertEqual("node", user) self.assertIn("git config --global", script) self.assertIn("user.name", calls[0][0]) self.assertIn("Eric Bauerfeld", calls[0][0]) self.assertIn("user.email", calls[1][0]) self.assertIn("eric@dideric.is", calls[1][0]) def test_name_only(self): plan = _plan(git_user={"name": "Bot"}) bottle = _make_bottle() _git._provision_git_user(plan, bottle) calls = self._git_config_calls(bottle) self.assertEqual(1, len(calls)) self.assertIn("user.name", calls[0][0]) self.assertIn("Bot", calls[0][0]) def test_email_only(self): plan = _plan(git_user={"email": "bot@example.com"}) bottle = _make_bottle() _git._provision_git_user(plan, bottle) calls = self._git_config_calls(bottle) self.assertEqual(1, len(calls)) self.assertIn("user.email", calls[0][0]) self.assertIn("bot@example.com", calls[0][0]) class TestProvisionWorkspace(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-workspace.") self.stage = Path(self._tmp.name) def tearDown(self): self._tmp.cleanup() def test_noop_when_copy_cwd_false(self): plan = _plan(copy_cwd=False, stage_dir=self.stage) bottle = _make_bottle() _workspace.provision_workspace(plan, bottle) bottle.cp_in.assert_not_called() bottle.exec.assert_not_called() def test_copies_workspace_to_plan_path_and_chowns(self): cwd = self.stage / "cwd" cwd.mkdir() plan = _plan(copy_cwd=True, user_cwd=str(cwd), stage_dir=self.stage) bottle = _make_bottle() _workspace.provision_workspace(plan, bottle) bottle.cp_in.assert_called_once_with( str(cwd), "/home/node/workspace", ) scripts = _exec_scripts(bottle) self.assertTrue( any("rm -rf /home/node/workspace" in s and "mkdir -p /home/node" in s for s in scripts) ) self.assertTrue( any("chown -R node:node /home/node/workspace" in s and "chmod 755 /home/node/workspace" in s for s in scripts) ) 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()