kstyagi/nest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update terminal support messaging and enhance guest agent provisioning logic

Browse Source
This commit is contained in:
kstyagi
2026-09-25 21:12:21 +05:30
parent d870c98bfa
commit 6573f483e6
10 changed files with 163 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_firecracker_config.py
+1
View File
@@ -46,6 +46,7 @@ class FirecrackerConfigTests(unittest.TestCase):
arguments = config["boot-source"]["boot_args"].split()
self.assertEqual("keep_bootcon" in arguments, architecture != "x86_64")
self.assertIn("console=ttyS0", arguments)
self.assertEqual(config["vsock"], {"guest_cid": 3, "uds_path": "/var/lib/uvm/vms/vm-test/vsock.sock"})
self.assertIn("ip=10.42.0.2::10.42.0.1:255.255.255.0::eth0:off", arguments)
client = RecordingFirecrackerClient()
client.configure_and_start(config)
tests/test_guest_agent_images.py
+47 -8
View File
@@ -74,10 +74,14 @@ class GuestAgentImageTests(unittest.TestCase):
def make_disk(self, marker="uvm-ubuntu-jammy-v3\n") -> None:
root = self.base / "tree"
if root.exists():
shutil.rmtree(root)
payloads = {
**self.retained, "/etc/uvm-image": marker,
**self.retained,
"/usr/lib/systemd/systemd": "systemd fixture", "/usr/bin/apt-get": "apt fixture",
}
if marker is not None:
payloads["/etc/uvm-image"] = marker
for path, content in payloads.items():
target = root / path.lstrip("/")
target.parent.mkdir(parents=True, exist_ok=True)
@@ -134,14 +138,49 @@ class GuestAgentImageTests(unittest.TestCase):
self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True)
self.assertEqual(self.guest("cat /home/worker/data"), self.retained["/home/worker/data"])
def test_current_templates_are_configured_and_older_templates_require_explicit_install(self) -> None:
def test_all_compatible_images_get_vsock_agent_setup_without_opt_in(self) -> None:
for marker in ("uvm-ubuntu-jammy-v1\n", "uvm-ubuntu-jammy-v3\n", "uvm-ubuntu-jammy-v4\n",
"uvm-ubuntu-jammy-v5\n", "custom-image\n", None):
with self.subTest(marker=marker):
self.make_disk(marker)
template = hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest()
self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-test", "worker"))
self.assertEqual(json.loads(self.guest("cat /etc/uvm-agent.json")), {"vm_id": "vm-test", "username": "worker"})
self.assertIn('Fast link dest: "/etc/systemd/system/uvm-agent.service"',
self.guest("stat /etc/systemd/system/multi-user.target.wants/uvm-agent.service"))
self.assertEqual(self.guest("cat /usr/lib/uvm/agent.py"), guest_agent_files()["/usr/lib/uvm/agent.py"].decode())
for path, content in self.retained.items():
self.assertEqual(self.guest(f"cat {path}"), content)
self.assertEqual(hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest(), template)
def test_current_agent_setup_is_read_only_on_later_boots_and_tracks_vm_identity(self) -> None:
self.make_disk()
self.images.provision_guest_agent(self.disk, "vm-test", "worker")
before = self.snapshot()
self.assertFalse(self.images.provision_guest_agent(self.disk, "vm-test", "worker"))
with patch.object(self.images, "_copy_sparse") as copy:
self.assertFalse(self.images.provision_guest_agent(self.disk, "vm-test", "worker"))
copy.assert_not_called()
self.assertEqual(self.snapshot(), before)
for version in (4, 5):
self.replace_guest_file("/etc/uvm-image", f"uvm-ubuntu-jammy-v{version}\n")
self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-test", "worker"))
self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-recreated", "worker"))
self.assertEqual(json.loads(self.guest("cat /etc/uvm-agent.json")), {"vm_id": "vm-recreated", "username": "worker"})
def test_default_setup_reenables_an_existing_agent_service(self) -> None:
self.make_disk()
self.images.provision_guest_agent(self.disk, "vm-test", "worker")
self.runner.run(("debugfs", "-w", "-R", "rm /etc/systemd/system/multi-user.target.wants/uvm-agent.service", self.disk), capture=True)
self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-test", "worker"))
self.assertIn('Fast link dest: "/etc/systemd/system/uvm-agent.service"',
self.guest("stat /etc/systemd/system/multi-user.target.wants/uvm-agent.service"))
def test_incompatible_images_fail_default_setup_without_changing_the_disk(self) -> None:
for binary in ("/usr/lib/systemd/systemd", "/usr/bin/apt-get"):
with self.subTest(binary=binary):
self.make_disk(marker=None)
self.runner.run(("debugfs", "-w", "-R", f"rm {binary}", self.disk), capture=True)
before = self.snapshot()
with self.assertRaisesRegex(UvmError, "automatic vsock guest-agent setup requires a systemd/apt image"):
self.images.provision_guest_agent(self.disk, "vm-test", "worker")
self.assertEqual(self.snapshot(), before)
def test_cached_template_without_a_manifest_only_adopts_the_known_v4_payload(self) -> None:
self.make_disk()
@@ -163,12 +202,12 @@ class GuestAgentImageTests(unittest.TestCase):
updated = guest_agent_files()
updated["/usr/lib/uvm/agent.py"] += b"\n# next packaged revision\n"
with patch("uvm.images.guest_agent_files", return_value=updated):
self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True)
self.images.provision_guest_agent(self.disk, "vm-test", "worker")
self.assertTrue(self.guest("cat /usr/lib/uvm/agent.py").endswith("# next packaged revision\n"))
self.replace_guest_file("/usr/lib/uvm/agent.py", "guest customization\n")
before = self.snapshot()
with self.assertRaisesRegex(UvmError, "customized guest agent file"):
self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True)
self.images.provision_guest_agent(self.disk, "vm-test", "worker")
self.assertEqual(self.snapshot(), before)
self.assertEqual(list(self.disk.parent.iterdir()), [self.disk])
tests/test_guest_exec_lifecycle.py
+60 -1
View File
@@ -4,7 +4,7 @@ from pathlib import Path
from unittest.mock import Mock
from tests.test_lifecycle import LifecycleTestCase
from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError
from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError, UvmError
from uvm.guest_exec import ExecResult
@@ -22,6 +22,65 @@ class GuestExecLifecycleTests(LifecycleTestCase):
vm = self.create_vm(username="worker")
self.assert_call_order(vm, "images.provision_agent", "process.start")
def test_stopped_boots_automatically_provision_agent_before_starting_the_vmm(self) -> None:
vm = self.create_vm(username="worker", password="retained-password")
identity = (vm.id, vm.guest_ip, vm.mac, vm.username, vm.password)
def provision(disk, vm_id, username):
self.host.call("images.provision_agent", vm_id)
record = self.store.get(vm_id)
self.assertTrue(self.store.operation_active)
self.assertEqual(self.process.observe(record), "gone")
self.assertEqual((record.id, record.guest_ip, record.mac, record.username, record.password), identity)
self.assertEqual((disk, username), (Path(vm.disk), "worker"))
self.provision_agent.side_effect = provision
for action in ("start", "restart", "resize"):
with self.subTest(action=action):
if action == "start":
self.service.stop(vm.id)
self.host.events.clear()
self.provision_agent.reset_mock()
self.provision.reset_mock()
if action == "resize":
started = self.service.resize(vm.id, cpu=2)
else:
started = getattr(self.service, action)(vm.id)
self.assertEqual(started.status, "running")
self.provision_agent.assert_called_once_with(Path(vm.disk), vm.id, "worker")
self.provision.assert_not_called()
self.assert_call_order(vm, "images.provision_agent", "process.start")
def test_live_or_ambiguous_vms_never_run_offline_agent_provisioning(self) -> None:
vm = self.create_vm()
self.provision_agent.reset_mock()
self.service.start(vm.id)
self.process.unknown.add(vm.id)
with self.assertRaises(ConflictError):
self.service.start(vm.id)
self.provision_agent.assert_not_called()
def test_default_agent_setup_failure_blocks_boot_and_keeps_the_retained_disk(self) -> None:
vm = self.create_vm()
self.service.stop(vm.id)
before = Path(vm.disk).read_bytes()
count = len(self.process.started)
self.provision_agent.side_effect = UvmError("agent setup failed")
with self.assertRaisesRegex(UvmError, "agent setup failed"):
self.service.start(vm.id)
self.assertEqual(len(self.process.started), count)
self.assertEqual(Path(vm.disk).read_bytes(), before)
self.assertEqual(self.store.get(vm.id).status, "failed")
self.provision_agent.side_effect = None
self.assertEqual(self.service.start(vm.id).status, "running")
def test_default_agent_setup_failure_rolls_back_a_new_vm_before_boot(self) -> None:
self.provision_agent.side_effect = UvmError("unsupported guest image")
with self.assertRaisesRegex(UvmError, "unsupported guest image"):
self.create_vm()
self.assertEqual(self.store.load().vms, {})
self.assertEqual(self.process.started, [])
def test_execution_pins_the_exact_vm_under_lock_and_waits_without_holding_it(self) -> None:
vm = self.create_vm(name="worker", ports=())
result = ExecResult(7, "output", "problem", False, False)
tests/test_guest_terminal.py
+1 -1
View File
@@ -128,7 +128,7 @@ class TerminalCompatibilityTests(unittest.TestCase):
thread = threading.Thread(target=old_agent)
thread.start()
try:
with self.assertRaisesRegex(GuestAgentUnavailableError, "Install / update terminal support"):
with self.assertRaisesRegex(GuestAgentUnavailableError, "Restart this VM to apply automatic guest-agent setup"):
terminal.start(24, 80)
finally:
thread.join(3)