chore: comment out workspace + capability_apply, fix circular imports

The recent refactor partially removed workspace planning and
capability-apply logic. This commit finishes the cleanup so the
test suite imports cleanly:

- Comment out workspace_plan field/property on BottlePlan and the
  provision_workspace dispatch.
- Comment out workspace usages in docker.util (build_image_with_cwd),
  smolmachines.provision.workspace, agent_provider.provision_git,
  smolmachines.backend.
- Comment out capability_apply imports in cli.start and cli.supervise;
  add a local CapabilityApplyError placeholder so the supervise CLI
  module still imports.
- Break the bottle_state → backend.docker → backend circular import
  by lazy-loading docker_mod inside bottle_identity, and by moving the
  resolve_common import inside BottleBackend.prepare.
- Delete tests for workspace and capability_apply (unit + integration).
- Update test fixtures to drop removed kwargs (container_name_pinned,
  derived_image, env_file, workspace_plan, agent_image_ref) from
  DockerBottlePlan / SmolmachinesBottlePlan constructors.
- Delete the obsolete test_smolmachines_prepare.py (tested the old
  resolve_plan signature; the shared prepare flow now lives in
  BottleBackend.prepare).
- Adjust test_supervise.py for the new Supervise.prepare signature
  (dockerfile_content arg removed).

