Merge main into fix/db-off-data-plane-469

Brings in main's 8 commits — #470 (backend-agnostic CI guards: BackendStatus
enum, quiet status(), is_backend_available/is_backend_ready, tests/_backend.py
skip_unless_backend) plus firecracker status() readiness checks (kvm/kernel/
dropbear/mke2fs).

Reconciled #470's additions into the reorg'd backend structure:
  * BackendStatus enum + the status(*, quiet=) ABC signature -> backend/base.py
  * is_backend_available / is_backend_ready -> backend/selection.py
  * exposed all three through the lazy backend facade (__init__).
Integration-test conflicts were our control_plane->orchestrator renames vs
main's skip_unless_docker -> skip_unless_backend swap: kept our refactored
import paths + main's guard. Repointed main's new is_backend_* tests to patch
selection._backends (our moved home) instead of the facade.

pyright: 0 errors. Full unit suite green (2272).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/unit/test_backend_selection.py

@@ -386,6 +386,60 @@ class TestHasBackend(unittest.TestCase):
         self.assertFalse(has_backend("nonexistent"))
 
 
+class TestIsBackendAvailable(unittest.TestCase):
+    def test_delegates_to_has_backend(self):
+        from bot_bottle.backend import is_backend_available
+        with patch.object(_sel, "_backends", {}):
+            self.assertFalse(is_backend_available("docker"))
+
+    def test_known_and_available(self):
+        class _Ready:
+            def is_available(self):
+                return True
+
+        from bot_bottle.backend import is_backend_available
+        with patch.object(_sel, "_backends", {"docker": _Ready()}):
+            self.assertTrue(is_backend_available("docker"))
+
+
+class TestIsBackendReady(unittest.TestCase):
+    def test_unknown_backend_returns_false(self):
+        from bot_bottle.backend import is_backend_ready
+        with patch.object(_sel, "_backends", {}):
+            self.assertFalse(is_backend_ready("docker"))
+
+    def test_ready_when_status_returns_zero(self):
+        class _ReadyBackend:
+            def status(self, *, quiet: bool = False) -> int:
+                return 0
+
+        from bot_bottle.backend import is_backend_ready
+        with patch.object(_sel, "_backends", {"docker": _ReadyBackend()}):
+            self.assertTrue(is_backend_ready("docker"))
+
+    def test_not_ready_when_status_nonzero(self):
+        class _BrokenBackend:
+            def status(self, *, quiet: bool = False) -> int:
+                return 1
+
+        from bot_bottle.backend import is_backend_ready
+        with patch.object(_sel, "_backends", {"docker": _BrokenBackend()}):
+            self.assertFalse(is_backend_ready("docker"))
+
+    def test_quiet_flag_forwarded(self):
+        calls = []
+
+        class _SpyBackend:
+            def status(self, *, quiet: bool = False) -> int:
+                calls.append(quiet)
+                return 0
+
+        from bot_bottle.backend import is_backend_ready
+        with patch.object(_sel, "_backends", {"docker": _SpyBackend()}):
+            is_backend_ready("docker", quiet=True)
+        self.assertEqual([True], calls)
+
+
 class TestEnsureOrchestrator(unittest.TestCase):
     """The backend-agnostic orchestrator bring-up entry point. Docker starts
     the orchestrator + gateway containers; firecracker boots the infra VM;

tests/unit/test_backend_setup.py

@@ -328,5 +328,201 @@ class TestNetpoolShellRenderers(unittest.TestCase):
         self.assertIn("BOT_BOTTLE_FC_POOL_SIZE=4", out)
 
 
+class TestFirecrackerBinaryCheck(unittest.TestCase):
+    def test_binary_missing_returns_false(self):
+        with patch.object(fc.shutil, "which", return_value=None):
+            self.assertFalse(fc._firecracker_binary_ok())
+
+    def test_binary_present_and_runs_ok(self):
+        with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
+             patch.object(fc.subprocess, "run",
+                          return_value=subprocess.CompletedProcess([], 0)):
+            self.assertTrue(fc._firecracker_binary_ok())
+
+    def test_binary_found_but_exits_nonzero(self):
+        with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
+             patch.object(fc.subprocess, "run",
+                          return_value=subprocess.CompletedProcess([], 1)):
+            self.assertFalse(fc._firecracker_binary_ok())
+
+    def test_binary_found_but_oserror(self):
+        with patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"), \
+             patch.object(fc.subprocess, "run", side_effect=OSError("exec failed")):
+            self.assertFalse(fc._firecracker_binary_ok())
+
+
+class TestFirecrackerKvmCheck(unittest.TestCase):
+    def test_kvm_device_absent_returns_false(self):
+        with patch.object(fc.os.path, "exists", return_value=False):
+            self.assertFalse(fc._kvm_accessible())
+
+    def test_kvm_accessible_when_ioctl_succeeds(self):
+        with patch.object(fc.os.path, "exists", return_value=True), \
+             patch.object(fc.os, "open", return_value=42), \
+             patch.object(fc.os, "close"), \
+             patch.object(fc.fcntl, "ioctl", return_value=12):
+            self.assertTrue(fc._kvm_accessible())
+
+    def test_kvm_open_rdwr_fails(self):
+        with patch.object(fc.os.path, "exists", return_value=True), \
+             patch.object(fc.os, "open", side_effect=OSError("Permission denied")):
+            self.assertFalse(fc._kvm_accessible())
+
+    def test_kvm_present_but_ioctl_fails(self):
+        with patch.object(fc.os.path, "exists", return_value=True), \
+             patch.object(fc.os, "open", return_value=42), \
+             patch.object(fc.os, "close"), \
+             patch.object(fc.fcntl, "ioctl", side_effect=OSError("permission denied")):
+            self.assertFalse(fc._kvm_accessible())
+
+
+class TestFirecrackerArtifactCheck(unittest.TestCase):
+    """status() reports missing guest kernel, dropbear binary, and mke2fs."""
+
+    def _apply_all_ok(self, stack: contextlib.ExitStack) -> None:
+        """Stub every status() check to pass except what the test overrides."""
+        stack.enter_context(patch.object(fc, "_firecracker_binary_ok", return_value=True))
+        stack.enter_context(patch.object(fc, "_kvm_accessible", return_value=True))
+        k: MagicMock = MagicMock()
+        k.is_file.return_value = True
+        stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
+        d: MagicMock = MagicMock()
+        d.is_file.return_value = True
+        stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
+        stack.enter_context(patch.object(fc.shutil, "which", return_value="/usr/bin/x"))
+        stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
+        stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
+        stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
+        stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
+
+    def test_status_fails_when_kernel_missing(self):
+        with contextlib.ExitStack() as stack:
+            self._apply_all_ok(stack)
+            k: MagicMock = MagicMock()
+            k.is_file.return_value = False
+            stack.enter_context(patch.object(fc.util, "kernel_path", return_value=k))
+            rc, out = _cap(fc.status)
+        self.assertEqual(1, rc)
+        self.assertIn("guest kernel: NOT found", out)
+
+    def test_status_fails_when_dropbear_missing(self):
+        with contextlib.ExitStack() as stack:
+            self._apply_all_ok(stack)
+            d: MagicMock = MagicMock()
+            d.is_file.return_value = False
+            stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=d))
+            rc, out = _cap(fc.status)
+        self.assertEqual(1, rc)
+        self.assertIn("dropbear: NOT found", out)
+
+    def test_status_fails_when_mke2fs_missing(self):
+        with contextlib.ExitStack() as stack:
+            self._apply_all_ok(stack)
+            stack.enter_context(patch.object(
+                fc.shutil, "which",
+                side_effect=lambda cmd: (None if cmd == "mke2fs" else "/usr/bin/x"),  # type: ignore[misc]
+            ))
+            rc, out = _cap(fc.status)
+        self.assertEqual(1, rc)
+        self.assertIn("mke2fs: NOT found", out)
+
+
+class TestFirecrackerStatusRuntime(unittest.TestCase):
+    """status() reports binary and KVM problems and returns non-zero."""
+
+    def _apply_pool_ok(self, stack: contextlib.ExitStack) -> None:
+        kernel_mock = MagicMock()
+        kernel_mock.is_file.return_value = True
+        dropbear_mock = MagicMock()
+        dropbear_mock.is_file.return_value = True
+        stack.enter_context(patch.object(fc.util, "kernel_path", return_value=kernel_mock))
+        stack.enter_context(patch.object(fc.util, "dropbear_path", return_value=dropbear_mock))
+        stack.enter_context(patch.object(netpool, "missing_taps", return_value=[]))
+        stack.enter_context(patch.object(netpool, "pool_size", return_value=8))
+        stack.enter_context(patch.object(netpool, "overlapping_routes", return_value=[]))
+        stack.enter_context(patch.object(fc, "_report_persistence", lambda: None))
+
+    def test_status_fails_when_binary_missing(self):
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(
+                patch.object(fc, "_firecracker_binary_ok", return_value=False))
+            stack.enter_context(
+                patch.object(fc, "_kvm_accessible", return_value=True))
+            stack.enter_context(
+                patch.object(fc.shutil, "which", return_value=None))
+            self._apply_pool_ok(stack)
+            rc, out = _cap(fc.status)
+        self.assertEqual(1, rc)
+        self.assertIn("NOT found on PATH", out)
+
+    def test_status_fails_when_kvm_not_accessible(self):
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(
+                patch.object(fc, "_firecracker_binary_ok", return_value=True))
+            stack.enter_context(
+                patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
+            stack.enter_context(
+                patch.object(fc, "_kvm_accessible", return_value=False))
+            stack.enter_context(
+                patch.object(fc.os.path, "exists", return_value=True))
+            self._apply_pool_ok(stack)
+            rc, out = _cap(fc.status)
+        self.assertEqual(1, rc)
+        self.assertIn("not accessible", out)
+
+    def test_status_ok_when_binary_and_kvm_ready(self):
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(
+                patch.object(fc, "_firecracker_binary_ok", return_value=True))
+            stack.enter_context(
+                patch.object(fc.shutil, "which", return_value="/usr/bin/firecracker"))
+            stack.enter_context(
+                patch.object(fc, "_kvm_accessible", return_value=True))
+            stack.enter_context(
+                patch.object(netpool, "nft_table_present", return_value=True))
+            self._apply_pool_ok(stack)
+            rc, out = _cap(fc.status)
+        self.assertEqual(0, rc)
+        self.assertIn("firecracker binary: ok", out)
+        self.assertIn("KVM:", out)
+
+
+class TestBackendStatusQuiet(unittest.TestCase):
+    """status(quiet=True) routes the underlying status() call through
+    redirect_stderr so diagnostic output is suppressed. Verify the return
+    code is propagated and that calling with quiet=False leaves the non-quiet
+    path active (covered by the other TestDockerSetupStatus tests)."""
+
+    def test_docker_quiet_true_propagates_return_code(self):
+        from bot_bottle.backend.docker.backend import DockerBottleBackend
+        with patch.object(dk, "status", return_value=0):
+            self.assertEqual(0, DockerBottleBackend.status(quiet=True))
+
+    def test_docker_quiet_true_suppresses_stderr(self):
+        import io as _io
+        from bot_bottle.backend.docker.backend import DockerBottleBackend
+
+        def _loud_status():
+            import sys
+            print("should be suppressed", file=sys.stderr)
+            return 0
+
+        with patch.object(dk, "status", side_effect=_loud_status):
+            buf = _io.StringIO()
+            with contextlib.redirect_stderr(buf):
+                DockerBottleBackend.status(quiet=True)
+        self.assertEqual("", buf.getvalue())
+
+    def test_firecracker_quiet_true_propagates_return_code(self):
+        from bot_bottle.backend.firecracker.backend import FirecrackerBottleBackend
+        with patch.object(fc, "status", return_value=1):
+            self.assertEqual(1, FirecrackerBottleBackend.status(quiet=True))
+
+    def test_macos_quiet_true_propagates_return_code(self):
+        from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
+        with patch.object(mc, "status", return_value=0):
+            self.assertEqual(0, MacosContainerBottleBackend.status(quiet=True))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_backend_skip_guards.py

@@ -0,0 +1,78 @@
+"""Tests for the backend-aware skip guards in ``tests/_backend.py``.
+
+The guards delegate their readiness check to
+``bot_bottle.backend.is_backend_ready`` (the probe behind ``./cli.py backend
+status``); here that probe is mocked so the unit job asserts the
+selection/skip logic without either backend present on the runner.
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import patch
+
+from tests._backend import (
+    selected_backend,
+    skip_unless_backend,
+    skip_unless_selected_backend_available,
+)
+
+
+def _skipped(decorated: type) -> bool:
+    return getattr(decorated, "__unittest_skip__", False)
+
+
+def _new_case() -> type:
+    return type("Case", (unittest.TestCase,), {})
+
+
+class TestSelectedBackend(unittest.TestCase):
+    def test_defaults_to_docker_when_unset(self):
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual("docker", selected_backend())
+
+    def test_reads_env(self):
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True):
+            self.assertEqual("firecracker", selected_backend())
+
+
+class TestSkipUnlessBackend(unittest.TestCase):
+    def test_skips_when_other_backend_selected(self):
+        # A different backend is selected — no host probe needed, skip.
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
+                patch("tests._backend.is_backend_ready") as has:
+            decorated = skip_unless_backend("docker")(_new_case())
+        self.assertTrue(_skipped(decorated))
+        has.assert_not_called()
+
+    def test_runs_when_selected_and_available(self):
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
+                patch("tests._backend.is_backend_ready", return_value=True):
+            decorated = skip_unless_backend("docker")(_new_case())
+        self.assertFalse(_skipped(decorated))
+
+    def test_skips_when_selected_but_unavailable(self):
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "docker"}, clear=True), \
+                patch("tests._backend.is_backend_ready", return_value=False):
+            decorated = skip_unless_backend("docker")(_new_case())
+        self.assertTrue(_skipped(decorated))
+
+
+class TestSkipUnlessSelectedBackendAvailable(unittest.TestCase):
+    def test_runs_when_selected_backend_available(self):
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
+                patch("tests._backend.is_backend_ready", return_value=True) as has:
+            decorated = skip_unless_selected_backend_available()(_new_case())
+        self.assertFalse(_skipped(decorated))
+        has.assert_called_once_with("firecracker", quiet=False)
+
+    def test_skips_when_selected_backend_unavailable(self):
+        with patch.dict(os.environ, {"BOT_BOTTLE_BACKEND": "firecracker"}, clear=True), \
+                patch("tests._backend.is_backend_ready", return_value=False):
+            decorated = skip_unless_selected_backend_available()(_new_case())
+        self.assertTrue(_skipped(decorated))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_docker_test_helpers.py

