bot-bottle/tests/unit/test_docker_launch_committed_image.py

"""Unit: Docker launch step uses committed image when available."""

from __future__ import annotations

import contextlib
import io
import tempfile
import unittest
from pathlib import Path
from typing import Any
from unittest import mock

from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import BottleSpec
from bot_bottle.backend.docker import launch as launch_mod
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 ManifestIndex


_SLUG = "dev-abc12"
_COMMITTED_TAG = f"bot-bottle-committed-{_SLUG}:latest"
_DEFAULT_IMAGE = "bot-bottle-claude:latest"

_IDX = ManifestIndex.from_json_obj({
    "bottles": {"dev": {}},
    "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})


def _plan(tmp: str) -> DockerBottlePlan:
    stage = Path(tmp)
    spec = BottleSpec(
        manifest=_IDX,
        agent_name="demo",
        copy_cwd=False,
        user_cwd=tmp,
        identity=_SLUG,
    )
    return DockerBottlePlan(
        spec=spec,
        manifest=_IDX.load_for_agent("demo"),
        stage_dir=stage,
        git_gate_plan=GitGatePlan(
            slug=_SLUG,
            entrypoint_script=stage / "entrypoint.sh",
            hook_script=stage / "hook.sh",
            access_hook_script=stage / "access-hook.sh",
            upstreams=(),
        ),
        egress_plan=EgressPlan(
            slug=_SLUG,
            routes_path=stage / "egress.yaml",
            routes=(),
            token_env_map={},
        ),
        supervise_plan=None,
        agent_provision=AgentProvisionPlan(
            template="claude",
            command="claude",
            prompt_mode="append_file",
            image=_DEFAULT_IMAGE,
            dockerfile="",
            guest_home="/home/node",
            instance_name=f"bot-bottle-{_SLUG}",
            prompt_file=stage / "prompt.txt",
            guest_env={},
        ),
        slug=_SLUG,
        forwarded_env={},
        use_runsc=False,
    )


class TestLaunchCommittedImage(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.mkdtemp(prefix="launch-committed-test.")

    def tearDown(self) -> None:
        import shutil
        shutil.rmtree(self._tmp, ignore_errors=True)

    def _run_launch(
        self,
        plan: DockerBottlePlan,
        *,
        committed_tag: str | None = None,
        image_present: bool = True,
    ) -> list[str]:
        """Drive launch() through its full sequence with the committed-image
        behaviour controlled by the arguments. Returns the images that were
        passed to `build_image` (empty list if it was never called)."""
        built: list[str] = []

        def fake_build(image: str, ctx: str, *, dockerfile: str = "") -> None:
            del ctx, dockerfile
            built.append(image)

        with mock.patch.object(
            launch_mod, "read_committed_image", return_value=committed_tag,
        ), mock.patch.object(
            launch_mod.docker_mod, "image_exists", return_value=image_present,
        ), mock.patch.object(
            launch_mod.docker_mod, "build_image", side_effect=fake_build,
        ), mock.patch.object(
            launch_mod, "egress_tls_init",
            return_value=(Path("/egress_ca"), Path("/egress_cert")),
        ), mock.patch.object(
            launch_mod.network_mod, "network_name_for_slug",
            return_value="bb-internal",
        ), mock.patch.object(
            launch_mod.network_mod, "network_egress_name_for_slug",
            return_value="bb-egress",
        ), mock.patch.object(
            launch_mod, "bottle_plan_to_compose",
            return_value={"services": {"agent": {}}},
        ), mock.patch.object(
            launch_mod, "write_compose_file",
            return_value=Path("/tmp/compose.yml"),
        ), mock.patch.object(launch_mod, "compose_up"), \
           mock.patch.object(launch_mod, "compose_dump_logs"), \
           mock.patch.object(launch_mod, "compose_down"), \
           contextlib.redirect_stderr(io.StringIO()):
            provision = mock.Mock(return_value=None)
            with launch_mod.launch(plan, provision=provision):
                pass

        return built

    def test_skips_build_when_committed_image_present(self) -> None:
        plan = _plan(self._tmp)
        built = self._run_launch(plan, committed_tag=_COMMITTED_TAG, image_present=True)
        self.assertEqual([], built, "build_image should not be called when committed image exists")

    def test_uses_committed_image_in_compose_spec(self) -> None:
        """The compose spec renderer receives the committed image tag via
        plan.image — captured here by checking what bottle_plan_to_compose
        was called with."""
        plan = _plan(self._tmp)
        captured_plans: list[DockerBottlePlan] = []

        def fake_compose(p: DockerBottlePlan) -> dict[str, Any]:
            captured_plans.append(p)
            return {"services": {"agent": {}}}

        with mock.patch.object(
            launch_mod, "read_committed_image", return_value=_COMMITTED_TAG,
        ), mock.patch.object(
            launch_mod.docker_mod, "image_exists", return_value=True,
        ), mock.patch.object(
            launch_mod.docker_mod, "build_image",
        ), mock.patch.object(
            launch_mod, "egress_tls_init",
            return_value=(Path("/egress_ca"), Path("/egress_cert")),
        ), mock.patch.object(
            launch_mod.network_mod, "network_name_for_slug",
            return_value="bb-internal",
        ), mock.patch.object(
            launch_mod.network_mod, "network_egress_name_for_slug",
            return_value="bb-egress",
        ), mock.patch.object(
            launch_mod, "bottle_plan_to_compose", side_effect=fake_compose,
        ), mock.patch.object(
            launch_mod, "write_compose_file",
            return_value=Path("/tmp/compose.yml"),
        ), mock.patch.object(launch_mod, "compose_up"), \
           mock.patch.object(launch_mod, "compose_dump_logs"), \
           mock.patch.object(launch_mod, "compose_down"), \
           contextlib.redirect_stderr(io.StringIO()):
            provision = mock.Mock(return_value=None)
            with launch_mod.launch(plan, provision=provision):
                pass

        self.assertEqual(1, len(captured_plans))
        self.assertEqual(_COMMITTED_TAG, captured_plans[0].image)

    def test_falls_back_to_build_when_no_committed_image(self) -> None:
        plan = _plan(self._tmp)
        built = self._run_launch(plan, committed_tag=None)
        self.assertEqual([_DEFAULT_IMAGE], built)

    def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
        plan = _plan(self._tmp)
        built = self._run_launch(
            plan, committed_tag=_COMMITTED_TAG, image_present=False,
        )
        self.assertEqual([_DEFAULT_IMAGE], built)


if __name__ == "__main__":
    unittest.main()