bot-bottle/tests/unit/test_smolmachines_sidecar_bundle.py

"""Unit: bundle bringup primitives for the smolmachines backend."""

from __future__ import annotations

import subprocess
import unittest
from pathlib import Path
from unittest.mock import patch

from bot_bottle.backend.smolmachines.sidecar_bundle import (
    BundleLaunchSpec,
    allocate_raw_host_ports,
    bundle_container_name,
    bundle_network_name,
    ensure_bundle_image,
    start_bundle_vm,
    stop_bundle_vm,
)


def _spec(**kwargs) -> BundleLaunchSpec:  # type: ignore
    defaults = dict(
        slug="demo-abc12",
        network_name="bot-bottle-bundle-demo-abc12",
        subnet="192.168.50.0/24",
        gateway="192.168.50.1",
        bundle_ip="192.168.50.2",
    )
    defaults.update(kwargs)
    return BundleLaunchSpec(**defaults)  # type: ignore


class TestNamingHelpers(unittest.TestCase):
    def test_network_name_uses_bundle_prefix(self):
        # Distinct from the docker backend's
        # `bot-bottle-net-<slug>` so two backends running the
        # same agent slug don't collide.
        self.assertEqual(
            "bot-bottle-bundle-myagent-xyz",
            bundle_network_name("myagent-xyz"),
        )

    def test_container_name_matches_docker_bundle_shape(self):
        # Same shape PRD 0024 chunk 5 set for the docker backend's
        # bundle container — dashboard prefix-discovery covers
        # both backends with one filter.
        self.assertEqual(
            "bot-bottle-sidecars-myagent-xyz",
            bundle_container_name("myagent-xyz"),
        )


class TestEnsureBundleImage(unittest.TestCase):
    def test_builds_sidecar_dockerfile_before_plain_docker_run(self):
        with patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle.docker_mod.build_image",
        ) as build:
            ensure_bundle_image()

        build.assert_called_once()
        args = build.call_args.args
        kwargs = build.call_args.kwargs
        self.assertEqual("bot-bottle-sidecars:latest", args[0])
        self.assertTrue((Path(args[1]) / "Dockerfile.sidecars").is_file())
        self.assertEqual("Dockerfile.sidecars", kwargs["dockerfile"])


class TestStartBundleVm(unittest.TestCase):
    def test_allocate_raw_host_ports_returns_bindable_ports(self):
        raw_ports = allocate_raw_host_ports((9099, 9420))

        self.assertEqual({9099, 9420}, set(raw_ports))
        for host_port in raw_ports.values():
            self.assertIsInstance(host_port, int)
            self.assertGreater(host_port, 0)
        self.assertNotEqual(raw_ports[9099], raw_ports[9420])

    def test_creates_sidecar_vm_with_ports_volumes_and_env(self):
        calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []

        def record(name):  # type: ignore
            def inner(*args, **kwargs):  # type: ignore
                calls.append((name, args, kwargs))
            return inner

        with patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle.allocate_raw_host_ports",
            return_value={9099: 55001, 9420: 55002},
        ), patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_create",
            side_effect=record("create"),
        ), patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_start",
            side_effect=record("start"),
        ), patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.wait_exec_ready",
            side_effect=record("wait"),
        ):
            launch = start_bundle_vm(
                _spec(
                    daemons_csv="egress,git-http",
                    environment=("FOO=bar", "TOKEN"),
                    volumes=(("/host/routes", "/etc/egress", True),),
                    ports_to_publish=(9099, 9420),
                ),
                from_path=Path("/cache/sidecar.smolmachine"),
                host_env={"TOKEN": "secret"},
            )

        self.assertEqual({9099: 55001, 9420: 55002}, launch.raw_ports)
        create = calls[0]
        self.assertEqual("create", create[0])
        self.assertEqual(("bot-bottle-sidecars-demo-abc12",), create[1])
        self.assertEqual(Path("/cache/sidecar.smolmachine"), create[2]["from_path"])
        self.assertTrue(create[2]["net"])
        self.assertEqual(((55001, 9099), (55002, 9420)), create[2]["ports"])
        self.assertEqual((("/host/routes", "/etc/egress", True),), create[2]["volumes"])
        self.assertEqual(
            {
                "BOT_BOTTLE_SIDECAR_DAEMONS": "egress,git-http",
                "FOO": "bar",
                "TOKEN": "secret",
            },
            create[2]["env"],
        )


class TestStopBundleVm(unittest.TestCase):
    def test_stops_then_deletes_sidecar_vm(self):
        with patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
        ) as stop, patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
        ) as delete:
            stop_bundle_vm("demo-abc12")
        stop.assert_called_once_with("bot-bottle-sidecars-demo-abc12")
        delete.assert_called_once_with("bot-bottle-sidecars-demo-abc12")

    def test_stop_and_delete_errors_are_warnings(self):
        from bot_bottle.backend.smolmachines.smolvm import SmolvmError
        result = subprocess.CompletedProcess(
            args=[],
            returncode=1,
            stdout="",
            stderr="failed",
        )
        with patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
            side_effect=SmolvmError(["smolvm", "machine", "stop"], result),
        ), patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
            side_effect=SmolvmError(["smolvm", "machine", "delete"], result),
        ), patch(
            "bot_bottle.backend.smolmachines.sidecar_bundle.warn",
        ) as warn:
            stop_bundle_vm("demo-abc12")

        self.assertEqual(2, warn.call_count)


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