bb21c294b4
- manifest.py: remove unused load_bottle_chain_from_dir import - manifest_extends.py: drop redundant ManifestEgressRoute annotation - test_cli_start_selector.py: remove unused call import - test_cli_tui.py: move Optional/constants to top, annotate FakeScreen, remove unused curses import - test_manifest_bottle_merge.py: add type args to dict, annotate **kwargs
201 lines
7.6 KiB
Python
201 lines
7.6 KiB
Python
"""Unit: runtime bottle composition (issue #269).
|
|
|
|
Tests for merge_bottles_runtime and ManifestIndex.load_for_agent with
|
|
the new bottle_names parameter.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from bot_bottle.manifest import ManifestBottle, ManifestError, ManifestIndex
|
|
from bot_bottle.manifest_extends import merge_bottles_runtime
|
|
|
|
|
|
def _index(bottles: dict[str, object], agents: dict[str, object]) -> ManifestIndex:
|
|
return ManifestIndex.from_json_obj({"bottles": bottles, "agents": agents})
|
|
|
|
|
|
def _bottle(**kwargs: object) -> ManifestBottle:
|
|
return ManifestBottle.from_dict("test", kwargs)
|
|
|
|
|
|
class TestMergeBottlesRuntime(unittest.TestCase):
|
|
def test_single_bottle_returns_as_is(self):
|
|
b = _bottle(env={"FOO": "1"})
|
|
result = merge_bottles_runtime([b])
|
|
self.assertEqual({"FOO": "1"}, dict(result.env))
|
|
|
|
def test_env_later_wins(self):
|
|
base = _bottle(env={"FOO": "base", "ONLY_BASE": "x"})
|
|
override = _bottle(env={"FOO": "override", "ONLY_OVERRIDE": "y"})
|
|
result = merge_bottles_runtime([base, override])
|
|
self.assertEqual("override", result.env["FOO"])
|
|
self.assertEqual("x", result.env["ONLY_BASE"])
|
|
self.assertEqual("y", result.env["ONLY_OVERRIDE"])
|
|
|
|
def test_egress_routes_concatenated(self):
|
|
from bot_bottle.manifest_egress import ManifestEgressConfig, ManifestEgressRoute
|
|
r1 = ManifestEgressRoute(Host="api.a.com")
|
|
r2 = ManifestEgressRoute(Host="api.b.com")
|
|
base = ManifestBottle(egress=ManifestEgressConfig(routes=(r1,)))
|
|
override = ManifestBottle(egress=ManifestEgressConfig(routes=(r2,)))
|
|
result = merge_bottles_runtime([base, override])
|
|
hosts = [r.Host for r in result.egress.routes]
|
|
self.assertIn("api.a.com", hosts)
|
|
self.assertIn("api.b.com", hosts)
|
|
|
|
def test_supervise_later_wins(self):
|
|
base = _bottle(supervise=True)
|
|
override = _bottle(supervise=False)
|
|
result = merge_bottles_runtime([base, override])
|
|
self.assertFalse(result.supervise)
|
|
|
|
def test_three_bottles_merged_left_to_right(self):
|
|
b1 = _bottle(env={"A": "1", "B": "1", "C": "1"})
|
|
b2 = _bottle(env={"B": "2", "C": "2"})
|
|
b3 = _bottle(env={"C": "3"})
|
|
result = merge_bottles_runtime([b1, b2, b3])
|
|
self.assertEqual("1", result.env["A"])
|
|
self.assertEqual("2", result.env["B"])
|
|
self.assertEqual("3", result.env["C"])
|
|
|
|
def test_empty_list_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
merge_bottles_runtime([])
|
|
|
|
|
|
class TestLoadForAgentWithBottleNames(unittest.TestCase):
|
|
def test_bottle_names_override_agent_bottle(self):
|
|
idx = _index(
|
|
bottles={
|
|
"base": {"env": {"X": "base"}},
|
|
"override": {"env": {"X": "override"}},
|
|
},
|
|
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
|
|
)
|
|
m = idx.load_for_agent("impl", ("override",))
|
|
self.assertEqual("override", m.bottle.env["X"])
|
|
|
|
def test_bottle_names_merged_in_order(self):
|
|
idx = _index(
|
|
bottles={
|
|
"a": {"env": {"X": "a", "A": "only-a"}},
|
|
"b": {"env": {"X": "b", "B": "only-b"}},
|
|
},
|
|
agents={"impl": {"bottle": "a", "skills": [], "prompt": ""}},
|
|
)
|
|
m = idx.load_for_agent("impl", ("a", "b"))
|
|
self.assertEqual("b", m.bottle.env["X"])
|
|
self.assertEqual("only-a", m.bottle.env["A"])
|
|
self.assertEqual("only-b", m.bottle.env["B"])
|
|
|
|
def test_empty_bottle_names_uses_agent_bottle(self):
|
|
idx = _index(
|
|
bottles={"base": {"env": {"X": "base"}}},
|
|
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
|
|
)
|
|
m = idx.load_for_agent("impl", ())
|
|
self.assertEqual("base", m.bottle.env["X"])
|
|
|
|
def test_no_bottle_and_no_bottle_names_raises(self):
|
|
idx = _index(
|
|
bottles={"base": {}},
|
|
agents={"impl": {"skills": [], "prompt": ""}},
|
|
)
|
|
with self.assertRaises(ManifestError) as ctx:
|
|
idx.load_for_agent("impl", ())
|
|
self.assertIn("no 'bottle' field", str(ctx.exception))
|
|
|
|
def test_unknown_bottle_name_raises(self):
|
|
idx = _index(
|
|
bottles={"base": {}},
|
|
agents={"impl": {"bottle": "base", "skills": [], "prompt": ""}},
|
|
)
|
|
with self.assertRaises(ManifestError) as ctx:
|
|
idx.load_for_agent("impl", ("nonexistent",))
|
|
self.assertIn("nonexistent", str(ctx.exception))
|
|
|
|
def test_agent_without_bottle_works_with_bottle_names(self):
|
|
idx = _index(
|
|
bottles={"base": {"env": {"X": "base"}}},
|
|
agents={"impl": {"skills": [], "prompt": ""}},
|
|
)
|
|
m = idx.load_for_agent("impl", ("base",))
|
|
self.assertEqual("base", m.bottle.env["X"])
|
|
|
|
|
|
class TestAllBottleNames(unittest.TestCase):
|
|
def test_eager_mode_returns_bottle_names(self):
|
|
idx = _index(
|
|
bottles={"alpha": {}, "beta": {}, "gamma": {}},
|
|
agents={"impl": {"bottle": "alpha", "skills": [], "prompt": ""}},
|
|
)
|
|
self.assertEqual(["alpha", "beta", "gamma"], idx.all_bottle_names)
|
|
|
|
def test_lazy_mode_scans_files(self):
|
|
home = Path(tempfile.mkdtemp(prefix="cb-home-"))
|
|
orig_home = os.environ.get("HOME")
|
|
os.environ["HOME"] = str(home)
|
|
try:
|
|
bottles_dir = home / ".bot-bottle" / "bottles"
|
|
agents_dir = home / ".bot-bottle" / "agents"
|
|
bottles_dir.mkdir(parents=True)
|
|
agents_dir.mkdir(parents=True)
|
|
(bottles_dir / "claude.md").write_text("---\n---\n")
|
|
(bottles_dir / "dev.md").write_text("---\n---\n")
|
|
(agents_dir / "impl.md").write_text("---\nbottle: claude\n---\n")
|
|
idx = ManifestIndex.resolve(str(home))
|
|
self.assertEqual(["claude", "dev"], idx.all_bottle_names)
|
|
finally:
|
|
if orig_home is None:
|
|
os.environ.pop("HOME", None)
|
|
else:
|
|
os.environ["HOME"] = orig_home
|
|
shutil.rmtree(home, ignore_errors=True)
|
|
|
|
|
|
class TestAgentOptionalBottleMd(unittest.TestCase):
|
|
"""Agent file without bottle: works when bottle_names are provided at launch."""
|
|
|
|
def setUp(self) -> None:
|
|
self.home = Path(tempfile.mkdtemp(prefix="cb-home-"))
|
|
self._orig_home = os.environ.get("HOME")
|
|
os.environ["HOME"] = str(self.home)
|
|
|
|
def tearDown(self) -> None:
|
|
if self._orig_home is None:
|
|
os.environ.pop("HOME", None)
|
|
else:
|
|
os.environ["HOME"] = self._orig_home
|
|
shutil.rmtree(self.home, ignore_errors=True)
|
|
|
|
def _write(self, rel: str, text: str) -> None:
|
|
p = self.home / ".bot-bottle" / rel
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(textwrap.dedent(text).lstrip("\n"))
|
|
|
|
def test_agent_without_bottle_resolves_with_bottle_names(self):
|
|
self._write("bottles/dev.md", "---\nenv:\n X: dev\n---\n")
|
|
self._write("agents/impl.md", "---\n---\nimpl agent.\n")
|
|
idx = ManifestIndex.resolve(str(self.home))
|
|
m = idx.load_for_agent("impl", ("dev",))
|
|
self.assertEqual("dev", m.bottle.env["X"])
|
|
|
|
def test_agent_without_bottle_fails_without_bottle_names(self):
|
|
self._write("bottles/dev.md", "---\n---\n")
|
|
self._write("agents/impl.md", "---\n---\nimpl agent.\n")
|
|
idx = ManifestIndex.resolve(str(self.home))
|
|
with self.assertRaises(ManifestError) as ctx:
|
|
idx.load_for_agent("impl", ())
|
|
self.assertIn("no 'bottle' field", str(ctx.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|