"""Unit: the macOS orchestrator (control plane) container lifecycle (PRD 0070).""" from __future__ import annotations import unittest from pathlib import Path from unittest.mock import Mock, patch from bot_bottle.backend.macos_container.orchestrator import ( ORCHESTRATOR_DB_VOLUME, ORCHESTRATOR_NAME, MacosOrchestrator, probe_orchestrator_url, ) from bot_bottle.orchestrator.lifecycle import OrchestratorStartError _ORCH = "bot_bottle.backend.macos_container.orchestrator" def _ok(stdout: str = "") -> Mock: return Mock(returncode=0, stdout=stdout, stderr="") def _fail(stderr: str = "boom") -> Mock: return Mock(returncode=1, stdout="", stderr=stderr) def _spec(src: str, tgt: str, readonly: bool = False) -> str: return f"type=bind,source={src},target={tgt}" + (",readonly" if readonly else "") class TestMacosOrchestratorRun(unittest.TestCase): def _run(self) -> list[str]: run = Mock(return_value=_ok()) with patch(f"{_ORCH}.container_mod") as mod, \ patch(f"{_ORCH}.host_orchestrator_token", return_value="k"): mod.dns_server.return_value = "1.1.1.1" mod.bind_mount_spec.side_effect = _spec mod.run_container_argv = run MacosOrchestrator(repo_root=Path("/r"))._run_container("h1") return run.call_args.args[0] def test_runs_on_the_control_network_only(self) -> None: argv = self._run() nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"] self.assertEqual(["bot-bottle-mac-control"], nets) # Image ENTRYPOINT is `-m bot_bottle.orchestrator`; these are its args. self.assertIn("--broker", argv) self.assertIn("stub", argv) self.assertIn("bot-bottle-orchestrator:latest", argv) def test_db_is_a_container_only_volume(self) -> None: argv = self._run() vols = [argv[i + 1] for i, a in enumerate(argv) if a == "--volume"] self.assertTrue(any(v.startswith(f"{ORCHESTRATOR_DB_VOLUME}:") for v in vols)) def test_no_ca_mount(self) -> None: # The CA lives with the gateway, not the control plane. argv = self._run() mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"] self.assertFalse([m for m in mounts if "/home/mitmproxy" in m]) def test_source_hash_is_labelled_for_recreate(self) -> None: self.assertIn("BOT_BOTTLE_SOURCE_HASH=h1", self._run()) def test_start_failure_raises(self) -> None: with patch(f"{_ORCH}.container_mod") as mod, \ patch(f"{_ORCH}.host_orchestrator_token", return_value="k"): mod.dns_server.return_value = "1.1.1.1" mod.run_container_argv = Mock(return_value=_fail()) with self.assertRaises(OrchestratorStartError): MacosOrchestrator(repo_root=Path("/r"))._run_container("h1") class TestMacosOrchestratorUrls(unittest.TestCase): def test_url_and_gateway_url_are_the_control_net_address(self) -> None: orch = MacosOrchestrator(port=8099) with patch(f"{_ORCH}.container_mod") as mod: mod.try_container_ipv4_on_network.return_value = "192.168.128.2" self.assertEqual("http://192.168.128.2:8099", orch.url()) self.assertEqual("http://192.168.128.2:8099", orch.gateway_url()) def test_url_empty_when_no_address(self) -> None: orch = MacosOrchestrator() with patch(f"{_ORCH}.container_mod") as mod: mod.try_container_ipv4_on_network.return_value = "" self.assertEqual("", orch.url()) def test_is_healthy_false_when_no_address(self) -> None: orch = MacosOrchestrator() with patch(f"{_ORCH}.container_mod") as mod: mod.try_container_ipv4_on_network.return_value = "" self.assertFalse(orch.is_healthy()) class TestMacosOrchestratorEnsureRunning(unittest.TestCase): def _orch(self) -> MacosOrchestrator: return MacosOrchestrator(repo_root=Path("/r")) def test_noop_when_current_and_healthy(self) -> None: orch = self._orch() with patch(f"{_ORCH}.source_hash", return_value="h1"), \ patch.object(orch, "_source_current", return_value=True), \ patch.object(orch, "is_healthy", return_value=True), \ patch.object(orch, "_run_container") as run: orch.ensure_running() run.assert_not_called() def test_recreates_when_source_changed(self) -> None: orch = self._orch() with patch(f"{_ORCH}.source_hash", return_value="h2"), \ patch.object(orch, "_source_current", return_value=False), \ patch.object(orch, "_run_container") as run, \ patch.object(orch, "_wait_healthy"): orch.ensure_running() run.assert_called_once() def test_recreates_when_current_but_wedged(self) -> None: orch = self._orch() with patch(f"{_ORCH}.source_hash", return_value="h1"), \ patch.object(orch, "_source_current", return_value=True), \ patch.object(orch, "is_healthy", return_value=False), \ patch.object(orch, "_run_container") as run, \ patch.object(orch, "_wait_healthy"): orch.ensure_running() run.assert_called_once() def test_never_healthy_raises(self) -> None: orch = self._orch() with patch(f"{_ORCH}.source_hash", return_value="h1"), \ patch.object(orch, "_source_current", return_value=False), \ patch.object(orch, "_run_container"), \ patch.object(orch, "is_healthy", return_value=False), \ patch(f"{_ORCH}.time.sleep"), \ patch(f"{_ORCH}.time.monotonic", side_effect=[0.0, 0.5, 2.0]): with self.assertRaises(OrchestratorStartError): orch.ensure_running(startup_timeout=1.0) class TestMacosOrchestratorBuildStop(unittest.TestCase): def test_ensure_built_builds_the_orchestrator_image(self) -> None: orch = MacosOrchestrator(repo_root=Path("/r")) with patch(f"{_ORCH}.container_mod") as mod: orch.ensure_built() _args, kwargs = mod.build_image.call_args self.assertEqual("Dockerfile.orchestrator", kwargs["dockerfile"]) def test_stop_removes_the_container(self) -> None: orch = MacosOrchestrator() with patch(f"{_ORCH}.container_mod") as mod: orch.stop() mod.force_remove_container.assert_called_once_with(ORCHESTRATOR_NAME) class TestProbeOrchestrator(unittest.TestCase): def test_returns_url_when_running(self) -> None: with patch(f"{_ORCH}.container_mod") as mod: mod.try_container_ipv4_on_network.return_value = "192.168.128.2" self.assertEqual("http://192.168.128.2:8099", probe_orchestrator_url()) def test_empty_when_absent(self) -> None: with patch(f"{_ORCH}.container_mod") as mod: mod.try_container_ipv4_on_network.return_value = "" self.assertEqual("", probe_orchestrator_url()) if __name__ == "__main__": unittest.main()