925 → 897 tests, all passing.
This commit is contained in:
2026-06-08 17:36:51 +00:00
committed by didericis
parent 74efb1c143
commit b38c6110f2
23 changed files with 150 additions and 957 deletions
-132
View File
@@ -1,132 +0,0 @@
"""Unit: capability_apply helpers (PRD 0016 Phase 2).
docker cp / exec / rm / network rm paths are covered by the
integration test in Phase 4. Here we cover:
- fetch_current_dockerfile fallback chain (per-bottle → repo)
- apply_capability_change writes the per-bottle Dockerfile and
returns the correct (before, after).
- apply_capability_change rejects empty input.
"""
import tempfile
import unittest
from pathlib import Path
from bot_bottle import supervise
from bot_bottle import bottle_state
from bot_bottle.backend.docker import capability_apply
from bot_bottle.backend.docker.capability_apply import (
CapabilityApplyError,
apply_capability_change,
fetch_current_dockerfile,
)
class _FakeHomeMixin:
def _setup_fake_home(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cap-apply-test.")
original = supervise.bot_bottle_root
def fake_root() -> Path:
return Path(self._tmp.name) / ".bot-bottle"
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
self._restore = lambda: setattr(supervise, "bot_bottle_root", original)
def _teardown_fake_home(self):
self._restore()
self._tmp.cleanup()
class TestFetchCurrentDockerfile(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
def tearDown(self):
self._teardown_fake_home()
def test_returns_per_bottle_dockerfile_when_present(self):
bottle_state.write_per_bottle_dockerfile("dev", "FROM rebuilt\n")
self.assertEqual("FROM rebuilt\n", fetch_current_dockerfile("dev"))
def test_falls_back_to_repo_dockerfile_when_no_override(self):
# The repo's Dockerfile actually exists; the test just checks
# we get its content (non-empty) when no per-bottle override
# is set.
content = fetch_current_dockerfile("dev-no-override")
self.assertIn("FROM ", content)
class TestApplyCapabilityChange(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
# Stub out the docker-dependent helpers. The orchestrator's
# job is to sequence write + snapshot + push + teardown; we
# validate that sequence here, not the docker primitives.
self._calls: list[str] = []
self._orig_snapshot = capability_apply.snapshot_transcript
self._orig_push = capability_apply._push_working_tree
self._orig_teardown = capability_apply._teardown_bottle
def stub_snapshot(slug: object) -> None: # type: ignore
self._calls.append(f"snapshot:{slug}")
def stub_push(slug: object) -> None: # type: ignore
self._calls.append(f"push:{slug}")
def stub_teardown(slug: object) -> None: # type: ignore
self._calls.append(f"teardown:{slug}")
capability_apply.snapshot_transcript = stub_snapshot # type: ignore[assignment]
capability_apply._push_working_tree = stub_push # type: ignore[assignment]
capability_apply._teardown_bottle = stub_teardown # type: ignore[assignment]
def tearDown(self):
capability_apply.snapshot_transcript = self._orig_snapshot # type: ignore[assignment]
capability_apply._push_working_tree = self._orig_push # type: ignore[assignment]
capability_apply._teardown_bottle = self._orig_teardown # type: ignore[assignment]
self._teardown_fake_home()
def test_writes_per_bottle_dockerfile_and_returns_before_after(self):
bottle_state.write_per_bottle_dockerfile("dev", "FROM old\n")
before, after = apply_capability_change("dev", "FROM new\nRUN apk add ripgrep\n")
self.assertEqual("FROM old\n", before)
self.assertEqual("FROM new\nRUN apk add ripgrep\n", after)
self.assertEqual(
"FROM new\nRUN apk add ripgrep\n",
bottle_state.per_bottle_dockerfile("dev"),
)
def test_calls_snapshot_push_teardown_in_order(self):
apply_capability_change("dev", "FROM new\n")
# Snapshot + push must happen BEFORE write_per_bottle_dockerfile
# (so they capture pre-rebuild state) and BEFORE teardown (so
# the agent container still exists to docker exec / cp from).
# Teardown must be last.
self.assertEqual(
["snapshot:dev", "push:dev", "teardown:dev"],
self._calls,
)
def test_marks_preserved_before_teardown(self):
# cli.py's session-end cleanup reads the marker after the
# bottle is torn down. The marker must therefore be written
# before teardown — otherwise the cleanup would see no
# marker and rm the state dir we just populated.
apply_capability_change("dev", "FROM new\n")
self.assertTrue(bottle_state.is_preserved("dev"))
def test_first_change_falls_back_to_repo_dockerfile_for_before(self):
# No per-bottle override yet — before-diff comes from the
# repo's Dockerfile.
before, after = apply_capability_change("dev-fresh", "FROM new\n")
self.assertIn("FROM ", before)
self.assertEqual("FROM new\n", after)
def test_empty_dockerfile_rejected(self):
with self.assertRaises(CapabilityApplyError):
apply_capability_change("dev", " \n\t\n")
if __name__ == "__main__":
unittest.main()
+5 -14
View File
@@ -29,29 +29,20 @@ class _FakeHomeMixin:
class TestCaptureSessionState(_FakeHomeMixin, unittest.TestCase):
# snapshot_transcript is commented out (capability_apply is disabled);
# capture_claude_session_state now only handles the preserve marker.
def setUp(self):
self._setup_fake_home()
# Stub the docker-dependent snapshot call so this stays a
# unit test. apply_capability_change's integration test
# covers the real docker cp path.
self._snap_calls: list[str] = []
self._orig_snap = start_mod.snapshot_transcript
start_mod.snapshot_transcript = lambda identity: ( # type: ignore
self._snap_calls.append(identity)
)
def tearDown(self):
start_mod.snapshot_transcript = self._orig_snap
self._teardown_fake_home()
def test_clean_exit_snapshots_but_does_not_mark(self):
def test_clean_exit_does_not_mark(self):
start_mod.capture_claude_session_state("dev-abc", exit_code=0)
self.assertEqual(["dev-abc"], self._snap_calls)
self.assertFalse(bottle_state.is_preserved("dev-abc"))
def test_crash_snapshots_and_marks(self):
def test_crash_marks_preserved(self):
start_mod.capture_claude_session_state("dev-abc", exit_code=137)
self.assertEqual(["dev-abc"], self._snap_calls)
self.assertTrue(bottle_state.is_preserved("dev-abc"))
def test_ctrl_c_treated_as_crash(self):
@@ -64,7 +55,7 @@ class TestCaptureSessionState(_FakeHomeMixin, unittest.TestCase):
# Backends without an identity field shouldn't crash this
# path (the _identity_from_plan helper falls back to "").
start_mod.capture_claude_session_state("", exit_code=137)
self.assertEqual([], self._snap_calls)
self.assertFalse(bottle_state.is_preserved(""))
class TestSettleState(_FakeHomeMixin, unittest.TestCase):
+2 -7
View File
@@ -33,7 +33,7 @@ from bot_bottle.egress import (
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import Manifest
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
SLUG = "demo-abc12"
@@ -153,12 +153,8 @@ def _plan(
stage_dir=STAGE,
slug=SLUG,
container_name=f"bot-bottle-{SLUG}",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
agent_image="bot-bottle-claude: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",
git_gate_plan=_git_gate_plan(upstreams),
@@ -174,7 +170,6 @@ def _plan(
guest_home="/home/node",
guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
@@ -210,7 +205,7 @@ class TestAgentAlwaysPresent(unittest.TestCase):
def test_agent_image_uses_runtime_image(self):
plan = _plan()
s = bottle_plan_to_compose(plan)["services"]["agent"]
self.assertEqual(plan.agent_image, s["image"])
self.assertEqual(plan.image, s["image"])
def test_agent_only_on_internal_network(self):
s = bottle_plan_to_compose(_plan())["services"]["agent"]
+1 -6
View File
@@ -24,7 +24,7 @@ from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
_URL = "http://supervise:9100/"
@@ -80,12 +80,8 @@ def _plan(
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="",
agent_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"),
git_gate_plan=GitGatePlan(
@@ -107,7 +103,6 @@ def _plan(
template="claude", command="claude", prompt_mode="append_file",
image="", dockerfile="", guest_home="/home/node", guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
+1 -6
View File
@@ -25,7 +25,7 @@ from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.manifest import Manifest
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
_URL = "http://supervise:9100/"
@@ -81,12 +81,8 @@ def _plan(
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="",
agent_image="bot-bottle-codex:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/state/demo-abc12/agent/prompt.txt"),
git_gate_plan=GitGatePlan(
@@ -108,7 +104,6 @@ def _plan(
template="codex", command="codex", prompt_mode="read_prompt_file",
image="", dockerfile="", guest_home="/home/node", guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
+1 -6
View File
@@ -22,7 +22,7 @@ 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.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
def _manifest() -> Manifest:
@@ -68,15 +68,10 @@ def _plan(tmp: str) -> DockerBottlePlan:
guest_home="/home/node",
guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
slug="test-teardown-00001",
container_name="bot-bottle-test-teardown-abc",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
agent_image="bot-bottle-claude:latest",
dockerfile_path="",
env_file=stage / "env",
forwarded_env={},
prompt_file=stage / "prompt.txt",
use_runsc=False,
+4 -24
View File
@@ -22,7 +22,7 @@ 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.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
class _Provider(AgentProvider):
@@ -65,12 +65,8 @@ def _plan(*, git_user: dict | None = None, # type: ignore
stage_dir=stage_dir or Path("/tmp/stage"),
slug="demo-abc12",
container_name="bot-bottle-demo-abc12",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
agent_image="bot-bottle-claude:latest",
dockerfile_path="",
env_file=Path("/tmp/agent.env"),
forwarded_env={},
prompt_file=Path("/tmp/prompt.txt"),
git_gate_plan=GitGatePlan(
@@ -97,7 +93,6 @@ def _plan(*, git_user: dict | None = None, # type: ignore
guest_home="/home/node",
guest_env={},
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
@@ -131,24 +126,9 @@ class TestProvisionGitUser(unittest.TestCase):
_PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
self.assertEqual([], _git_config_exec_calls(bottle))
def test_copies_cwd_git_to_workspace_plan_path(self):
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()
_PROVIDER.provision_git(bottle, plan)
bottle.cp_in.assert_called_once_with(
f"{cwd}/.git",
"/home/node/workspace/.git",
)
chown_calls = [
c for c in bottle.exec.call_args_list
if "chown" in (c.args[0] if c.args else "")
and "/home/node/workspace/.git" in (c.args[0] if c.args else "")
]
self.assertEqual(1, len(chown_calls))
self.assertIn("node:node", chown_calls[0].args[0])
# def test_copies_cwd_git_to_workspace_plan_path(self):
# # DISABLED — workspace planning is currently commented out.
# pass
def test_sets_name_and_email(self):
plan = _plan(
-56
View File
@@ -14,7 +14,6 @@ from pathlib import Path
from unittest.mock import patch
from bot_bottle.backend.docker import util as docker_mod
from bot_bottle.workspace import WorkspacePlan
def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess: # type: ignore
@@ -70,60 +69,5 @@ class TestSave(unittest.TestCase):
)
class TestBuildImageWithCwd(unittest.TestCase):
def test_uses_workspace_plan_paths(self):
with tempfile.TemporaryDirectory(prefix="bb-docker-cwd.") as tmp:
workspace = WorkspacePlan(
enabled=True,
host_path=Path(tmp),
guest_home="/guest/home",
guest_path="/guest/home/workspace",
workdir="/guest/home/workspace",
)
with patch.object(docker_mod.subprocess, "run") as run:
docker_mod.build_image_with_cwd("derived:tag", "base:tag", workspace)
argv = run.call_args.args[0]
dockerfile = run.call_args.kwargs["input"]
self.assertEqual(["docker", "build", "-t", "derived:tag", "-f", "-"], argv[:6])
self.assertTrue(argv[6].endswith("/context"))
self.assertIn("FROM base:tag\n", dockerfile)
self.assertIn(
"COPY --chown=node:node workspace/. /guest/home/workspace\n",
dockerfile,
)
self.assertIn("WORKDIR /guest/home/workspace\n", dockerfile)
def test_staged_context_includes_hidden_files_but_not_git_dir(self):
with tempfile.TemporaryDirectory(prefix="bb-docker-cwd.") as tmp:
root = Path(tmp)
(root / ".gitignore").write_text("*.pyc\n")
(root / ".dockerignore").write_text(".gitignore\n")
(root / ".env.example").write_text("SAFE=1\n")
(root / ".git").mkdir()
(root / ".git" / "config").write_text("[core]\n")
workspace = WorkspacePlan(
enabled=True,
host_path=root,
guest_home="/guest/home",
guest_path="/guest/home/workspace",
workdir="/guest/home/workspace",
)
def inspect_context(*args, **kwargs): # type: ignore
context = Path(args[0][-1])
staged = context / "workspace"
self.assertTrue((staged / ".gitignore").is_file())
self.assertTrue((staged / ".dockerignore").is_file())
self.assertTrue((staged / ".env.example").is_file())
self.assertFalse((staged / ".git").exists())
return _ok()
with patch.object(
docker_mod.subprocess, "run", side_effect=inspect_context,
):
docker_mod.build_image_with_cwd("derived:tag", "base:tag", workspace)
if __name__ == "__main__":
unittest.main()
+1 -7
View File
@@ -20,7 +20,7 @@ from bot_bottle.backend.smolmachines.bottle_plan import SmolmachinesBottlePlan
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import Manifest
from bot_bottle.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
def _manifest() -> Manifest:
@@ -100,15 +100,10 @@ def _docker_plan(spec: BottleSpec, tmp: str) -> DockerBottlePlan:
egress_plan=_egress_plan(tmp),
supervise_plan=None,
agent_provision=_agent_provision(),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
slug="test-00001",
container_name="bot-bottle-test-00001",
container_name_pinned=False,
image="bot-bottle-claude:latest",
derived_image="",
agent_image="bot-bottle-claude:latest",
dockerfile_path="",
env_file=stage / "env",
forwarded_env={},
prompt_file=stage / "prompt.txt",
use_runsc=False,
@@ -124,7 +119,6 @@ def _smolmachines_plan(spec: BottleSpec, tmp: str) -> SmolmachinesBottlePlan:
egress_plan=_egress_plan(tmp),
supervise_plan=None,
agent_provision=_agent_provision(),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
slug="test-00001",
bundle_subnet="10.99.0.0/24",
bundle_gateway="10.99.0.1",
-133
View File
@@ -1,133 +0,0 @@
"""Unit: smolmachines prepare.py env resolution (PRD 0038)."""
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.env import ResolvedEnv
class TestSmolmachinesResolveEnv(unittest.TestCase):
"""resolve_plan() must call resolve_env() and build guest_env
from the resolved values rather than from raw bottle.env."""
def _run_resolve_plan(
self,
resolved: ResolvedEnv,
*,
extra_host_env: dict[str, str] | None = None,
) -> dict[str, str]:
from bot_bottle.backend import BottleSpec
from bot_bottle.manifest import Manifest
with tempfile.TemporaryDirectory() as tmp:
stage = Path(tmp) / "stage"
stage.mkdir()
# Minimal manifest with one env literal so the spec is valid.
manifest = Manifest.from_json_obj({
"agents": {"myagent": {"bottle": "mybottle"}},
"bottles": {"mybottle": {"env": {"PLAIN": "literal-value"}}},
})
spec = BottleSpec(
manifest=manifest,
agent_name="myagent",
copy_cwd=False,
user_cwd=tmp,
identity="test-slug-00001",
)
from bot_bottle import supervise as _sup
orig_root = _sup.bot_bottle_root
_sup.bot_bottle_root = lambda: Path(tmp) / ".bot-bottle" # type: ignore[assignment]
host_env = {**os.environ, **(extra_host_env or {})} # type: ignore
try:
with (
patch("bot_bottle.backend.smolmachines.resolve_plan.resolve_env",
return_value=resolved) as mock_resolve,
patch("bot_bottle.backend.smolmachines.resolve_plan.smolmachines_preflight"),
patch("bot_bottle.backend.smolmachines.resolve_plan.smolmachines_bundle_subnet",
return_value=("10.99.0.0/24", "10.99.0.1", "10.99.0.2")),
patch("bot_bottle.backend.resolve_common.GitGate") as mock_gg,
patch("bot_bottle.backend.resolve_common.Egress") as mock_eg,
patch("bot_bottle.backend.resolve_common.Supervise"),
patch(
"bot_bottle.backend.smolmachines.resolve_plan.agent_provision_plan"
) as mock_app,
):
mock_gg.return_value.prepare.return_value = MagicMock()
mock_eg.return_value.prepare.return_value = MagicMock()
def _make_provision(**kwargs): # type: ignore
return AgentProvisionPlan(
template="claude",
command="claude",
prompt_mode="append_file",
dockerfile="",
image="bot-bottle-claude:latest",
guest_home="/home/node",
guest_env=dict(kwargs.get("guest_env") or {}),
)
mock_app.side_effect = lambda **kw: _make_provision(**kw) # type: ignore
from bot_bottle.backend.smolmachines.resolve_plan import resolve_plan
plan = resolve_plan(spec, stage_dir=stage)
mock_resolve.assert_called_once_with(manifest, "myagent")
return dict(plan.guest_env)
finally:
_sup.bot_bottle_root = orig_root # type: ignore[assignment]
def test_literal_env_reaches_guest_env(self):
resolved = ResolvedEnv(
literals={"PLAIN": "hello"},
forwarded={},
)
guest_env = self._run_resolve_plan(resolved)
self.assertEqual("hello", guest_env["PLAIN"])
def test_forwarded_env_reaches_guest_env(self):
# Secrets / interpolated values land in forwarded; they must
# still reach the guest (argv exposure is the known gap).
resolved = ResolvedEnv(
literals={},
forwarded={"SECRET": "s3cr3t", "INTERP": "resolved-val"},
)
guest_env = self._run_resolve_plan(resolved)
self.assertEqual("s3cr3t", guest_env["SECRET"])
self.assertEqual("resolved-val", guest_env["INTERP"])
def test_raw_manifest_sentinel_not_in_guest_env(self):
# Before the fix, ?prompt and ${HOST} would appear verbatim.
# After the fix, resolve_env() is called so the caller sees
# the mocked resolved values (no raw sentinel survives).
resolved = ResolvedEnv(
literals={},
forwarded={"MY_SECRET": "actual-value"},
)
guest_env = self._run_resolve_plan(resolved)
for v in guest_env.values():
self.assertFalse(
v.startswith("?"),
f"raw secret sentinel survived in guest_env: {v!r}",
)
self.assertFalse(
v.startswith("${"),
f"raw interpolation sentinel survived in guest_env: {v!r}",
)
def test_tls_trust_env_always_present(self):
resolved = ResolvedEnv(literals={}, forwarded={})
guest_env = self._run_resolve_plan(resolved)
for key in ("NODE_EXTRA_CA_CERTS", "SSL_CERT_FILE", "REQUESTS_CA_BUNDLE"):
self.assertIn(key, guest_env, f"{key} missing from guest_env")
if __name__ == "__main__":
unittest.main()
+13 -67
View File
@@ -26,16 +26,16 @@ from bot_bottle.backend.smolmachines.bottle import SmolmachinesBottle
from bot_bottle.backend.smolmachines.bottle_plan import (
SmolmachinesBottlePlan,
)
from bot_bottle.backend.smolmachines.provision import (
workspace as _workspace,
)
# from bot_bottle.backend.smolmachines.provision import (
# workspace as _workspace,
# )
from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.backend.util import AGENT_CA_PATH
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import ManifestGitEntry, Manifest
from bot_bottle.supervise import SupervisePlan
from bot_bottle.workspace import workspace_plan
# from bot_bottle.workspace import workspace_plan
class _Provider(AgentProvider):
@@ -172,7 +172,6 @@ def _plan(
codex_auth_file=codex_auth_file,
guest_env=dict(guest_env or {}),
),
workspace_plan=workspace_plan(spec, guest_home="/home/node"),
)
@@ -359,33 +358,13 @@ class TestProvisionGit(unittest.TestCase):
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_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()
_PROVIDER.provision_git(bottle, plan)
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_copies_cwd_git_when_copy_cwd_and_git_present(self):
# # DISABLED — workspace planning is currently commented out.
# pass
def test_skips_cwd_when_copy_cwd_false(self):
plan = _plan(copy_cwd=False, stage_dir=self.stage)
bottle = _make_bottle()
_PROVIDER.provision_git(bottle, plan)
bottle.cp_in.assert_not_called()
# def test_skips_cwd_when_copy_cwd_false(self):
# # DISABLED — workspace planning is currently commented out.
# pass
def test_writes_gitconfig_with_ip_port_form_for_smolmachines(self):
# Smolmachines's TSI-allowlisted guest dials git-gate via
@@ -506,42 +485,9 @@ class TestProvisionGitUser(unittest.TestCase):
self.assertIn("bot@example.com", calls[0][0])
class TestProvisionWorkspace(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="cb-prov-workspace.") # pylint: disable=consider-using-with
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 TestProvisionWorkspace(unittest.TestCase):
# # DISABLED — workspace planning / provision_workspace are commented out.
# pass
if __name__ == "__main__":
+5 -11
View File
@@ -359,25 +359,19 @@ class TestSupervisePrepare(unittest.TestCase):
return lambda: setattr(supervise, "bot_bottle_root", original)
def test_prepare_creates_queue_and_current_config(self):
plan = _StubSupervise().prepare(
"dev", self.stage_dir,
dockerfile_content="FROM python:3.13\n",
)
plan = _StubSupervise().prepare("dev", self.stage_dir)
self.assertTrue(plan.queue_dir.is_dir())
self.assertTrue(plan.current_config_dir.is_dir())
self.assertEqual(
"FROM python:3.13\n",
(plan.current_config_dir / "Dockerfile").read_text(),
)
self.assertEqual("dev", plan.slug)
self.assertEqual("", plan.internal_network)
def test_prepare_only_writes_dockerfile_to_current_config(self):
def test_prepare_writes_no_files_to_current_config(self):
# dockerfile_content is no longer accepted by prepare.
# routes.yaml + allowlist live behind the
# `list-egress-routes` MCP tool now (PRD 0017 chunk 3).
# `list-egress-routes` MCP tool (PRD 0017 chunk 3).
plan = _StubSupervise().prepare("dev", self.stage_dir)
files = sorted(p.name for p in plan.current_config_dir.iterdir())
self.assertEqual(["Dockerfile"], files)
self.assertEqual([], files)
if __name__ == "__main__":
+6 -112
View File
@@ -115,13 +115,8 @@ class TestDiscoverPending(_FakeHomeMixin, unittest.TestCase):
class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
def setUp(self):
self._setup_fake_home()
self._original_apply_capability = supervise_cli.apply_capability_change
supervise_cli.apply_capability_change = lambda slug, content: ( # type: ignore
"FROM old\n", content,
)
def tearDown(self):
supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
def _enqueue(self, tool: str = TOOL_CAPABILITY_BLOCK):
@@ -161,67 +156,9 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
self.assertEqual([], read_audit_entries("egress", "dev"))
class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
"""PRD 0016 Phase 3: approve() on a capability-block proposal
calls apply_capability_change, archives the proposal afterward
(sidecar is gone so it can't archive itself), and writes no
audit entry (capability-block has none per PRD 0013)."""
def setUp(self):
self._setup_fake_home()
self._original = supervise_cli.apply_capability_change
def tearDown(self):
supervise_cli.apply_capability_change = self._original
self._teardown_fake_home()
def _enqueue_capability(self, proposed: str = "FROM python:3.13\nRUN apk add ripgrep\n"):
p = Proposal.new(
bottle_slug="dev", tool=TOOL_CAPABILITY_BLOCK,
proposed_file=proposed,
justification="need ripgrep",
current_file_hash=sha256_hex(proposed),
now=FIXED,
)
qdir = supervise.queue_dir_for_slug("dev")
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def test_capability_block_calls_apply_with_proposed_file(self):
calls = []
supervise_cli.apply_capability_change = lambda slug, content: ( # type: ignore
calls.append((slug, content)) or ("FROM old\n", content)
)
qp = self._enqueue_capability("FROM bookworm\n")
supervise_cli.approve(qp)
self.assertEqual([("dev", "FROM bookworm\n")], calls)
def test_apply_failure_blocks_response_and_keeps_pending(self):
supervise_cli.apply_capability_change = lambda slug, content: (_ for _ in ()).throw( # type: ignore
CapabilityApplyError("teardown failed")
)
qp = self._enqueue_capability()
with self.assertRaises(CapabilityApplyError):
supervise_cli.approve(qp)
self.assertEqual(
[qp.proposal.id],
[p.id for p in supervise.list_pending_proposals(qp.queue_dir)],
)
def test_no_audit_log_for_capability(self):
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore
qp = self._enqueue_capability()
supervise_cli.approve(qp)
self.assertEqual([], read_audit_entries("egress", "dev"))
def test_proposal_archived_after_apply(self):
supervise_cli.apply_capability_change = lambda slug, content: ("FROM old\n", content) # type: ignore
qp = self._enqueue_capability()
supervise_cli.approve(qp)
self.assertEqual([], supervise.list_pending_proposals(qp.queue_dir))
processed = list((qp.queue_dir / "processed").glob("*.json"))
self.assertEqual(2, len(processed))
# class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
# # DISABLED — capability_apply functionality is currently commented out.
# pass
class TestEditInEditor(unittest.TestCase):
@@ -268,52 +205,9 @@ class TestEditInEditor(unittest.TestCase):
os.environ["EDITOR"] = original_editor
class TestCapabilityBlockSmolmachinesGuard(_FakeHomeMixin, unittest.TestCase):
"""approve() must refuse capability-block for smolmachines bottles and
pass it through for Docker bottles (PRD 0039)."""
def setUp(self):
self._setup_fake_home()
self._original_apply_capability = supervise_cli.apply_capability_change
supervise_cli.apply_capability_change = lambda slug, content: ("", content) # type: ignore
def tearDown(self):
supervise_cli.apply_capability_change = self._original_apply_capability
self._teardown_fake_home()
def _enqueue_capability(self, slug: str = "dev") -> "supervise_cli.QueuedProposal":
p = _proposal(slug=slug, tool=TOOL_CAPABILITY_BLOCK)
qdir = supervise.queue_dir_for_slug(slug)
qdir.mkdir(parents=True, exist_ok=True)
supervise.write_proposal(qdir, p)
return supervise_cli.QueuedProposal(proposal=p, queue_dir=qdir)
def _write_metadata(self, slug: str, compose_project: str) -> None:
from bot_bottle.bottle_state import BottleMetadata, write_metadata
write_metadata(BottleMetadata(
identity=slug,
agent_name="myagent",
cwd="",
copy_cwd=False,
started_at="2026-06-02T00:00:00+00:00",
compose_project=compose_project,
))
def test_smolmachines_bottle_raises_capability_apply_error(self):
self._write_metadata("dev", compose_project="")
qp = self._enqueue_capability("dev")
with self.assertRaises(CapabilityApplyError) as ctx:
supervise_cli.approve(qp)
self.assertIn("smolmachines", str(ctx.exception))
def test_docker_bottle_calls_apply_capability_change(self):
self._write_metadata("dev", compose_project="bot-bottle-dev")
qp = self._enqueue_capability("dev")
supervise_cli.approve(qp) # must not raise
def test_no_metadata_falls_through_to_docker_path(self):
qp = self._enqueue_capability("dev")
supervise_cli.approve(qp) # must not raise
# class TestCapabilityBlockSmolmachinesGuard(_FakeHomeMixin, unittest.TestCase):
# # DISABLED — capability_apply functionality is currently commented out.
# pass
if __name__ == "__main__":
-58
View File
@@ -1,58 +0,0 @@
"""Unit: backend-neutral workspace planning."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from bot_bottle.backend import BottleSpec
from bot_bottle.manifest import Manifest
from bot_bottle.workspace import workspace_plan
def _spec(*, copy_cwd: bool, user_cwd: str) -> BottleSpec:
manifest = Manifest.from_json_obj({
"bottles": {"dev": {}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
return BottleSpec(
manifest=manifest,
agent_name="demo",
copy_cwd=copy_cwd,
user_cwd=user_cwd,
)
class TestWorkspacePlan(unittest.TestCase):
def test_disabled_uses_guest_home_as_workdir(self):
plan = workspace_plan(
_spec(copy_cwd=False, user_cwd="/tmp/project"),
guest_home="/home/node",
)
self.assertFalse(plan.enabled)
self.assertEqual("/home/node", plan.guest_path)
self.assertEqual("/home/node", plan.workdir)
def test_enabled_uses_workspace_under_guest_home(self):
plan = workspace_plan(
_spec(copy_cwd=True, user_cwd="/tmp/project"),
guest_home="/guest/home",
)
self.assertTrue(plan.enabled)
self.assertEqual(Path("/tmp/project"), plan.host_path)
self.assertEqual("/guest/home/workspace", plan.guest_path)
self.assertEqual("/guest/home/workspace", plan.workdir)
def test_detects_host_git_dir(self):
with tempfile.TemporaryDirectory(prefix="bb-workspace.") as tmp:
Path(tmp, ".git").mkdir()
plan = workspace_plan(
_spec(copy_cwd=True, user_cwd=tmp),
guest_home="/home/node",
)
self.assertTrue(plan.has_host_git_dir)
if __name__ == "__main__":
unittest.main()