kstyagi
dca7f78dae
- 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.
174 lines
8.0 KiB
Python
174 lines
8.0 KiB
Python
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)
|