@@ -1,35 +0,0 @@
-"""Tests for integration-test backend selection helpers."""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import patch
-
-from tests._docker import skip_unless_docker_or_firecracker
-
-
-class TestSkipUnlessDockerOrFirecracker(unittest.TestCase):
-    def test_firecracker_runs_when_docker_tests_are_disabled(self):
-        with patch.dict(
-            os.environ,
-            {"BOT_BOTTLE_BACKEND": "firecracker", "SKIP_DOCKER_TESTS": "1"},
-            clear=True,
-        ):
-            decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
-
-        self.assertFalse(getattr(decorated, "__unittest_skip__", False))
-
-    def test_non_firecracker_still_skips_when_docker_tests_are_disabled(self):
-        with patch.dict(
-            os.environ,
-            {"BOT_BOTTLE_BACKEND": "docker", "SKIP_DOCKER_TESTS": "1"},
-            clear=True,
-        ):
-            decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
-
-        self.assertTrue(getattr(decorated, "__unittest_skip__", False))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_firecracker_backend.py

@@ -132,30 +132,56 @@ class TestFirecrackerStatus(unittest.TestCase):
             rc = fc_setup.status()
         return rc, buf.getvalue()
 
+    def _stub_artifacts(self, fc_setup: object) -> tuple[MagicMock, MagicMock]:
+        k: MagicMock = MagicMock()
+        k.is_file.return_value = True
+        d: MagicMock = MagicMock()
+        d.is_file.return_value = True
+        return k, d
+
     def test_ready_when_taps_present_even_if_nft_unverifiable(self):
         from bot_bottle.backend.firecracker import setup as fc_setup
