didericis-claude
e6cafa39e0
Cast Die.code (_ExitCode) to int before passing to BottleError — Die always holds an int but SystemExit.code is typed as str|int|None in typeshed. Replace untyped lambda with str() as identity for _uniquify_label_headless mock (fixes reportUnknownLambdaType). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
268 lines
10 KiB
Python
268 lines
10 KiB
Python
"""Unit: bot_bottle public Python API (bot_bottle/__init__.py surface).
|
|
|
|
Covers start_headless, resume_headless, freeze, and destroy — the four
|
|
operations the bot-bottle-orchestrator's ProgrammaticBottleRunner uses.
|
|
All I/O is stubbed so no container is created.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bot_bottle import BottleError, destroy, freeze, resume_headless, start_headless
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_manifest(agent_name: str = "implementer", bottle_name: str = "claude"):
|
|
manifest = MagicMock()
|
|
manifest.agents = {agent_name: MagicMock(bottle=bottle_name)}
|
|
manifest.all_agent_names = [agent_name]
|
|
manifest.all_bottle_names = [bottle_name]
|
|
manifest.home_md = None # eager mode — _peek_agent_bottle uses agents dict
|
|
manifest.require_agent = MagicMock(return_value=None)
|
|
return manifest
|
|
|
|
|
|
def _metadata(
|
|
slug: str = "implementer-abc12",
|
|
agent_name: str = "implementer",
|
|
backend: str = "docker",
|
|
):
|
|
md = MagicMock()
|
|
md.identity = slug
|
|
md.agent_name = agent_name
|
|
md.cwd = "/repo"
|
|
md.copy_cwd = False
|
|
md.bottle_names = ["claude"]
|
|
md.backend = backend
|
|
return md
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# start_headless
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStartHeadless(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._manifest = _make_manifest()
|
|
patch("bot_bottle.api.ManifestIndex.resolve", return_value=self._manifest).start()
|
|
self._launch = patch(
|
|
"bot_bottle.api._launch_bottle", return_value=("implementer-abc12", 0)
|
|
).start()
|
|
patch(
|
|
"bot_bottle.api._uniquify_label_headless", side_effect=str
|
|
).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def _spec(self):
|
|
self._launch.assert_called_once()
|
|
return self._launch.call_args[0][0]
|
|
|
|
def test_returns_slug_on_success(self):
|
|
slug = start_headless("implementer", prompt="Do it")
|
|
self.assertEqual("implementer-abc12", slug)
|
|
|
|
def test_passes_assume_yes_and_prompt(self):
|
|
start_headless("implementer", prompt="Do it")
|
|
kwargs = self._launch.call_args[1]
|
|
self.assertTrue(kwargs["assume_yes"])
|
|
self.assertEqual("Do it", kwargs["headless_prompt_text"])
|
|
|
|
def test_explicit_bottles_forwarded(self):
|
|
start_headless("implementer", prompt="Do it", bottles=["dev", "claude"])
|
|
self.assertEqual(("dev", "claude"), self._spec().bottle_names)
|
|
|
|
def test_default_bottle_resolved_from_manifest(self):
|
|
start_headless("implementer", prompt="Do it")
|
|
self.assertEqual(("claude",), self._spec().bottle_names)
|
|
|
|
def test_forge_env_on_spec(self):
|
|
env = {"FORGE_GITEA_API": "https://gitea.example.com/api/v1", "FORGE_OWNER": "acme"}
|
|
start_headless("implementer", prompt="Do it", forge_env=env)
|
|
self.assertEqual(env, self._spec().forge_env)
|
|
|
|
def test_no_forge_env_defaults_to_empty_dict(self):
|
|
start_headless("implementer", prompt="Do it")
|
|
self.assertEqual({}, self._spec().forge_env)
|
|
|
|
def test_nonzero_exit_raises_bottle_error(self):
|
|
self._launch.return_value = ("implementer-abc12", 1)
|
|
with self.assertRaises(BottleError) as ctx:
|
|
start_headless("implementer", prompt="Do it")
|
|
self.assertEqual(1, ctx.exception.exit_code)
|
|
|
|
def test_no_default_bottle_raises_bottle_error(self):
|
|
manifest = _make_manifest(bottle_name="")
|
|
with patch("bot_bottle.api.ManifestIndex.resolve", return_value=manifest):
|
|
with self.assertRaises(BottleError):
|
|
start_headless("implementer", prompt="Do it")
|
|
self._launch.assert_not_called()
|
|
|
|
def test_backend_name_forwarded(self):
|
|
start_headless("implementer", prompt="Do it", backend_name="docker")
|
|
self.assertEqual("docker", self._launch.call_args[1]["backend_name"])
|
|
|
|
def test_label_forwarded_to_spec(self):
|
|
start_headless("implementer", prompt="Do it", label="nightly")
|
|
self.assertEqual("nightly", self._spec().label)
|
|
|
|
def test_color_forwarded_to_spec(self):
|
|
start_headless("implementer", prompt="Do it", color="green")
|
|
self.assertEqual("green", self._spec().color)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# resume_headless
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestResumeHeadless(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._md = _metadata()
|
|
patch("bot_bottle.api.read_metadata", return_value=self._md).start()
|
|
manifest = _make_manifest()
|
|
patch("bot_bottle.api.ManifestIndex.resolve", return_value=manifest).start()
|
|
self._launch = patch(
|
|
"bot_bottle.api._launch_bottle", return_value=("implementer-abc12", 0)
|
|
).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def _spec(self):
|
|
self._launch.assert_called_once()
|
|
return self._launch.call_args[0][0]
|
|
|
|
def test_passes_assume_yes_and_prompt(self):
|
|
resume_headless("implementer-abc12", prompt="Address review")
|
|
kwargs = self._launch.call_args[1]
|
|
self.assertTrue(kwargs["assume_yes"])
|
|
self.assertEqual("Address review", kwargs["headless_prompt_text"])
|
|
|
|
def test_identity_set_on_spec(self):
|
|
resume_headless("implementer-abc12", prompt="Prompt")
|
|
self.assertEqual("implementer-abc12", self._spec().identity)
|
|
|
|
def test_forge_env_on_spec(self):
|
|
env = {"FORGE_ISSUE_NUMBER": "42"}
|
|
resume_headless("implementer-abc12", prompt="Prompt", forge_env=env)
|
|
self.assertEqual(env, self._spec().forge_env)
|
|
|
|
def test_missing_state_raises_bottle_error(self):
|
|
with patch("bot_bottle.api.read_metadata", return_value=None):
|
|
with self.assertRaises(BottleError):
|
|
resume_headless("no-such-abc12", prompt="Prompt")
|
|
self._launch.assert_not_called()
|
|
|
|
def test_nonzero_exit_raises_bottle_error(self):
|
|
self._launch.return_value = ("implementer-abc12", 2)
|
|
with self.assertRaises(BottleError) as ctx:
|
|
resume_headless("implementer-abc12", prompt="Prompt")
|
|
self.assertEqual(2, ctx.exception.exit_code)
|
|
|
|
def test_backend_from_metadata_when_not_supplied(self):
|
|
resume_headless("implementer-abc12", prompt="Prompt")
|
|
self.assertEqual("docker", self._launch.call_args[1]["backend_name"])
|
|
|
|
def test_explicit_backend_overrides_metadata(self):
|
|
resume_headless(
|
|
"implementer-abc12", prompt="Prompt", backend_name="smolmachines"
|
|
)
|
|
self.assertEqual("smolmachines", self._launch.call_args[1]["backend_name"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# freeze
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFreeze(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
patch("bot_bottle.api.read_metadata", return_value=_metadata()).start()
|
|
self._freezer = MagicMock()
|
|
self._get_freezer = patch(
|
|
"bot_bottle.api.get_freezer", return_value=self._freezer
|
|
).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def test_calls_commit_slug(self):
|
|
freeze("implementer-abc12")
|
|
self._freezer.commit_slug.assert_called_once_with("implementer-abc12")
|
|
|
|
def test_backend_from_metadata_when_not_supplied(self):
|
|
freeze("implementer-abc12")
|
|
self._get_freezer.assert_called_once_with("docker")
|
|
|
|
def test_explicit_backend_used(self):
|
|
freeze("implementer-abc12", backend_name="smolmachines")
|
|
self._get_freezer.assert_called_once_with("smolmachines")
|
|
|
|
def test_commit_cancelled_raises_bottle_error(self):
|
|
from bot_bottle.backend.freeze import CommitCancelled
|
|
self._freezer.commit_slug.side_effect = CommitCancelled("declined")
|
|
with self.assertRaises(BottleError):
|
|
freeze("implementer-abc12")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# destroy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDestroy(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
patch("bot_bottle.api.read_metadata", return_value=_metadata()).start()
|
|
self._dd = patch("bot_bottle.api._destroy_docker").start()
|
|
patch("bot_bottle.api.clear_preserve_marker").start()
|
|
self._cleanup = patch("bot_bottle.api.cleanup_state").start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def test_docker_backend_calls_destroy_docker(self):
|
|
destroy("implementer-abc12")
|
|
self._dd.assert_called_once_with("implementer-abc12")
|
|
|
|
def test_state_dir_always_cleaned(self):
|
|
destroy("implementer-abc12")
|
|
self._cleanup.assert_called_once_with("implementer-abc12")
|
|
|
|
def test_smolmachines_backend_calls_destroy_smolmachines(self):
|
|
patch(
|
|
"bot_bottle.api.read_metadata",
|
|
return_value=_metadata(backend="smolmachines"),
|
|
).start()
|
|
ds = patch("bot_bottle.api._destroy_smolmachines").start()
|
|
destroy("implementer-abc12")
|
|
ds.assert_called_once_with("implementer-abc12")
|
|
self._dd.assert_not_called()
|
|
|
|
def test_missing_metadata_defaults_to_docker(self):
|
|
patch("bot_bottle.api.read_metadata", return_value=None).start()
|
|
destroy("no-state-abc12")
|
|
self._dd.assert_called_once_with("no-state-abc12")
|
|
|
|
def test_explicit_backend_overrides_metadata(self):
|
|
ds = patch("bot_bottle.api._destroy_smolmachines").start()
|
|
destroy("implementer-abc12", backend_name="smolmachines")
|
|
ds.assert_called_once_with("implementer-abc12")
|
|
self._dd.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# public surface exported from bot_bottle.__init__
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPublicSurface(unittest.TestCase):
|
|
def test_importable_from_package(self):
|
|
import bot_bottle
|
|
for name in ("BottleError", "start_headless", "resume_headless", "freeze", "destroy"):
|
|
self.assertTrue(hasattr(bot_bottle, name), f"missing: {name}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|