from __future__ import annotations import argparse import importlib.util import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from uvm.cli import build_parser, main from uvm.config import Settings from uvm.domain import PortRule, VmRecord, VmSpec from uvm.errors import ConfigurationError, ConflictError, NotFoundError, UvmError, ValidationError from uvm.lifecycle import ListedVm, OperationResult from uvm.server import _http_status_for, create_api, run_server def fake_application(*, api_token: str | None = None): settings = SimpleNamespace( app_name="uvm", default_vcpu=1, default_ram_mib=512, default_ssh_user="root", api_token=api_token, api_tls_cert=None, api_tls_key=None, ) return SimpleNamespace(settings=settings) class ServerCliTests(unittest.TestCase): def test_parser_accepts_the_requested_serve_invocation(self) -> None: parser = build_parser(fake_application()) args = parser.parse_args(["--serve", "--port", "8123", "--host", "127.0.0.1"]) self.assertTrue(args.serve) self.assertEqual(args.port, 8123) self.assertEqual(args.host, "127.0.0.1") self.assertIsNone(args.command) def test_parser_rejects_an_invalid_server_port(self) -> None: parser = build_parser(fake_application()) with self.assertRaises(SystemExit): parser.parse_args(["--serve", "--port", "70000"]) def test_create_parser_defaults_guest_credentials_to_root(self) -> None: args = build_parser(fake_application()).parse_args(["create"]) self.assertEqual(args.username, "root") self.assertEqual(args.password, "root") def test_main_delegates_serve_mode_to_the_server_launcher(self) -> None: application = fake_application() with ( patch("uvm.cli.build_application", return_value=application), patch("uvm.server.run_server") as run_server_mock, ): result = main(["--serve", "--host", "127.0.0.1", "--port", "8123"]) self.assertEqual(result, 0) run_server_mock.assert_called_once_with(application, host="127.0.0.1", port=8123) def test_server_requires_an_api_token_even_on_loopback(self) -> None: with self.assertRaises(UvmError): run_server(fake_application(api_token=None), host="127.0.0.1", port=8000) def test_non_loopback_server_requires_tls(self) -> None: with self.assertRaises(UvmError): run_server(fake_application(api_token="test-token"), host="0.0.0.0", port=8000) def test_factory_requires_an_explicit_host(self) -> None: with self.assertRaises(UvmError): create_api(fake_application(api_token="test-token")) def test_invalid_tls_material_is_rejected_before_uvicorn_starts(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: certificate = Path(temporary_directory) / "cert.pem" key = Path(temporary_directory) / "key.pem" certificate.touch() key.touch() application = fake_application(api_token="test-token") application.settings.api_tls_cert = certificate application.settings.api_tls_key = key with self.assertRaises(UvmError): run_server(application, host="127.0.0.1", port=8000) def test_non_loopback_server_uses_configured_tls(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: certificate = Path(temporary_directory) / "cert.pem" key = Path(temporary_directory) / "key.pem" certificate.touch() key.touch() application = fake_application(api_token="test-token") application.settings.api_tls_cert = certificate application.settings.api_tls_key = key fake_uvicorn = SimpleNamespace(run=lambda *_args, **_kwargs: None) with ( patch("uvm.server._validate_server_settings"), patch("uvm.server.create_api", return_value=object()), patch.dict(sys.modules, {"uvicorn": fake_uvicorn}), patch.object(fake_uvicorn, "run") as run_mock, ): run_server(application, host="0.0.0.0", port=8443) run_mock.assert_called_once() self.assertEqual(run_mock.call_args.kwargs["ssl_certfile"], str(certificate)) self.assertEqual(run_mock.call_args.kwargs["ssl_keyfile"], str(key)) def test_api_token_rejects_non_ascii_or_whitespace(self) -> None: for token in ("s\u00e9cret", "contains space", ""): with self.subTest(token=token): with self.assertRaises(ConfigurationError): Settings(api_token=token) class ServerErrorTests(unittest.TestCase): def test_domain_errors_map_to_http_status_codes(self) -> None: for error, expected in ( (NotFoundError("unknown guest"), 404), (ConflictError("ambiguous name"), 409), (ValidationError("invalid name"), 422), (ConfigurationError("invalid settings"), 422), (UvmError("this operation needs root"), 403), (UvmError("host operation failed"), 500), ): with self.subTest(error=error): self.assertEqual(_http_status_for(error), expected) FASTAPI_AVAILABLE = ( importlib.util.find_spec("fastapi") is not None and importlib.util.find_spec("httpx") is not None ) if FASTAPI_AVAILABLE: from fastapi.testclient import TestClient class FakeLifecycle: def __init__(self) -> None: self.last_spec: VmSpec | None = None self.observed_status = "running" self.policy_status = "enforced" self.vm = VmRecord( id="vm-test", name="test-vm", 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/firecracker.sock", config="/tmp/config.json", log="/tmp/firecracker.log", password="guest-secret", status="running", ports=(PortRule(22),), operation=None, disk_ready=True, boot_id="test-boot", ) def list_vms(self) -> list[ListedVm]: return [ListedVm( vm=self.vm, observed_status=self.observed_status, policy_status=self.policy_status, )] def inspect(self, identifier: str) -> ListedVm: if identifier not in {self.vm.id, self.vm.guest_ip, self.vm.name}: raise NotFoundError(f"VM not found: {identifier}") return ListedVm( vm=self.vm, observed_status=self.observed_status, policy_status=self.policy_status, ) def create(self, spec: VmSpec) -> VmRecord: self.last_spec = spec self.vm.cpu = spec.cpu self.vm.ram_mib = spec.ram_mib if spec.guest_ip is not None: self.vm.guest_ip = str(spec.guest_ip) self.vm.username = spec.username self.vm.password = spec.password self.vm.name = spec.name self.vm.ports = spec.ports self.vm.status = "running" self.vm.operation = None self.observed_status = "running" self.policy_status = "enforced" return self.vm def start(self, identifier: str) -> VmRecord: self.inspect(identifier) self.vm.status = "running" self.observed_status = "running" self.policy_status = "enforced" return self.vm def restart(self, identifier: str, *, timeout: float | None = None) -> VmRecord: self.inspect(identifier) self.vm.status = "running" self.observed_status = "running" self.policy_status = "enforced" return self.vm def stop(self, identifier: str, *, timeout: float | None = None) -> VmRecord: self.inspect(identifier) self.vm.status = "stopped" self.vm.pid = None self.vm.operation = None self.observed_status = "stopped" self.policy_status = "inactive" return self.vm def destroy(self, identifier: str, *, timeout: float | None = None) -> VmRecord: self.inspect(identifier) return self.vm def read_log_tail(self, identifier: str, *, tail: int) -> SimpleNamespace: self.inspect(identifier) return SimpleNamespace(content=f"last {tail} log lines", truncated=False) def diagnostics(self) -> list[dict]: return [{"name": "kvm", "ok": True, "message": "available"}] def resolve_targets(self, identifiers: list[str], *, all_vms: bool = False) -> list[VmRecord]: if all_vms and identifiers: raise ValidationError("--all cannot be combined with explicit VM targets") if not all_vms and not identifiers: raise ValidationError("specify a VM or --all") if all_vms: return [self.vm] return [self.inspect(identifier).vm for identifier in identifiers] def run_many( self, action: str, vm_ids: list[str], *, timeout: float | None = None, ) -> list[OperationResult]: return [OperationResult(vm_id, action, "succeeded") for vm_id in vm_ids] def set_ports(self, identifier: str, ports: tuple[PortRule, ...], *, mode: str = "set") -> VmRecord: self.inspect(identifier) if mode == "set": self.vm.ports = ports elif mode == "open": existing = {p.port: p for p in self.vm.ports or ()} for p in ports: existing.setdefault(p.port, p) self.vm.ports = tuple(sorted(existing.values(), key=lambda r: r.port)) elif mode == "close": drop = {p.port for p in ports} self.vm.ports = tuple(p for p in self.vm.ports or () if p.port not in drop) self.policy_status = "enforced" return self.vm def sync_ports(self, identifier: str) -> VmRecord: self.inspect(identifier) self.policy_status = "enforced" return self.vm @unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required for API tests") class ApiRouterTests(unittest.TestCase): def setUp(self) -> None: application = fake_application(api_token="test-token") self.lifecycle = FakeLifecycle() application.lifecycle = self.lifecycle application.installer = SimpleNamespace( install=lambda **_kwargs: ( Path("/tmp/firecracker"), SimpleNamespace(kernel=Path("/tmp/vmlinux"), rootfs=Path("/tmp/rootfs")), ) ) application.state_store = SimpleNamespace(initialize=lambda: None) self.headers = {"X-UVM-Token": "test-token"} self.client = TestClient(create_api(application, host="127.0.0.1")) self.addCleanup(self.client.close) def test_health_is_available_without_credentials(self) -> None: response = self.client.get("/health") self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {"status": "ok"}) def test_dashboard_and_assets_are_available_without_credentials(self) -> None: root = self.client.get("/", follow_redirects=False) dashboard = self.client.get("/dashboard") stylesheet = self.client.get("/dashboard/static/css/dashboard.css") script = self.client.get("/dashboard/static/js/dashboard.js") self.assertEqual(root.status_code, 307) self.assertEqual(root.headers["location"], "/dashboard") self.assertEqual(dashboard.status_code, 200) self.assertIn("UVM Control", dashboard.text) self.assertEqual(stylesheet.status_code, 200) self.assertIn("--acid", stylesheet.text) self.assertEqual(script.status_code, 200) self.assertIn("sessionStorage", script.text) def test_vm_routes_require_and_accept_the_api_token(self) -> None: for method, path, payload in ( ("GET", "/vms", {}), ("POST", "/vms", {}), ("GET", "/vms/test-vm", {}), ("GET", "/vms/test-vm/logs", {}), ("POST", "/vms/actions", {"action": "start", "identifiers": ["vm-test"]}), ("POST", "/vms/test-vm/start", {}), ("POST", "/vms/test-vm/restart", {}), ("POST", "/vms/test-vm/stop", {}), ("DELETE", "/vms/test-vm", {}), ("POST", "/vms/test-vm/ports", {}), ("POST", "/vms/test-vm/ports/sync", {}), ("GET", "/diagnostics", {}), ("POST", "/install", {}), ): for headers in ({}, {"X-UVM-Token": "wrong-token"}): with self.subTest(method=method, path=path, headers=headers): response = self.client.request(method, path, headers=headers, json=payload) self.assertEqual(response.status_code, 401) self.assertIsNone(self.lifecycle.last_spec) self.assertEqual(self.lifecycle.vm.status, "running") authorized = self.client.get("/vms", headers=self.headers) self.assertEqual(authorized.status_code, 200) self.assertEqual(authorized.json()[0]["id"], "vm-test") self.assertNotIn("password", authorized.json()[0]) def test_vm_lifecycle_routes_and_extra_field_validation(self) -> None: created = self.client.post( "/vms", headers=self.headers, json={ "cpu": 1, "ram": "512", "username": "root", "password": "api-secret", }, ) detail = self.client.get("/vms/vm-test", headers=self.headers) stopped = self.client.post("/vms/vm-test/stop", headers=self.headers) destroyed = self.client.delete("/vms/vm-test", headers=self.headers) invalid = self.client.post("/vms", headers=self.headers, json={"forse": True}) self.assertEqual(created.status_code, 201) self.assertEqual(created.json()["username"], "root") self.assertEqual(self.lifecycle.last_spec.password, "api-secret") for response in (created, detail, stopped, destroyed): self.assertNotIn("password", response.json()) self.assertNotIn("api-secret", response.text) self.assertEqual(detail.status_code, 200) self.assertEqual(stopped.json()["status"], "stopped") self.assertEqual(stopped.json()["policy_status"], "inactive") self.assertEqual(destroyed.json(), {"id": "vm-test", "status": "terminated"}) self.assertEqual(invalid.status_code, 422) def test_vm_create_defaults_guest_credentials_to_root(self) -> None: response = self.client.post( "/vms", headers=self.headers, json={"cpu": 1, "ram": "512"}, ) self.assertEqual(response.status_code, 201) self.assertEqual(self.lifecycle.last_spec.username, "root") self.assertEqual(self.lifecycle.last_spec.password, "root") self.assertIsNone(self.lifecycle.last_spec.name) self.assertEqual(self.lifecycle.last_spec.ports, (PortRule(22),)) def test_create_accepts_valid_names(self) -> None: for name in (None, "a", "web-1", "a" * 63): with self.subTest(name=name): response = self.client.post("/vms", headers=self.headers, json={"name": name}) self.assertEqual(response.status_code, 201) self.assertEqual(self.lifecycle.last_spec.name, name) self.assertEqual(response.json()["name"], name) def test_create_rejects_invalid_names_before_lifecycle_calls(self) -> None: for name in ("", "-web", "9web", "Web", "web_name", "web.name", "two words", "a" * 64, 1, True): with self.subTest(name=name): response = self.client.post("/vms", headers=self.headers, json={"name": name}) self.assertEqual(response.status_code, 422) self.assertIsNone(self.lifecycle.last_spec) def test_create_normalizes_private_ports_and_default_ssh(self) -> None: cases = ( ({}, (PortRule(22),)), ({"ports": None}, (PortRule(22),)), ({"ports": []}, (PortRule(22),)), ( {"ports": [{"port": 443, "protocol": "HTTPS"}, {"port": 80, "protocol": "HTTP"}]}, (PortRule(22), PortRule(80, "HTTP"), PortRule(443, "HTTPS")), ), ( {"ports": [{"port": 80}, {"port": 22}, {"port": 80, "protocol": "TCP"}]}, (PortRule(22), PortRule(80)), ), ({"ports": [{"port": 22, "protocol": "HTTP"}]}, (PortRule(22, "HTTP"),)), ({"ports": [{"port": 65535}, {"port": 1}]}, (PortRule(1), PortRule(22), PortRule(65535))), ({"no_default_ports": True}, ()), ({"no_default_ports": True, "ports": None}, ()), ({"no_default_ports": True, "ports": []}, ()), ( {"no_default_ports": True, "ports": [{"port": 443, "protocol": "HTTPS"}]}, (PortRule(443, "HTTPS"),), ), ({"no_default_ports": True, "ports": [{"port": 22}]}, (PortRule(22),)), ) for payload, expected in cases: with self.subTest(payload=payload): response = self.client.post("/vms", headers=self.headers, json=payload) self.assertEqual(response.status_code, 201, response.text) self.assertEqual(self.lifecycle.last_spec.ports, expected) self.assertEqual(response.json()["ports"], [rule.to_dict() for rule in expected]) self.assertEqual(response.json()["policy_status"], "enforced") def test_create_rejects_invalid_ports_before_lifecycle_calls(self) -> None: invalid_rules = [ {"port": value} for value in (0, -1, 65536, True, False, 22.0, 22.5, "22", None, {}, []) ] + [ {"port": 80, "protocol": value} for value in ("UDP", "http", "HTTP ", "", None, True, 1) ] + [ {}, {"protocol": "TCP"}, {"port": 80, "public_port": 8080}, {"port": 80, "host_port": 8080}, ] payloads = [{"ports": [rule]} for rule in invalid_rules] + [ {"ports": value} for value in ("22:TCP", {}, [22], ["22:TCP"], [None]) ] + [ {"ports": [{"port": 80}, {"port": 80, "protocol": "HTTP"}]}, {"ports": [{"port": 443, "protocol": "HTTP"}, {"port": 443, "protocol": "HTTPS"}]}, {"public_ports": [80]}, ] + [ {"no_default_ports": value} for value in (None, 0, 1, "true", "false") ] for payload in payloads: with self.subTest(payload=payload): response = self.client.post("/vms", headers=self.headers, json=payload) self.assertEqual(response.status_code, 422, response.text) self.assertIsNone(self.lifecycle.last_spec) def test_list_and_get_whitelist_fields_and_preserve_observed_policy(self) -> None: expected_fields = { "id", "name", "cpu", "ram_mib", "guest_ip", "gateway", "mac", "username", "status", "observed_status", "ports", "policy_status", "operation", "pid", "created_at", "updated_at", "last_error", } self.lifecycle.vm.operation = "ports" self.lifecycle.observed_status = "dead" for ports, policy_status in ( (None, "unmanaged"), ((), "enforced"), ((PortRule(443, "HTTPS"),), "missing"), ((PortRule(22),), "unknown"), ((PortRule(22),), "drift"), ): with self.subTest(ports=ports, policy_status=policy_status): self.lifecycle.vm.ports = ports self.lifecycle.policy_status = policy_status listed = self.client.get("/vms", headers=self.headers) detail = self.client.get("/vms/test-vm", headers=self.headers) self.assertEqual(listed.status_code, 200) self.assertEqual(detail.status_code, 200) self.assertIsInstance(listed.json(), list) self.assertEqual(listed.json(), [detail.json()]) body = detail.json() self.assertEqual(set(body), expected_fields) self.assertEqual(body["name"], "test-vm") self.assertEqual(body["status"], "running") self.assertEqual(body["observed_status"], "dead") self.assertEqual(body["policy_status"], policy_status) self.assertEqual(body["operation"], "ports") self.assertEqual(body["ports"], None if ports is None else [p.to_dict() for p in ports]) self.assertNotIn("guest-secret", detail.text) def test_get_delegates_id_ip_and_name_resolution_to_inspect(self) -> None: with ( patch.object(self.lifecycle, "inspect", wraps=self.lifecycle.inspect) as inspect, patch.object(self.lifecycle, "list_vms", side_effect=AssertionError("must use inspect")), ): for identifier in ("vm-test", "10.42.0.2", "test-vm"): with self.subTest(identifier=identifier): response = self.client.get(f"/vms/{identifier}", headers=self.headers) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["id"], "vm-test") inspect.assert_called_with(identifier) def test_stop_and_destroy_pass_identifiers_to_lifecycle(self) -> None: with ( patch.object(self.lifecycle, "stop", wraps=self.lifecycle.stop) as stop, patch.object(self.lifecycle, "destroy", wraps=self.lifecycle.destroy) as destroy, ): for identifier in ("vm-test", "10.42.0.2", "test-vm"): with self.subTest(identifier=identifier): stopped = self.client.post(f"/vms/{identifier}/stop", headers=self.headers) self.assertEqual(stopped.status_code, 200) self.assertEqual(stopped.json()["id"], "vm-test") self.assertEqual(stopped.json()["policy_status"], "inactive") self.assertEqual(stopped.json()["ports"], [{"port": 22, "protocol": "TCP"}]) stop.assert_called_with(identifier) destroyed = self.client.delete(f"/vms/{identifier}", headers=self.headers) self.assertEqual(destroyed.status_code, 200) self.assertEqual(destroyed.json()["id"], "vm-test") destroy.assert_called_with(identifier) def test_lifecycle_timeouts_are_forwarded_when_supplied(self) -> None: with ( patch.object(self.lifecycle, "restart", wraps=self.lifecycle.restart) as restart, patch.object(self.lifecycle, "stop", wraps=self.lifecycle.stop) as stop, patch.object(self.lifecycle, "destroy", wraps=self.lifecycle.destroy) as destroy, ): self.assertEqual( self.client.post("/vms/vm-test/restart?timeout=1.5", headers=self.headers).status_code, 200, ) self.assertEqual( self.client.post("/vms/vm-test/stop?timeout=2.5", headers=self.headers).status_code, 200, ) self.assertEqual( self.client.delete("/vms/vm-test?timeout=3.5", headers=self.headers).status_code, 200, ) restart.assert_called_once_with("vm-test", timeout=1.5) stop.assert_called_once_with("vm-test", timeout=2.5) destroy.assert_called_once_with("vm-test", timeout=3.5) def test_start_restart_and_port_routes_update_vm_state(self) -> None: started = self.client.post("/vms/vm-test/start", headers=self.headers) self.assertEqual(started.status_code, 200) self.assertEqual(started.json()["status"], "running") self.assertEqual(started.json()["policy_status"], "enforced") restarted = self.client.post("/vms/vm-test/restart", headers=self.headers) self.assertEqual(restarted.status_code, 200) self.assertEqual(restarted.json()["status"], "running") self.lifecycle.vm.status = "stopped" self.lifecycle.policy_status = "inactive" sync = self.client.post("/vms/vm-test/ports/sync", headers=self.headers) self.assertEqual(sync.status_code, 200) self.assertEqual(sync.json()["id"], "vm-test") self.assertEqual(sync.json()["policy_status"], "inactive") def test_bulk_action_resolves_targets_before_running_fixed_ids(self) -> None: with ( patch.object(self.lifecycle, "resolve_targets", wraps=self.lifecycle.resolve_targets) as resolve, patch.object(self.lifecycle, "run_many", wraps=self.lifecycle.run_many) as run_many, ): response = self.client.post( "/vms/actions", headers=self.headers, json={"action": "restart", "identifiers": ["test-vm"], "timeout": 1.5}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), [{"id": "vm-test", "action": "restart", "outcome": "succeeded", "error": None}]) resolve.assert_called_once_with(["test-vm"], all_vms=False) run_many.assert_called_once_with("restart", ["vm-test"], timeout=1.5) def test_bulk_action_requires_exactly_one_selection_strategy(self) -> None: for payload in ( {"action": "start"}, {"action": "start", "identifiers": ["vm-test"], "all_vms": True}, ): with self.subTest(payload=payload): response = self.client.post("/vms/actions", headers=self.headers, json=payload) self.assertEqual(response.status_code, 422) def test_log_and_diagnostic_routes_return_bounded_observability_data(self) -> None: logs = self.client.get("/vms/test-vm/logs?tail=5", headers=self.headers) diagnostics = self.client.get("/diagnostics", headers=self.headers) self.assertEqual(logs.status_code, 200) self.assertEqual(logs.json(), {"content": "last 5 log lines", "truncated": False}) self.assertEqual(diagnostics.status_code, 200) self.assertEqual(diagnostics.json(), [{"name": "kvm", "ok": True, "message": "available"}]) def test_port_update_modes_mutate_rules(self) -> None: set_resp = self.client.post( "/vms/vm-test/ports", headers=self.headers, json={"mode": "set", "ports": [{"port": 80, "protocol": "HTTP"}, {"port": 443, "protocol": "HTTPS"}]}, ) self.assertEqual(set_resp.status_code, 200) self.assertEqual(set_resp.json()["ports"], [{"port": 80, "protocol": "HTTP"}, {"port": 443, "protocol": "HTTPS"}]) open_resp = self.client.post( "/vms/vm-test/ports", headers=self.headers, json={"mode": "open", "ports": [{"port": 22, "protocol": "TCP"}]}, ) self.assertEqual(open_resp.status_code, 200) self.assertEqual(open_resp.json()["ports"], [{"port": 22, "protocol": "TCP"}, {"port": 80, "protocol": "HTTP"}, {"port": 443, "protocol": "HTTPS"}]) close_resp = self.client.post( "/vms/vm-test/ports", headers=self.headers, json={"mode": "close", "ports": [{"port": 80, "protocol": "HTTP"}]}, ) self.assertEqual(close_resp.status_code, 200) self.assertEqual(close_resp.json()["ports"], [{"port": 22, "protocol": "TCP"}, {"port": 443, "protocol": "HTTPS"}]) def test_create_claims_enforcement_only_for_successful_managed_vms(self) -> None: for ports, status, operation, policy_status in ( (None, "running", None, "unmanaged"), ((), "running", None, "enforced"), ((PortRule(22),), "failed", None, "unknown"), ((PortRule(22),), "starting", "create", "unknown"), ((PortRule(22),), "running", "ports", "unknown"), ): with self.subTest(ports=ports, status=status, operation=operation): self.lifecycle.vm.ports = ports self.lifecycle.vm.status = status self.lifecycle.vm.operation = operation with patch.object(self.lifecycle, "create", return_value=self.lifecycle.vm): response = self.client.post("/vms", headers=self.headers, json={}) self.assertEqual(response.status_code, 201) self.assertEqual(response.json()["policy_status"], policy_status) def test_lifecycle_errors_use_typed_status_codes(self) -> None: for method, path, operation in ( ("GET", "/vms", "list_vms"), ("GET", "/vms/test-vm", "inspect"), ("GET", "/vms/test-vm/logs", "read_log_tail"), ("POST", "/vms", "create"), ("POST", "/vms/test-vm/start", "start"), ("POST", "/vms/test-vm/restart", "restart"), ("POST", "/vms/test-vm/stop", "stop"), ("DELETE", "/vms/test-vm", "destroy"), ("POST", "/vms/test-vm/ports", "set_ports"), ("POST", "/vms/test-vm/ports/sync", "sync_ports"), ("GET", "/diagnostics", "diagnostics"), ): for error, expected in ( (NotFoundError("unknown guest"), 404), (ConflictError("ambiguous name"), 409), (ValidationError("invalid request"), 422), (ConfigurationError("invalid settings"), 422), (UvmError("this operation needs root"), 403), (UvmError("host operation failed"), 500), ): with self.subTest(operation=operation, error=error): with patch.object(self.lifecycle, operation, side_effect=error): response = self.client.request(method, path, headers=self.headers, json={}) self.assertEqual(response.status_code, expected) self.assertEqual(response.json(), {"error": {"message": str(error)}}) def test_missing_vm_returns_not_found(self) -> None: response = self.client.get("/vms/missing", headers=self.headers) self.assertEqual(response.status_code, 404) self.assertEqual(response.json(), {"error": {"message": "VM not found: missing"}}) def test_api_schema_exposes_dashboard_and_management_operations(self) -> None: schema = self.client.get("/openapi.json").json() self.assertEqual( {(path, method) for path, methods in schema["paths"].items() for method in methods}, { ("/", "get"), ("/dashboard", "get"), ("/health", "get"), ("/install", "post"), ("/vms", "get"), ("/vms", "post"), ("/vms/actions", "post"), ("/vms/{identifier}", "get"), ("/vms/{identifier}/logs", "get"), ("/vms/{vm_id}/start", "post"), ("/vms/{vm_id}/restart", "post"), ("/vms/{vm_id}/stop", "post"), ("/vms/{vm_id}", "delete"), ("/vms/{vm_id}/ports", "post"), ("/vms/{vm_id}/ports/sync", "post"), ("/diagnostics", "get"), }, ) self.assertEqual( set(schema["components"]["schemas"]["PortRuleModel"]["properties"]), {"port", "protocol"}, ) def test_install_route_uses_the_existing_installer(self) -> None: response = self.client.post("/install", headers=self.headers, json={"force": True}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["firecracker"], "/tmp/firecracker")