"""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, agents: dict) -> ManifestIndex: return ManifestIndex.from_json_obj({"bottles": bottles, "agents": agents}) def _bottle(**kwargs) -> 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()