+
+        def _which(cmd: str) -> str | None:
+            return None if cmd == "nft" else f"/usr/bin/{cmd}"
+
+        k, d = self._stub_artifacts(fc_setup)
         with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
              patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
-             patch.object(fc_setup.shutil, "which", return_value=None):
+             patch.object(fc_setup.shutil, "which", side_effect=_which), \
+             patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
+             patch.object(fc_setup, "_kvm_accessible", return_value=True), \
+             patch.object(fc_setup.util, "kernel_path", return_value=k), \
+             patch.object(fc_setup.util, "dropbear_path", return_value=d):
             rc, out = self._run()
         self.assertEqual(0, rc)
         self.assertIn("unverified", out)
 
     def test_not_ready_when_taps_missing(self):
         from bot_bottle.backend.firecracker import setup as fc_setup
+        k, d = self._stub_artifacts(fc_setup)
         with patch.object(fc_setup.netpool, "missing_taps", return_value=["bbfc0"]), \
              patch.object(fc_setup.netpool, "overlapping_routes", return_value=[]), \
-             patch.object(fc_setup.shutil, "which", return_value=None):
+             patch.object(fc_setup.shutil, "which", return_value=None), \
+             patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
+             patch.object(fc_setup, "_kvm_accessible", return_value=True), \
+             patch.object(fc_setup.util, "kernel_path", return_value=k), \
+             patch.object(fc_setup.util, "dropbear_path", return_value=d):
             rc, _ = self._run()
         self.assertEqual(1, rc)
 
     def test_not_ready_on_range_overlap(self):
         from bot_bottle.backend.firecracker import netpool
         from bot_bottle.backend.firecracker import setup as fc_setup
+        k, d = self._stub_artifacts(fc_setup)
         conflict = netpool.RouteConflict(dst="10.243.0.0/24", dev="eth0")
         with patch.object(fc_setup.netpool, "missing_taps", return_value=[]), \
              patch.object(fc_setup.netpool, "overlapping_routes", return_value=[conflict]), \
-             patch.object(fc_setup.shutil, "which", return_value=None):
+             patch.object(fc_setup.shutil, "which", return_value=None), \
+             patch.object(fc_setup, "_firecracker_binary_ok", return_value=True), \
+             patch.object(fc_setup, "_kvm_accessible", return_value=True), \
+             patch.object(fc_setup.util, "kernel_path", return_value=k), \
+             patch.object(fc_setup.util, "dropbear_path", return_value=d):
             rc, out = self._run()
         self.assertEqual(1, rc)
         self.assertIn("CLASHES", out)