79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from uvm.config import Settings
|
|
from uvm.domain import VmRecord
|
|
from uvm.errors import FirecrackerError
|
|
from uvm.firecracker.process import FirecrackerProcessManager
|
|
|
|
|
|
def make_vm(pid: int | None, start_time: str | None) -> VmRecord:
|
|
return VmRecord(
|
|
id="vm-test",
|
|
cpu=1,
|
|
ram_mib=512,
|
|
guest_ip="10.42.0.2",
|
|
gateway="10.42.0.1",
|
|
tap="uvm-test",
|
|
mac="02:fc:00:00:00:01",
|
|
socket="/tmp/uvm-missing.sock",
|
|
config="/tmp/config.json",
|
|
log="/tmp/firecracker.log",
|
|
pid=pid,
|
|
process_start_time=start_time,
|
|
)
|
|
|
|
|
|
class FirecrackerProcessSafetyTests(unittest.TestCase):
|
|
def test_refuses_to_signal_an_unidentified_legacy_pid(self) -> None:
|
|
manager = FirecrackerProcessManager(Settings())
|
|
vm = make_vm(pid=12345, start_time=None)
|
|
|
|
with patch("uvm.firecracker.process.os.kill") as kill:
|
|
with self.assertRaises(FirecrackerError):
|
|
manager.terminate(vm)
|
|
|
|
kill.assert_called_once_with(12345, 0)
|
|
|
|
def test_allows_cleanup_when_an_unidentified_legacy_pid_is_already_gone(self) -> None:
|
|
manager = FirecrackerProcessManager(Settings())
|
|
vm = make_vm(pid=12345, start_time=None)
|
|
|
|
with patch("uvm.firecracker.process.os.kill", side_effect=ProcessLookupError):
|
|
manager.terminate(vm)
|
|
|
|
def test_rejects_non_positive_pids_without_signaling(self) -> None:
|
|
manager = FirecrackerProcessManager(Settings())
|
|
vm = make_vm(pid=0, start_time="42")
|
|
|
|
with patch("uvm.firecracker.process.os.kill") as kill:
|
|
with self.assertRaises(FirecrackerError):
|
|
manager.terminate(vm)
|
|
|
|
kill.assert_not_called()
|
|
|
|
def test_interrupt_during_startup_terminates_the_child_process(self) -> None:
|
|
class FakePopen:
|
|
pid = 12345
|
|
|
|
def poll(self) -> int | None:
|
|
return None
|
|
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
|
settings = Settings(base=Path(temporary_directory), terminate_timeout_s=0)
|
|
manager = FirecrackerProcessManager(settings, popen=lambda *_args, **_kwargs: FakePopen())
|
|
manager._wait_for_socket = lambda *_args: (_ for _ in ()).throw(KeyboardInterrupt)
|
|
vm = make_vm(pid=None, start_time=None)
|
|
vm.log = str(Path(temporary_directory) / "firecracker.log")
|
|
vm.socket = str(Path(temporary_directory) / "firecracker.sock")
|
|
|
|
with patch("uvm.firecracker.process.os.kill") as kill:
|
|
with self.assertRaises(KeyboardInterrupt):
|
|
manager.start(vm)
|
|
|
|
self.assertEqual(kill.call_count, 2)
|