diff --git a/tests/unit/test_firecracker_cleanup.py b/tests/unit/test_firecracker_cleanup.py new file mode 100644 index 0000000..d5e6018 --- /dev/null +++ b/tests/unit/test_firecracker_cleanup.py @@ -0,0 +1,158 @@ +"""Unit: Firecracker cleanup (orphan enumeration + removal) and the +thin backend classmethod delegates. + +Cleanup shells out to pgrep/docker and touches the run dir; mock those so +the orphan-discovery + removal logic is covered without a live host. The +delegate tests just assert each backend's setup/status/teardown/cleanup +classmethods forward to their module. +""" + +from __future__ import annotations + +import subprocess +import unittest +from unittest.mock import patch + +from bot_bottle.backend.firecracker import cleanup as fc_cleanup +from bot_bottle.backend.firecracker.bottle_cleanup_plan import ( + FirecrackerBottleCleanupPlan, +) + + +def _proc(stdout: str = "", returncode: int = 0) -> "subprocess.CompletedProcess[str]": + return subprocess.CompletedProcess([], returncode, stdout=stdout, stderr="") + + +class TestOrphanEnumeration(unittest.TestCase): + def test_orphan_vm_pids_filters_by_run_dir(self): + run_root = str(fc_cleanup._run_root()) + out = ( + f"111 firecracker --config-file {run_root}/dev-a/config.json\n" + "222 firecracker --config-file /somewhere/else/config.json\n" + "notanint firecracker --config-file " + run_root + "/x\n" + ) + with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)): + self.assertEqual([111], fc_cleanup._orphan_vm_pids()) + + def test_orphan_vm_pids_empty_when_pgrep_fails(self): + with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)): + self.assertEqual([], fc_cleanup._orphan_vm_pids()) + + def test_sidecar_containers_sorted(self): + with patch.object(fc_cleanup.subprocess, "run", + return_value=_proc("bot-bottle-sidecars-b\nbot-bottle-sidecars-a\n")): + self.assertEqual( + ["bot-bottle-sidecars-a", "bot-bottle-sidecars-b"], + fc_cleanup._sidecar_containers(), + ) + + def test_sidecar_containers_empty_on_failure(self): + with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)): + self.assertEqual([], fc_cleanup._sidecar_containers()) + + def test_run_dirs_empty_when_absent(self): + with patch.object(fc_cleanup.util, "cache_dir") as cache: + cache.return_value.__truediv__.return_value.is_dir.return_value = False + self.assertEqual([], fc_cleanup._run_dirs()) + + def test_prepare_cleanup_assembles_plan(self): + with patch.object(fc_cleanup, "_orphan_vm_pids", return_value=[7]), \ + patch.object(fc_cleanup, "_sidecar_containers", return_value=["c1"]), \ + patch.object(fc_cleanup, "_run_dirs", return_value=["/run/x"]): + plan = fc_cleanup.prepare_cleanup() + self.assertEqual((7,), plan.vm_pids) + self.assertEqual(("c1",), plan.containers) + self.assertEqual(("/run/x",), plan.run_dirs) + + +class TestCleanupRemoval(unittest.TestCase): + def test_cleanup_kills_removes_and_rmtrees(self): + plan = FirecrackerBottleCleanupPlan( + vm_pids=(101,), containers=("bot-bottle-sidecars-x",), + run_dirs=("/run/dev-x",), + ) + with patch.object(fc_cleanup.os, "kill") as kill, \ + patch.object(fc_cleanup.subprocess, "run") as run, \ + patch.object(fc_cleanup.shutil, "rmtree") as rmtree, \ + patch.object(fc_cleanup, "info"): + fc_cleanup.cleanup(plan) + kill.assert_called_once() + run.assert_called_once() + self.assertIn("bot-bottle-sidecars-x", run.call_args.args[0]) + rmtree.assert_called_once_with("/run/dev-x", ignore_errors=True) + + def test_cleanup_tolerates_dead_pid(self): + plan = FirecrackerBottleCleanupPlan(vm_pids=(999,)) + with patch.object(fc_cleanup.os, "kill", side_effect=ProcessLookupError), \ + patch.object(fc_cleanup, "info"): + fc_cleanup.cleanup(plan) # must not raise + + +class TestCleanupPlan(unittest.TestCase): + def test_empty(self): + self.assertTrue(FirecrackerBottleCleanupPlan().empty) + self.assertFalse(FirecrackerBottleCleanupPlan(vm_pids=(1,)).empty) + + def test_print_empty(self): + with patch("bot_bottle.backend.firecracker.bottle_cleanup_plan.info") as info: + FirecrackerBottleCleanupPlan().print() + info.assert_called_once() + self.assertIn("nothing to remove", info.call_args.args[0]) + + def test_print_lists_resources(self): + plan = FirecrackerBottleCleanupPlan( + vm_pids=(5,), containers=("c",), run_dirs=("/r",), + ) + with patch("bot_bottle.backend.firecracker.bottle_cleanup_plan.info") as info: + plan.print() + joined = " ".join(c.args[0] for c in info.call_args_list) + self.assertIn("pid 5", joined) + self.assertIn("container: c", joined) + self.assertIn("run dir: /r", joined) + + +class TestBackendDelegates(unittest.TestCase): + """Each backend's setup/status/teardown/cleanup classmethods forward + to their module — assert the forwarding (and return-code passthrough).""" + + def test_firecracker_delegates(self): + from bot_bottle.backend.firecracker import FirecrackerBottleBackend + with patch("bot_bottle.backend.firecracker.setup.setup", return_value=1) as s, \ + patch("bot_bottle.backend.firecracker.setup.status", return_value=2), \ + patch("bot_bottle.backend.firecracker.setup.teardown", return_value=3): + self.assertEqual(1, FirecrackerBottleBackend.setup()) + self.assertEqual(2, FirecrackerBottleBackend.status()) + self.assertEqual(3, FirecrackerBottleBackend.teardown()) + s.assert_called_once_with() + with patch("bot_bottle.backend.firecracker.cleanup.prepare_cleanup") as pc, \ + patch("bot_bottle.backend.firecracker.cleanup.cleanup") as cl, \ + patch("bot_bottle.backend.firecracker.enumerate.enumerate_active", + return_value=[]): + b = FirecrackerBottleBackend() + b.prepare_cleanup() + b.cleanup(FirecrackerBottleCleanupPlan()) + self.assertEqual([], b.enumerate_active()) + pc.assert_called_once_with() + cl.assert_called_once() + + def test_docker_delegates(self): + from bot_bottle.backend.docker import DockerBottleBackend + with patch("bot_bottle.backend.docker.setup.setup", return_value=4), \ + patch("bot_bottle.backend.docker.setup.status", return_value=5), \ + patch("bot_bottle.backend.docker.setup.teardown", return_value=6): + self.assertEqual(4, DockerBottleBackend.setup()) + self.assertEqual(5, DockerBottleBackend.status()) + self.assertEqual(6, DockerBottleBackend.teardown()) + + def test_macos_delegates(self): + from bot_bottle.backend.macos_container import MacosContainerBottleBackend + with patch("bot_bottle.backend.macos_container.setup.setup", return_value=7), \ + patch("bot_bottle.backend.macos_container.setup.status", return_value=8), \ + patch("bot_bottle.backend.macos_container.setup.teardown", return_value=9): + self.assertEqual(7, MacosContainerBottleBackend.setup()) + self.assertEqual(8, MacosContainerBottleBackend.status()) + self.assertEqual(9, MacosContainerBottleBackend.teardown()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_firecracker_helpers.py b/tests/unit/test_firecracker_helpers.py new file mode 100644 index 0000000..a0c7ada --- /dev/null +++ b/tests/unit/test_firecracker_helpers.py @@ -0,0 +1,117 @@ +"""Unit: small Firecracker helpers — guest-IP parsing, the unprivileged +nft/TAP probes, console-tail, and the require_firecracker preflight +branches. Mock subprocess/os so nothing needs KVM or a live VM. +""" + +from __future__ import annotations + +import json +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from bot_bottle.backend.firecracker import firecracker_vm, freezer, netpool, util + + +class TestGuestIpFromConfig(unittest.TestCase): + def _write(self, obj: object) -> Path: + d = tempfile.mkdtemp(prefix="fc-cfg.") + p = Path(d) / "config.json" + p.write_text(json.dumps(obj)) + self.addCleanup(lambda: p.unlink(missing_ok=True)) + return p + + def test_extracts_guest_ip(self): + cfg = {"boot-source": {"boot_args": + "console=ttyS0 ip=10.243.0.1::10.243.0.0:255.255.255.254::eth0:off"}} + self.assertEqual("10.243.0.1", freezer._guest_ip_from_config(self._write(cfg))) + + def test_no_ip_token(self): + self.assertEqual("", freezer._guest_ip_from_config( + self._write({"boot-source": {"boot_args": "console=ttyS0"}}))) + + def test_missing_file(self): + self.assertEqual("", freezer._guest_ip_from_config(Path("/nope/config.json"))) + + def test_malformed_json(self): + d = tempfile.mkdtemp(prefix="fc-cfg.") + p = Path(d) / "config.json" + p.write_text("{not json") + self.addCleanup(lambda: p.unlink(missing_ok=True)) + self.assertEqual("", freezer._guest_ip_from_config(p)) + + +class TestNetpoolProbes(unittest.TestCase): + def test_run_ok_true_on_zero(self): + with patch.object(netpool.subprocess, "run", + return_value=subprocess.CompletedProcess([], 0)): + self.assertTrue(netpool._run_ok(["true"])) + + def test_run_ok_false_on_missing_binary(self): + with patch.object(netpool.subprocess, "run", side_effect=FileNotFoundError): + self.assertFalse(netpool._run_ok(["nft"])) + + def test_tap_and_nft_probes(self): + with patch.object(netpool, "_run_ok", return_value=True) as ok: + self.assertTrue(netpool.tap_present("bbfc0")) + self.assertTrue(netpool.nft_table_present()) + self.assertEqual(2, ok.call_count) + + def test_missing_taps(self): + with patch.dict("os.environ", {"BOT_BOTTLE_FC_POOL_SIZE": "2"}), \ + patch.object(netpool, "tap_present", side_effect=[True, False]): + self.assertEqual(["bbfc1"], netpool.missing_taps()) + + +class TestConsoleTail(unittest.TestCase): + def test_reads_tail(self): + d = tempfile.mkdtemp(prefix="fc-con.") + p = Path(d) / "console.log" + p.write_text("l1\nl2\nl3\n") + self.addCleanup(lambda: p.unlink(missing_ok=True)) + out = firecracker_vm._console_tail(p, lines=2) + self.assertIn("l2", out) + self.assertIn("l3", out) + self.assertNotIn("l1", out) + + def test_missing_file(self): + self.assertIn("no console log", + firecracker_vm._console_tail(Path("/nope/console.log"))) + + +class TestRequireFirecracker(unittest.TestCase): + def _die(self): + return patch.object(util, "die", side_effect=SystemExit("die")) + + def test_dies_when_not_linux(self): + with patch.object(util, "is_linux", return_value=False), self._die(): + with self.assertRaises(SystemExit): + util.require_firecracker() + + def test_dies_when_binary_missing(self): + with patch.object(util, "is_linux", return_value=True), \ + patch.object(util.shutil, "which", return_value=None), \ + patch.object(util, "info"), self._die(): + with self.assertRaises(SystemExit): + util.require_firecracker() + + def test_dies_when_kvm_missing(self): + with patch.object(util, "is_linux", return_value=True), \ + patch.object(util.shutil, "which", return_value="/usr/bin/firecracker"), \ + patch.object(util.os.path, "exists", return_value=False), self._die(): + with self.assertRaises(SystemExit): + util.require_firecracker() + + def test_dies_when_kernel_missing(self): + with patch.object(util, "is_linux", return_value=True), \ + patch.object(util.shutil, "which", return_value="/usr/bin/firecracker"), \ + patch.object(util, "_require_kvm", lambda: None), \ + patch.object(util.Path, "is_file", return_value=False), self._die(): + with self.assertRaises(SystemExit): + util.require_firecracker() + + +if __name__ == "__main__": + unittest.main()