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:
@@ -1,219 +0,0 @@
|
||||
"""Integration: drive `apply_capability_change` against a real
|
||||
container that mimics the agent's name + filesystem layout (PRD 0016).
|
||||
|
||||
The real `cli.py start <agent>` flow is too heavy for an integration
|
||||
test (it builds the agent image, brings up all the sidecars, attaches
|
||||
an interactive agent session). Instead, this test stages the
|
||||
minimum the orchestrator interacts with:
|
||||
|
||||
- A lightweight `alpine:latest sleep infinity` container named
|
||||
`bot-bottle-<slug>` (matches the agent container name pattern)
|
||||
on the per-bottle internal network.
|
||||
- A marker file under `/home/node/.claude/` so we can assert the
|
||||
transcript snapshot path actually transferred bytes.
|
||||
|
||||
Then `apply_capability_change` runs and we verify:
|
||||
- Per-bottle Dockerfile written.
|
||||
- Containers + networks removed.
|
||||
- Transcript snapshot dir on the host has the marker file.
|
||||
|
||||
docker exec / cp / rm work across the docker socket boundary, so
|
||||
this test runs in DinD too — no act_runner skip needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from bot_bottle import supervise
|
||||
from bot_bottle import bottle_state
|
||||
from bot_bottle.backend.docker.capability_apply import apply_capability_change
|
||||
from bot_bottle.backend.docker.network import (
|
||||
network_create_egress,
|
||||
network_create_internal,
|
||||
network_remove,
|
||||
)
|
||||
from bot_bottle.backend.docker.sidecar_bundle import (
|
||||
sidecar_bundle_container_name,
|
||||
)
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
|
||||
ALPINE_IMAGE = "alpine:latest"
|
||||
|
||||
|
||||
@skip_unless_docker()
|
||||
class TestCapabilityApply(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
r = subprocess.run(
|
||||
["docker", "pull", ALPINE_IMAGE],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
raise unittest.SkipTest(f"could not pull {ALPINE_IMAGE}")
|
||||
|
||||
def setUp(self):
|
||||
self.slug = f"cb-test-cap-{os.getpid()}-{int(time.time())}"
|
||||
self.agent_name = f"bot-bottle-{self.slug}"
|
||||
self.sidecar_names: list[str] = []
|
||||
self.internal_net = ""
|
||||
self.egress_net = ""
|
||||
# Fake home so tests don't touch ~/.bot-bottle/.
|
||||
self._tmp = tempfile.TemporaryDirectory(prefix="cap-apply-int.")
|
||||
self._original_root = supervise.bot_bottle_root
|
||||
|
||||
def fake_root() -> Path:
|
||||
return Path(self._tmp.name) / ".bot-bottle"
|
||||
|
||||
supervise.bot_bottle_root = fake_root # type: ignore[assignment]
|
||||
|
||||
def tearDown(self):
|
||||
supervise.bot_bottle_root = self._original_root # type: ignore[assignment]
|
||||
for name in [self.agent_name, *self.sidecar_names]:
|
||||
subprocess.run(
|
||||
["docker", "rm", "-f", name],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
|
||||
)
|
||||
for n in (self.internal_net, self.egress_net):
|
||||
if n:
|
||||
network_remove(n)
|
||||
self._tmp.cleanup()
|
||||
|
||||
def _bring_up_fake_bottle(self) -> None:
|
||||
self.internal_net = network_create_internal(self.slug)
|
||||
self.egress_net = network_create_egress(self.slug)
|
||||
# Agent container with the canonical name.
|
||||
r = subprocess.run(
|
||||
[
|
||||
"docker", "run", "-d",
|
||||
"--name", self.agent_name,
|
||||
"--network", self.internal_net,
|
||||
ALPINE_IMAGE,
|
||||
"sh", "-c",
|
||||
"mkdir -p /home/node/.claude && "
|
||||
"echo 'transcript-marker' > /home/node/.claude/sessions.json && "
|
||||
"sleep 3600",
|
||||
],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
self.assertEqual(0, r.returncode, r.stderr)
|
||||
# Also start a fake sidecar bundle so teardown has something
|
||||
# extra to clean up (mirrors a real bottle's container set).
|
||||
sidecar = sidecar_bundle_container_name(self.slug)
|
||||
subprocess.run(
|
||||
[
|
||||
"docker", "run", "-d",
|
||||
"--name", sidecar,
|
||||
"--network", self.internal_net,
|
||||
ALPINE_IMAGE, "sleep", "3600",
|
||||
],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
self.sidecar_names.append(sidecar)
|
||||
|
||||
def _containers_named_like(self) -> list[str]:
|
||||
"""All running/stopped containers whose names start with
|
||||
the bottle's slug — both agent + sidecars."""
|
||||
r = subprocess.run(
|
||||
[
|
||||
"docker", "ps", "-a",
|
||||
"--filter", f"name={self.agent_name}",
|
||||
"--format", "{{.Names}}",
|
||||
],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
return [line for line in (r.stdout or "").splitlines() if line]
|
||||
|
||||
def _networks_named_like(self) -> list[str]:
|
||||
r = subprocess.run(
|
||||
[
|
||||
"docker", "network", "ls",
|
||||
"--filter", f"name={self.slug}",
|
||||
"--format", "{{.Name}}",
|
||||
],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
return [line for line in (r.stdout or "").splitlines() if line]
|
||||
|
||||
def test_apply_writes_dockerfile_and_tears_down(self):
|
||||
self._bring_up_fake_bottle()
|
||||
self.assertIn(self.agent_name, self._containers_named_like())
|
||||
|
||||
new_dockerfile = "FROM python:3.13\nRUN apk add ripgrep\n"
|
||||
before, after = apply_capability_change(self.slug, new_dockerfile)
|
||||
|
||||
# Before is the repo Dockerfile (no prior per-bottle override);
|
||||
# after is what we passed in.
|
||||
self.assertIn("FROM ", before)
|
||||
self.assertEqual(new_dockerfile, after)
|
||||
|
||||
# Per-bottle Dockerfile written on the host.
|
||||
self.assertEqual(
|
||||
new_dockerfile,
|
||||
bottle_state.per_bottle_dockerfile(self.slug),
|
||||
)
|
||||
|
||||
# Agent + sidecars gone.
|
||||
self.assertEqual([], self._containers_named_like())
|
||||
# Networks removed (matching the slug substring).
|
||||
nets = self._networks_named_like()
|
||||
self.assertEqual([], nets)
|
||||
# Mark them as already cleaned so tearDown is idempotent.
|
||||
self.internal_net = ""
|
||||
self.egress_net = ""
|
||||
self.sidecar_names = []
|
||||
|
||||
def test_transcript_snapshot_captured(self):
|
||||
self._bring_up_fake_bottle()
|
||||
apply_capability_change(self.slug, "FROM x\n")
|
||||
snap = bottle_state.transcript_snapshot_dir(self.slug)
|
||||
self.assertTrue(snap.is_dir(), f"transcript snapshot dir {snap} missing")
|
||||
# docker cp <container>:/home/node/.claude <dst> produces
|
||||
# <dst>/.claude/sessions.json (it preserves the source dir name
|
||||
# inside the destination if the destination already exists).
|
||||
# Walk the snapshot looking for the marker contents.
|
||||
marker_found = False
|
||||
for path in snap.rglob("sessions.json"):
|
||||
if "transcript-marker" in path.read_text():
|
||||
marker_found = True
|
||||
break
|
||||
self.assertTrue(marker_found, f"marker not found under {snap}")
|
||||
# Cleaned up by apply already.
|
||||
self.internal_net = ""
|
||||
self.egress_net = ""
|
||||
self.sidecar_names = []
|
||||
|
||||
def test_subsequent_apply_uses_per_bottle_dockerfile_for_before(self):
|
||||
# First change: before is repo's Dockerfile.
|
||||
self._bring_up_fake_bottle()
|
||||
first_before, _ = apply_capability_change(self.slug, "FROM v1\n")
|
||||
self.assertIn("FROM ", first_before)
|
||||
|
||||
# Second change: before is "FROM v1\n" (the per-bottle override
|
||||
# from the first change), proving the state persists across
|
||||
# rebuilds.
|
||||
self._bring_up_fake_bottle()
|
||||
second_before, second_after = apply_capability_change(self.slug, "FROM v2\n")
|
||||
self.assertEqual("FROM v1\n", second_before)
|
||||
self.assertEqual("FROM v2\n", second_after)
|
||||
self.internal_net = ""
|
||||
self.egress_net = ""
|
||||
self.sidecar_names = []
|
||||
|
||||
def test_teardown_idempotent_when_nothing_running(self):
|
||||
# No bottle ever brought up — teardown still doesn't raise.
|
||||
apply_capability_change(self.slug, "FROM x\n")
|
||||
self.assertEqual(
|
||||
"FROM x\n",
|
||||
bottle_state.per_bottle_dockerfile(self.slug),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
@@ -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):
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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()
|
||||
@@ -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__":
|
||||
|
||||
@@ -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__":
|
||||
|
||||
@@ -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__":
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user