bot-bottle/tests/unit/test_manifest_bottle_merge.py

"""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_nested_containers_survives_a_later_bottle_that_omits_the_key(self):
        """A bottle that never mentions nested_containers must not silently
        drop the capability: `--bottle with-docker --bottle claude-dev`."""
        enabled = _bottle(nested_containers=True)
        quiet = _bottle(env={"X": "1"})
        self.assertTrue(merge_bottles_runtime([enabled, quiet]).nested_containers)
        self.assertTrue(merge_bottles_runtime([quiet, enabled]).nested_containers)
        self.assertFalse(merge_bottles_runtime([quiet, quiet]).nested_containers)

    def test_nested_containers_explicit_false_overrides_earlier_true(self):
        """An explicit nested_containers: false in a later bottle must win
        over an earlier true (last-wins, presence-aware)."""
        enabled = _bottle(nested_containers=True)
        disabled = _bottle(nested_containers=False)
        self.assertFalse(merge_bottles_runtime([enabled, disabled]).nested_containers)

    def test_nested_containers_explicit_true_overrides_earlier_false(self):
        enabled = _bottle(nested_containers=True)
        disabled = _bottle(nested_containers=False)
        self.assertTrue(merge_bottles_runtime([disabled, enabled]).nested_containers)

    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()