feat: introduce Nest workspace for local Firecracker management
- Added a new server module for the Nest workspace, integrating with UVM's API. - Updated project description to reflect the new Nest functionality. - Created tests for the Nest server and CLI, ensuring proper functionality and error handling. - Refactored existing UVM server code to separate Nest and UVM functionalities. - Removed the old UVM dashboard assets package as it is now part of the Nest workspace. - Enhanced error messages for clarity regarding server shutdown commands.
This commit is contained in:
@@ -0,0 +1,173 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import sys
|
||||
import unittest
|
||||
from contextlib import redirect_stderr
|
||||
from html.parser import HTMLParser
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from nest.cli import main
|
||||
from nest.server import create_app, run_server
|
||||
from tests.test_server import FASTAPI_AVAILABLE, FakeLifecycle, fake_application
|
||||
from uvm.errors import UvmError
|
||||
|
||||
|
||||
if FASTAPI_AVAILABLE:
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
class NestCliTests(unittest.TestCase):
|
||||
def test_serve_launches_the_nest_workspace(self) -> None:
|
||||
application = fake_application(api_token="test-token")
|
||||
with (
|
||||
patch("nest.cli.build_application", return_value=application),
|
||||
patch("nest.server.run_server") as launch,
|
||||
):
|
||||
result = main(["--serve", "--host", "127.0.0.1", "--port", "8123"])
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
launch.assert_called_once_with(application, host="127.0.0.1", port=8123)
|
||||
|
||||
def test_invalid_invocations_do_not_build_host_services(self) -> None:
|
||||
for arguments in (
|
||||
[], ["--serve", "list"], ["--serve", "--port", "0"],
|
||||
["--serve", "--port", "65536"], ["--serve", "--port", "1.5"],
|
||||
):
|
||||
with self.subTest(arguments=arguments), redirect_stderr(io.StringIO()):
|
||||
with patch("nest.cli.build_application") as build:
|
||||
with self.assertRaises(SystemExit) as raised:
|
||||
main(arguments)
|
||||
self.assertEqual(raised.exception.code, 2)
|
||||
build.assert_not_called()
|
||||
|
||||
def test_startup_errors_are_reported_as_nest_errors(self) -> None:
|
||||
output = io.StringIO()
|
||||
with (
|
||||
patch("nest.cli.build_application", side_effect=UvmError("invalid settings")),
|
||||
redirect_stderr(output),
|
||||
):
|
||||
result = main(["--serve"])
|
||||
|
||||
self.assertEqual(result, 1)
|
||||
self.assertEqual(output.getvalue(), "nest: error: invalid settings\n")
|
||||
|
||||
def test_factory_requires_an_explicit_host(self) -> None:
|
||||
with self.assertRaisesRegex(UvmError, "nest --serve"):
|
||||
create_app(fake_application(api_token="test-token"))
|
||||
|
||||
def test_workspace_enforces_backend_token_and_tls_requirements(self) -> None:
|
||||
for token, host in ((None, "127.0.0.1"), ("test-token", "0.0.0.0")):
|
||||
for launch in (create_app, run_server):
|
||||
with self.subTest(token=token, host=host, launch=launch.__name__):
|
||||
options = {"port": 8000} if launch is run_server else {}
|
||||
with self.assertRaises(UvmError):
|
||||
launch(fake_application(api_token=token), host=host, **options)
|
||||
|
||||
|
||||
class _DashboardAssets(HTMLParser):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.urls: list[str] = []
|
||||
|
||||
def handle_starttag(self, tag, attrs) -> None:
|
||||
attributes = dict(attrs)
|
||||
if tag == "link" and attributes.get("rel") in {"preload", "stylesheet"}:
|
||||
self.urls.append(attributes["href"])
|
||||
if tag == "script" and attributes.get("src"):
|
||||
self.urls.append(attributes["src"])
|
||||
|
||||
|
||||
@unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required for workspace tests")
|
||||
class NestWorkspaceTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.application = fake_application(api_token="test-token")
|
||||
self.application.lifecycle = FakeLifecycle()
|
||||
self.application.installer = SimpleNamespace(install=Mock())
|
||||
self.headers = {"X-UVM-Token": "test-token"}
|
||||
self.api = create_app(self.application, host="127.0.0.1")
|
||||
self.client = TestClient(self.api)
|
||||
self.addCleanup(self.client.close)
|
||||
|
||||
def test_nest_dashboard_and_its_linked_assets_are_public(self) -> None:
|
||||
root = self.client.get("/", follow_redirects=False)
|
||||
self.assertEqual(root.status_code, 307)
|
||||
self.assertEqual(root.headers["location"], "/dashboard")
|
||||
dashboard = self.client.get(root.headers["location"])
|
||||
self.assertEqual(dashboard.status_code, 200)
|
||||
self.assertIn("<title>Nest</title>", dashboard.text)
|
||||
self.assertIn("Welcome to Nest", dashboard.text)
|
||||
self.assertNotIn("test-token", dashboard.text)
|
||||
|
||||
assets = _DashboardAssets()
|
||||
assets.feed(dashboard.text)
|
||||
self.assertTrue(assets.urls)
|
||||
for url in assets.urls:
|
||||
with self.subTest(url=url):
|
||||
self.assertTrue(url.startswith("/dashboard/static/"))
|
||||
response = self.client.get(url)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.content)
|
||||
self.assertNotIn("text/html", response.headers["content-type"])
|
||||
|
||||
font = self.client.get("/dashboard/static/fonts/manrope-latin.woff2")
|
||||
self.assertTrue(font.content.startswith(b"wOF2"))
|
||||
license_response = self.client.get("/dashboard/static/fonts/OFL.txt")
|
||||
self.assertEqual(license_response.status_code, 200)
|
||||
self.assertIn("SIL OPEN FONT LICENSE", license_response.text)
|
||||
self.assertEqual(self.client.get("/dashboard/static/missing.js").status_code, 404)
|
||||
|
||||
def test_machine_controls_use_the_authenticated_backend(self) -> None:
|
||||
for method, path in (
|
||||
("GET", "/vms"), ("GET", "/diagnostics"),
|
||||
("POST", "/vms/vm-test/stop"), ("POST", "/install"), ("POST", "/exit"),
|
||||
):
|
||||
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.application.installer.install.assert_not_called()
|
||||
self.assertEqual(self.application.lifecycle.vm.status, "running")
|
||||
|
||||
listed = self.client.get("/vms", headers=self.headers)
|
||||
self.assertEqual(listed.status_code, 200)
|
||||
self.assertEqual(listed.json()[0]["id"], "vm-test")
|
||||
self.assertNotIn("password", listed.json()[0])
|
||||
stopped = self.client.post("/vms/vm-test/stop", headers=self.headers)
|
||||
self.assertEqual(stopped.status_code, 200)
|
||||
self.assertEqual(stopped.json()["status"], "stopped")
|
||||
self.assertEqual(self.application.lifecycle.vm.status, "stopped")
|
||||
|
||||
def test_workspace_preserves_api_errors_and_liveness(self) -> None:
|
||||
missing = self.client.get("/vms/missing", headers=self.headers)
|
||||
self.assertEqual(missing.status_code, 404)
|
||||
self.assertEqual(missing.json(), {"error": {"message": "VM not found: missing"}})
|
||||
health = self.client.get("/health")
|
||||
self.assertEqual(health.status_code, 200)
|
||||
self.assertEqual(health.json(), {"status": "ok"})
|
||||
|
||||
def test_workspace_api_reference_includes_the_dashboard_and_machine_routes(self) -> None:
|
||||
schema = self.client.get("/openapi.json").json()
|
||||
self.assertEqual(schema["info"]["title"], "Nest")
|
||||
for path in ("/", "/dashboard", "/vms", "/diagnostics", "/vms/{vm_id}/resize"):
|
||||
self.assertIn(path, schema["paths"])
|
||||
|
||||
def test_launcher_serves_nest_and_wires_graceful_shutdown(self) -> None:
|
||||
server = SimpleNamespace(run=Mock(), should_exit=False)
|
||||
config = object()
|
||||
fake_uvicorn = SimpleNamespace(Config=Mock(return_value=config), Server=Mock(return_value=server))
|
||||
with patch.dict(sys.modules, {"uvicorn": fake_uvicorn}):
|
||||
run_server(self.application, host="127.0.0.1", port=8123)
|
||||
|
||||
api = fake_uvicorn.Config.call_args.args[0]
|
||||
fake_uvicorn.Config.assert_called_once_with(api, host="127.0.0.1", port=8123)
|
||||
fake_uvicorn.Server.assert_called_once_with(config)
|
||||
server.run.assert_called_once_with()
|
||||
self.assertIs(api.state.uvicorn_server, server)
|
||||
with TestClient(api) as client:
|
||||
self.assertIn("<title>Nest</title>", client.get("/dashboard").text)
|
||||
response = client.post("/exit", headers=self.headers)
|
||||
self.assertEqual(response.status_code, 202)
|
||||
self.assertEqual(response.json(), {"status": "stopping"})
|
||||
self.assertTrue(server.should_exit)
|
||||
+8
-18
@@ -322,20 +322,10 @@ class ApiRouterTests(unittest.TestCase):
|
||||
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_standalone_uvm_api_does_not_serve_the_nest_dashboard(self) -> None:
|
||||
for path in ("/", "/dashboard", "/dashboard/static/js/dashboard.js"):
|
||||
with self.subTest(path=path):
|
||||
self.assertEqual(self.client.get(path).status_code, 404)
|
||||
|
||||
def test_vm_routes_require_and_accept_the_api_token(self) -> None:
|
||||
for method, path, payload in (
|
||||
@@ -381,7 +371,9 @@ class ApiRouterTests(unittest.TestCase):
|
||||
response = self.client.post("/exit", headers=self.headers)
|
||||
|
||||
self.assertEqual(response.status_code, 503)
|
||||
self.assertEqual(response.json(), {"detail": "server shutdown is unavailable outside uvm --serve"})
|
||||
self.assertEqual(response.json(), {
|
||||
"detail": "server shutdown is unavailable outside nest --serve or uvm --serve",
|
||||
})
|
||||
|
||||
def test_vm_lifecycle_routes_and_extra_field_validation(self) -> None:
|
||||
created = self.client.post(
|
||||
@@ -773,14 +765,12 @@ class ApiRouterTests(unittest.TestCase):
|
||||
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:
|
||||
def test_api_schema_exposes_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"),
|
||||
("/exit", "post"),
|
||||
("/install", "post"),
|
||||
|
||||
Reference in New Issue
Block a user