import sys import tempfile import types import unittest from unittest.mock import MagicMock, patch docker_stub = types.ModuleType("docker") docker_stub.errors = types.SimpleNamespace( ImageNotFound=type("ImageNotFound", (Exception,), {}), NotFound=type("NotFound", (Exception,), {}), ) sys.modules.setdefault("docker", docker_stub) from core.docker_manager import BotDockerManager class BotDockerManagerTests(unittest.TestCase): def setUp(self) -> None: self._tmpdir = tempfile.TemporaryDirectory() def tearDown(self) -> None: self._tmpdir.cleanup() def _make_manager(self) -> BotDockerManager: manager = BotDockerManager.__new__(BotDockerManager) manager.client = MagicMock() manager.host_data_root = self._tmpdir.name manager.base_image = "nanobot-base" manager.network_name = "" manager.active_monitors = {} manager._last_delivery_error = {} manager._storage_limit_supported = True manager._storage_limit_warning_emitted = False return manager @staticmethod def _build_container( *, status: str, image: str, image_id: str | None = None, nano_cpus: int, memory_bytes: int, storage_opt_size: str, source_mount: str, network_name: str, bootstrap_label: str | None = "env-json-v1", ) -> MagicMock: actual_image_id = image_id or image container = MagicMock() container.status = status container.reload = MagicMock() container.start = MagicMock() container.stop = MagicMock() container.remove = MagicMock() container.image = types.SimpleNamespace(id=actual_image_id) container.attrs = { "Image": actual_image_id, "Config": { "Image": image, "Labels": ( {"dashboard.runtime_bootstrap": bootstrap_label} if bootstrap_label is not None else {} ), }, "HostConfig": { "NanoCpus": nano_cpus, "Memory": memory_bytes, "StorageOpt": {"size": storage_opt_size}, }, "Mounts": [ { "Source": source_mount, "Destination": "/root/.nanobot", "RW": True, } ], "NetworkSettings": { "Networks": {network_name: {"IPAddress": "172.18.0.2"}}, }, } return container def test_stop_bot_keeps_container_by_default(self) -> None: manager = self._make_manager() container = MagicMock() container.status = "running" container.reload = MagicMock() container.stop = MagicMock() container.remove = MagicMock() manager.client.containers.get.return_value = container result = manager.stop_bot("demo") self.assertTrue(result) container.stop.assert_called_once_with(timeout=5) container.remove.assert_not_called() def test_stop_bot_remove_true_deletes_container(self) -> None: manager = self._make_manager() container = MagicMock() container.status = "exited" container.reload = MagicMock() container.stop = MagicMock() container.remove = MagicMock() manager.client.containers.get.return_value = container result = manager.stop_bot("demo", remove=True) self.assertTrue(result) container.stop.assert_not_called() container.remove.assert_called_once_with() def test_start_bot_reuses_compatible_stopped_container(self) -> None: manager = self._make_manager() image_tag = "nanobot-base:v1" image_id = "sha256:img-v1" workspace_mount = f"{self._tmpdir.name}/demo/.nanobot" container = self._build_container( status="exited", image=image_tag, image_id=image_id, nano_cpus=1_000_000_000, memory_bytes=1024 * 1024 * 1024, storage_opt_size="10G", source_mount=workspace_mount, network_name="bridge", ) manager.client.images.get.return_value = types.SimpleNamespace(id=image_id) manager.client.containers.get.return_value = container result = manager.start_bot( "demo", image_tag=image_tag, env_vars={"TZ": "UTC", "API_KEY": "updated-secret"}, cpu_cores=1.0, memory_mb=1024, storage_gb=10, ) self.assertTrue(result) container.start.assert_called_once_with() container.remove.assert_not_called() manager.client.containers.run.assert_not_called() def test_start_bot_recreates_incompatible_stopped_container(self) -> None: manager = self._make_manager() image_tag = "nanobot-base:v1" desired_image_id = "sha256:img-v1" workspace_mount = f"{self._tmpdir.name}/demo/.nanobot" container = self._build_container( status="exited", image="nanobot-base:old", image_id="sha256:img-old", nano_cpus=1_000_000_000, memory_bytes=1024 * 1024 * 1024, storage_opt_size="10G", source_mount=workspace_mount, network_name="bridge", ) manager.client.images.get.return_value = types.SimpleNamespace(id=desired_image_id) manager.client.containers.get.return_value = container manager._run_container_with_storage_fallback = MagicMock(return_value=MagicMock()) result = manager.start_bot( "demo", image_tag=image_tag, env_vars={"TZ": "Asia/Shanghai"}, cpu_cores=1.0, memory_mb=1024, storage_gb=10, ) self.assertTrue(result) container.start.assert_not_called() container.remove.assert_called_once_with(force=True) manager._run_container_with_storage_fallback.assert_called_once() def test_start_bot_recreates_stopped_container_when_image_id_changes_under_same_tag(self) -> None: manager = self._make_manager() image_tag = "nanobot-base:v1" workspace_mount = f"{self._tmpdir.name}/demo/.nanobot" container = self._build_container( status="exited", image=image_tag, image_id="sha256:img-old", nano_cpus=1_000_000_000, memory_bytes=1024 * 1024 * 1024, storage_opt_size="10G", source_mount=workspace_mount, network_name="bridge", ) manager.client.images.get.return_value = types.SimpleNamespace(id="sha256:img-new") manager.client.containers.get.return_value = container manager._run_container_with_storage_fallback = MagicMock(return_value=MagicMock()) result = manager.start_bot( "demo", image_tag=image_tag, env_vars={"TZ": "Asia/Shanghai"}, cpu_cores=1.0, memory_mb=1024, storage_gb=10, ) self.assertTrue(result) container.start.assert_not_called() container.remove.assert_called_once_with(force=True) manager._run_container_with_storage_fallback.assert_called_once() def test_start_bot_recreates_container_without_new_entrypoint(self) -> None: manager = self._make_manager() image_tag = "nanobot-base:v1" image_id = "sha256:img-v1" workspace_mount = f"{self._tmpdir.name}/demo/.nanobot" container = self._build_container( status="exited", image=image_tag, image_id=image_id, nano_cpus=1_000_000_000, memory_bytes=1024 * 1024 * 1024, storage_opt_size="10G", source_mount=workspace_mount, network_name="bridge", bootstrap_label=None, ) manager.client.images.get.return_value = types.SimpleNamespace(id=image_id) manager.client.containers.get.return_value = container manager._run_container_with_storage_fallback = MagicMock(return_value=MagicMock()) result = manager.start_bot( "demo", image_tag=image_tag, env_vars={"TZ": "Asia/Shanghai"}, cpu_cores=1.0, memory_mb=1024, storage_gb=10, ) self.assertTrue(result) container.start.assert_not_called() container.remove.assert_called_once_with(force=True) manager._run_container_with_storage_fallback.assert_called_once() def test_start_bot_recreates_running_container_when_image_id_changes_under_same_tag(self) -> None: manager = self._make_manager() image_tag = "nanobot-base:v1" workspace_mount = f"{self._tmpdir.name}/demo/.nanobot" container = self._build_container( status="running", image=image_tag, image_id="sha256:img-old", nano_cpus=1_000_000_000, memory_bytes=1024 * 1024 * 1024, storage_opt_size="10G", source_mount=workspace_mount, network_name="bridge", ) manager.client.images.get.return_value = types.SimpleNamespace(id="sha256:img-new") manager.client.containers.get.return_value = container manager._run_container_with_storage_fallback = MagicMock(return_value=MagicMock()) result = manager.start_bot( "demo", image_tag=image_tag, env_vars={"TZ": "Asia/Shanghai"}, cpu_cores=1.0, memory_mb=1024, storage_gb=10, ) self.assertTrue(result) container.remove.assert_called_once_with(force=True) manager._run_container_with_storage_fallback.assert_called_once() def test_send_command_waits_for_dashboard_ready(self) -> None: manager = self._make_manager() manager._wait_for_dashboard_ready = MagicMock(return_value=True) manager._send_command_via_exec = MagicMock(return_value=True) result = manager.send_command("demo", "hello") self.assertTrue(result) manager._wait_for_dashboard_ready.assert_called_once_with("demo") manager._send_command_via_exec.assert_called_once_with("demo", "hello", []) def test_send_command_returns_false_when_dashboard_never_becomes_ready(self) -> None: manager = self._make_manager() def _wait_timeout(bot_id: str) -> bool: manager._last_delivery_error[bot_id] = "Dashboard channel was not ready within 15s" return False manager._wait_for_dashboard_ready = MagicMock(side_effect=_wait_timeout) manager._send_command_via_exec = MagicMock() manager._send_command_via_host_http = MagicMock() result = manager.send_command("demo", "hello") self.assertFalse(result) manager._send_command_via_exec.assert_not_called() manager._send_command_via_host_http.assert_not_called() self.assertEqual( manager.get_last_delivery_error("demo"), "Dashboard channel was not ready within 15s", ) def test_wait_for_dashboard_ready_returns_true_after_start_log(self) -> None: manager = self._make_manager() manager.get_bot_status = MagicMock(return_value="RUNNING") manager.get_recent_logs = MagicMock( side_effect=[ ["Agent loop started"], ["2026-04-25 | INFO | nanobot.channels.dashboard:start:66 - ready"], ] ) with patch("core.docker_manager.time.sleep", return_value=None): result = manager._wait_for_dashboard_ready( "demo", timeout_seconds=2.0, poll_interval_seconds=0.1, ) self.assertTrue(result) def test_wait_for_dashboard_ready_sets_timeout_error(self) -> None: manager = self._make_manager() manager.get_bot_status = MagicMock(return_value="RUNNING") manager.get_recent_logs = MagicMock(return_value=["Agent loop started"]) time_values = iter([0.0, 0.2, 0.4, 1.2]) with ( patch("core.docker_manager.time.monotonic", side_effect=lambda: next(time_values)), patch("core.docker_manager.time.sleep", return_value=None), ): result = manager._wait_for_dashboard_ready( "demo", timeout_seconds=1.0, poll_interval_seconds=0.1, ) self.assertFalse(result) self.assertEqual( manager.get_last_delivery_error("demo"), "Dashboard channel was not ready within 1s", ) if __name__ == "__main__": unittest.main()