kstyagi
d8a25209b9
- Introduced a new method `installed_kernel` in `ImageStore` to handle kernel verification and checksum validation. - Updated `Installer` to include additional packages: `nftables` and `conntrack`. - Refined `LifecycleService` to improve error handling and state management during VM creation, starting, stopping, and deletion. - Enhanced port management with new methods for setting and syncing ports, including validation for port rules. - Improved error handling in `StateStore` to raise specific `NotFoundError` and `ConflictError` exceptions. - Updated network management to ensure proper TAP device handling and firewall integration. - Enhanced API responses in `vms.py` to include policy status for better client-side handling. - Added validation for VM names and ports in `validation.py` to enforce stricter rules and prevent conflicts.
508 lines
22 KiB
Python
508 lines
22 KiB
Python
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
|
|
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 stop(self, identifier: str) -> 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) -> VmRecord:
|
|
self.inspect(identifier)
|
|
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_vm_routes_require_and_accept_the_api_token(self) -> None:
|
|
for method, path in (
|
|
("GET", "/vms"),
|
|
("POST", "/vms"),
|
|
("GET", "/vms/test-vm"),
|
|
("POST", "/vms/test-vm/stop"),
|
|
("DELETE", "/vms/test-vm"),
|
|
("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={})
|
|
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_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"),
|
|
("POST", "/vms", "create"),
|
|
("POST", "/vms/test-vm/stop", "stop"),
|
|
("DELETE", "/vms/test-vm", "destroy"),
|
|
):
|
|
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_keeps_only_existing_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},
|
|
{
|
|
("/health", "get"),
|
|
("/install", "post"),
|
|
("/vms", "get"),
|
|
("/vms", "post"),
|
|
("/vms/{identifier}", "get"),
|
|
("/vms/{vm_id}/stop", "post"),
|
|
("/vms/{vm_id}", "delete"),
|
|
},
|
|
)
|
|
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")
|