kstyagi
645bf6321a
- Introduced `SessionRename` model for renaming sessions with validation on name length and content. - Implemented `rename_session` endpoint in `session_routes.py` to handle session renaming. - Added `delete_session` endpoint to remove sessions, ensuring proper cleanup of associated data. - Enhanced `SessionStore` with `rename` and `delete` methods, including validation and transaction management. - Updated `SessionManager` to manage session renaming and deletion, ensuring live terminal state is preserved. - Added tests for renaming and deleting sessions, ensuring proper functionality and validation. - Introduced a new `tool_catalog.py` for managing guest tool selections and installation recipes.
503 lines
24 KiB
Python
503 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from nest.server import create_app
|
|
from nest.session_store import SessionStore
|
|
from nest.sessions import SessionManager
|
|
from tests.test_server import FASTAPI_AVAILABLE, fake_application
|
|
from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError, StateError, ValidationError
|
|
|
|
|
|
if FASTAPI_AVAILABLE:
|
|
from fastapi.testclient import TestClient
|
|
from starlette.websockets import WebSocketDisconnect
|
|
|
|
|
|
class FakeTerminal:
|
|
def __init__(self) -> None:
|
|
self.events = asyncio.Queue()
|
|
self.events.put_nowait({"type": "output", "data": b"guest$ "})
|
|
self.input = []
|
|
self.sizes = []
|
|
self.closed = False
|
|
self.end_calls = 0
|
|
|
|
async def read(self):
|
|
event = await self.events.get()
|
|
if isinstance(event, Exception):
|
|
raise event
|
|
return event
|
|
|
|
async def write(self, data):
|
|
self.input.append(data)
|
|
await self.events.put({"type": "output", "data": data + b"\r\nguest$ "})
|
|
|
|
async def resize(self, rows, cols):
|
|
self.sizes.append((rows, cols))
|
|
|
|
async def end(self):
|
|
self.end_calls += 1
|
|
await self.events.put({"type": "exit", "exit_code": 0})
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def session_application(base):
|
|
application = fake_application(api_token="test-token")
|
|
application.settings.base = Path(base)
|
|
application.terminals = []
|
|
application.vm_ids = {"vm-test", "vm-other"}
|
|
|
|
def get(vm_id):
|
|
if vm_id not in application.vm_ids:
|
|
raise NotFoundError("VM not found")
|
|
return SimpleNamespace(id=vm_id)
|
|
|
|
def open_terminal(vm_id, **_size):
|
|
get(vm_id)
|
|
terminal = FakeTerminal()
|
|
application.terminals.append(terminal)
|
|
return terminal, "worker"
|
|
|
|
application.state_store = SimpleNamespace(get=get)
|
|
application.lifecycle = SimpleNamespace(open_terminal=open_terminal)
|
|
return application
|
|
|
|
|
|
class SessionStoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.path = Path(temporary.name) / "nest" / "sessions.sqlite3"
|
|
self.store = SessionStore(self.path, max_output=8)
|
|
self.addCleanup(self.store.close)
|
|
|
|
def test_history_is_scoped_by_exact_vm_id_and_survives_reopening(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.activate(record["id"], "worker")
|
|
self.store.append(record["id"], b"hello")
|
|
self.store.finish(record["id"], "closed", 7)
|
|
self.assertEqual(self.store.list("vm-b"), [])
|
|
with self.assertRaises(NotFoundError):
|
|
self.store.get("vm-b", record["id"])
|
|
self.store.close()
|
|
reopened = SessionStore(self.path)
|
|
self.addCleanup(reopened.close)
|
|
saved, start, output = reopened.read("vm-a", record["id"], 0)
|
|
self.assertEqual((saved["status"], saved["exit_code"], saved["username"], start, output), ("closed", 7, "worker", 0, b"hello"))
|
|
self.assertEqual(self.path.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(self.path.parent.stat().st_mode & 0o777, 0o700)
|
|
|
|
def test_restarted_server_marks_only_live_sessions_interrupted(self):
|
|
live = self.store.create("vm-a", 24, 80)
|
|
ended = self.store.create("vm-a", 24, 80)
|
|
self.store.activate(live["id"], "worker")
|
|
self.store.finish(ended["id"], "closed", 0)
|
|
self.store.close()
|
|
reopened = SessionStore(self.path)
|
|
self.addCleanup(reopened.close)
|
|
self.assertEqual(reopened.get("vm-a", live["id"])["status"], "interrupted")
|
|
self.assertIsNotNone(reopened.get("vm-a", live["id"])["ended_at"])
|
|
self.assertEqual(reopened.get("vm-a", ended["id"])["status"], "closed")
|
|
|
|
def test_transcript_is_bounded_and_offsets_support_exact_replay(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.append(record["id"], b"abcdefghij")
|
|
self.store.append(record["id"], b"kl")
|
|
saved, start, output = self.store.read("vm-a", record["id"], 0)
|
|
self.assertEqual((start, output, saved["output_end"], saved["truncated"]), (4, b"efghijkl", 12, True))
|
|
self.assertEqual(self.store.read("vm-a", record["id"], 9)[1:], (9, b"jkl"))
|
|
self.assertEqual(self.store.read("vm-a", record["id"], 12)[2], b"")
|
|
|
|
def test_second_server_cannot_reconcile_another_servers_live_sessions(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.activate(record["id"], "worker")
|
|
with self.assertRaises(ConflictError):
|
|
SessionStore(self.path)
|
|
self.assertEqual(self.store.get("vm-a", record["id"])["status"], "active")
|
|
|
|
def test_symlinked_history_file_is_not_opened_or_rewritten(self):
|
|
self.store.close()
|
|
self.path.unlink()
|
|
target = self.path.parent / "unrelated"
|
|
target.write_bytes(b"keep this")
|
|
self.path.symlink_to(target)
|
|
with self.assertRaises(OSError):
|
|
SessionStore(self.path)
|
|
self.assertEqual(target.read_bytes(), b"keep this")
|
|
|
|
def test_rename_preserves_session_identity_state_and_output_across_reopening(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.activate(record["id"], "worker")
|
|
self.store.append(record["id"], b"history")
|
|
before = self.store.get("vm-a", record["id"])
|
|
renamed = self.store.rename("vm-a", record["id"], " Déploiement 'alpha' ")
|
|
self.assertEqual(renamed["name"], "Déploiement 'alpha'")
|
|
for key in before.keys() - {"name", "updated_at"}:
|
|
self.assertEqual(renamed[key], before[key])
|
|
self.store.close()
|
|
reopened = SessionStore(self.path)
|
|
self.addCleanup(reopened.close)
|
|
saved, _start, output = reopened.read("vm-a", record["id"], 0)
|
|
self.assertEqual(saved["name"], renamed["name"])
|
|
self.assertEqual(output, b"history")
|
|
|
|
def test_invalid_session_names_do_not_change_the_record(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
for name in ("", " ", "a" * 81, "two\nlines", "bad\x00name", "\ud800", 123, None):
|
|
with self.subTest(name=repr(name)), self.assertRaises(ValidationError):
|
|
self.store.rename("vm-a", record["id"], name)
|
|
self.assertEqual(self.store.get("vm-a", record["id"]), record)
|
|
|
|
def test_rename_and_delete_cannot_cross_vm_boundaries(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.finish(record["id"], "closed", 0)
|
|
for operation in (lambda: self.store.rename("vm-b", record["id"], "other"),
|
|
lambda: self.store.delete("vm-b", record["id"])):
|
|
with self.assertRaises(NotFoundError):
|
|
operation()
|
|
self.assertEqual(self.store.get("vm-a", record["id"])["name"], record["name"])
|
|
|
|
def test_delete_removes_metadata_and_transcript_and_keeps_other_sessions(self):
|
|
first = self.store.create("vm-a", 24, 80)
|
|
second = self.store.create("vm-a", 24, 80)
|
|
self.store.append(first["id"], b"remove")
|
|
self.store.append(second["id"], b"keep")
|
|
self.store.finish(first["id"], "closed", 0)
|
|
self.store.delete("vm-a", first["id"])
|
|
with self.assertRaises(NotFoundError):
|
|
self.store.read("vm-a", first["id"], 0)
|
|
with sqlite3.connect(self.path) as db:
|
|
self.assertEqual(db.execute("SELECT COUNT(*) FROM output WHERE session_id=?", (first["id"],)).fetchone()[0], 0)
|
|
self.assertEqual(self.store.read("vm-a", second["id"], 0)[2], b"keep")
|
|
|
|
def test_failed_delete_rolls_back_metadata_and_output_together(self):
|
|
record = self.store.create("vm-a", 24, 80)
|
|
self.store.append(record["id"], b"keep")
|
|
self.store.finish(record["id"], "closed", 0)
|
|
with sqlite3.connect(self.path) as db:
|
|
db.execute("CREATE TRIGGER reject_delete BEFORE DELETE ON sessions BEGIN SELECT RAISE(ABORT, 'test failure'); END")
|
|
with self.assertRaises(StateError):
|
|
self.store.delete("vm-a", record["id"])
|
|
self.assertEqual(self.store.read("vm-a", record["id"], 0)[2], b"keep")
|
|
|
|
def test_running_history_requires_closure_and_default_names_avoid_existing_labels(self):
|
|
records = [self.store.create("vm-a", 24, 80) for _ in range(3)]
|
|
with self.assertRaises(ConflictError):
|
|
self.store.delete("vm-a", records[1]["id"])
|
|
self.store.finish(records[1]["id"], "closed", 0)
|
|
self.store.delete("vm-a", records[1]["id"])
|
|
created = self.store.create("vm-a", 24, 80)
|
|
self.assertNotIn(created["name"], {records[0]["name"], records[2]["name"]})
|
|
|
|
|
|
class SessionManagerTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.application = session_application(self.temporary.name)
|
|
self.manager = SessionManager(self.application)
|
|
self.addAsyncCleanup(self.manager.shutdown)
|
|
|
|
async def test_failed_launch_is_saved_in_history(self):
|
|
def unavailable(*_args, **_kwargs):
|
|
raise GuestAgentUnavailableError("update the guest agent")
|
|
|
|
self.application.lifecycle.open_terminal = unavailable
|
|
with self.assertRaises(GuestAgentUnavailableError):
|
|
await self.manager.create("vm-test", 24, 80)
|
|
history = await self.manager.list("vm-test")
|
|
self.assertEqual(history[0]["status"], "failed")
|
|
self.assertEqual(history[0]["error"], "update the guest agent")
|
|
|
|
async def test_shutdown_closes_live_channels_and_saves_interrupted_history(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
await self.manager.shutdown()
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
store = SessionStore(Path(self.temporary.name) / "nest" / "sessions.sqlite3")
|
|
self.addCleanup(store.close)
|
|
self.assertEqual(store.get("vm-test", record["id"])["status"], "interrupted")
|
|
|
|
async def test_cancelling_the_create_request_does_not_orphan_the_shell(self):
|
|
import threading
|
|
entered, release = threading.Event(), threading.Event()
|
|
connect = self.application.lifecycle.open_terminal
|
|
|
|
def slow_connect(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(2)
|
|
return connect(*args, **kwargs)
|
|
|
|
self.application.lifecycle.open_terminal = slow_connect
|
|
request = asyncio.create_task(self.manager.create("vm-test", 24, 80))
|
|
await asyncio.to_thread(entered.wait, 2)
|
|
request.cancel()
|
|
release.set()
|
|
await asyncio.gather(request, return_exceptions=True)
|
|
await self.manager.shutdown()
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
|
|
async def test_guest_loss_ends_the_session_and_retains_output(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
terminal = self.application.terminals[0]
|
|
await terminal.events.put(GuestExecutionError("VM connection lost", outcome_unknown=True))
|
|
async with asyncio.timeout(2):
|
|
while (await self.manager.get("vm-test", record["id"]))["status"] == "active":
|
|
await asyncio.sleep(0.01)
|
|
saved, _start, output = await self.manager.read("vm-test", record["id"], 0)
|
|
self.assertEqual(saved["status"], "interrupted")
|
|
self.assertEqual(output, b"guest$ ")
|
|
self.assertTrue(terminal.closed)
|
|
|
|
async def test_shutdown_finishes_inflight_output_before_releasing_history_ownership(self):
|
|
import threading
|
|
entered, release = threading.Event(), threading.Event()
|
|
history = self.manager._history()
|
|
append = history.append
|
|
|
|
def slow_append(*args):
|
|
entered.set()
|
|
release.wait(2)
|
|
append(*args)
|
|
|
|
with patch.object(history, "append", side_effect=slow_append):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
self.assertTrue(await asyncio.to_thread(entered.wait, 2))
|
|
closing = asyncio.create_task(self.manager.shutdown())
|
|
try:
|
|
await asyncio.sleep(0.03)
|
|
self.assertFalse(closing.done())
|
|
with self.assertRaises(ConflictError):
|
|
SessionStore(history.path)
|
|
finally:
|
|
release.set()
|
|
await closing
|
|
reopened = SessionStore(history.path)
|
|
self.addCleanup(reopened.close)
|
|
self.assertEqual(reopened.read("vm-test", record["id"], 0)[2], b"guest$ ")
|
|
|
|
async def test_rename_keeps_the_same_live_terminal(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
terminal = self.application.terminals[0]
|
|
renamed = await self.manager.rename("vm-test", record["id"], "Deployment")
|
|
self.assertEqual((renamed["id"], renamed["status"], renamed["name"]), (record["id"], "active", "Deployment"))
|
|
await self.manager.write("vm-test", record["id"], b"pwd\r")
|
|
self.assertEqual(terminal.input, [b"pwd\r"])
|
|
self.assertFalse(terminal.closed)
|
|
self.assertEqual(len(self.application.terminals), 1)
|
|
|
|
async def test_delete_closes_only_its_live_terminal_and_removes_all_history(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
other = await self.manager.create("vm-test", 24, 80)
|
|
await self.manager.write("vm-test", record["id"], b"last output\r")
|
|
await self.manager.delete("vm-test", record["id"])
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
self.assertFalse(self.application.terminals[1].closed)
|
|
self.assertEqual([item["id"] for item in await self.manager.list("vm-test")], [other["id"]])
|
|
with sqlite3.connect(self.manager._history().path) as db:
|
|
self.assertEqual(db.execute("SELECT COUNT(*) FROM output WHERE session_id=?", (record["id"],)).fetchone()[0], 0)
|
|
|
|
async def test_delete_waits_for_an_inflight_launch_before_closing_its_channel(self):
|
|
import threading
|
|
entered, release = threading.Event(), threading.Event()
|
|
connect = self.application.lifecycle.open_terminal
|
|
|
|
def slow_connect(*args, **kwargs):
|
|
entered.set()
|
|
release.wait(2)
|
|
return connect(*args, **kwargs)
|
|
|
|
self.application.lifecycle.open_terminal = slow_connect
|
|
creating = asyncio.create_task(self.manager.create("vm-test", 24, 80))
|
|
self.assertTrue(await asyncio.to_thread(entered.wait, 2))
|
|
record = (await self.manager.list("vm-test"))[0]
|
|
deleting = asyncio.create_task(self.manager.delete("vm-test", record["id"]))
|
|
try:
|
|
await asyncio.sleep(0.03)
|
|
self.assertFalse(deleting.done())
|
|
finally:
|
|
release.set()
|
|
await asyncio.gather(creating, deleting)
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
self.assertEqual(await self.manager.list("vm-test"), [])
|
|
|
|
async def test_cancelled_delete_request_finishes_cleanup_even_during_shutdown(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
terminal = self.application.terminals[0]
|
|
entered, release = asyncio.Event(), asyncio.Event()
|
|
end = terminal.end
|
|
|
|
async def slow_end():
|
|
entered.set()
|
|
await release.wait()
|
|
await end()
|
|
|
|
terminal.end = slow_end
|
|
request = asyncio.create_task(self.manager.delete("vm-test", record["id"]))
|
|
await asyncio.wait_for(entered.wait(), timeout=2)
|
|
request.cancel()
|
|
await asyncio.gather(request, return_exceptions=True)
|
|
release.set()
|
|
await self.manager.shutdown()
|
|
reopened = SessionStore(Path(self.temporary.name) / "nest" / "sessions.sqlite3")
|
|
self.addCleanup(reopened.close)
|
|
self.assertTrue(terminal.closed)
|
|
self.assertEqual(reopened.list("vm-test"), [])
|
|
|
|
async def test_close_and_delete_are_serialized_for_the_same_session(self):
|
|
record = await self.manager.create("vm-test", 24, 80)
|
|
results = await asyncio.gather(self.manager.end("vm-test", record["id"]),
|
|
self.manager.delete("vm-test", record["id"]), return_exceptions=True)
|
|
self.assertTrue(all(not isinstance(result, Exception) or isinstance(result, NotFoundError) for result in results))
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
self.assertEqual(self.application.terminals[0].end_calls, 1)
|
|
self.assertEqual(await self.manager.list("vm-test"), [])
|
|
|
|
|
|
@unittest.skipUnless(FASTAPI_AVAILABLE, "FastAPI and httpx are required")
|
|
class SessionApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = self.enterContext(tempfile.TemporaryDirectory())
|
|
self.application = session_application(self.temporary)
|
|
self.api = create_app(self.application, host="127.0.0.1")
|
|
self.client = self.enterContext(TestClient(self.api))
|
|
self.headers = {"X-UVM-Token": "test-token"}
|
|
self.base = "/vms/vm-test/sessions"
|
|
|
|
def create(self):
|
|
response = self.client.post(self.base, headers=self.headers, json={})
|
|
self.assertEqual(response.status_code, 201, response.text)
|
|
return response.json()
|
|
|
|
def connect(self, record, *, vm_id="vm-test", offset=0):
|
|
ws = self.enterContext(self.client.websocket_connect(f"/vms/{vm_id}/sessions/{record['id']}/terminal"))
|
|
ws.send_json({"type": "auth", "token": "test-token", "offset": offset})
|
|
return ws
|
|
|
|
def output(self, ws):
|
|
while True:
|
|
message = ws.receive_json()
|
|
if message["type"] == "output":
|
|
data = base64.b64decode(message["data"])
|
|
end = message["offset"] + len(data)
|
|
ws.send_json({"type": "ack", "offset": end})
|
|
return data, end
|
|
self.assertEqual(message["type"], "session", message)
|
|
|
|
def test_all_http_operations_and_websocket_output_require_authentication(self):
|
|
for method, path in (("GET", self.base), ("POST", self.base),
|
|
("GET", self.base + "/session-unknown"), ("POST", self.base + "/session-unknown/close"),
|
|
("PATCH", self.base + "/session-unknown"), ("DELETE", self.base + "/session-unknown")):
|
|
self.assertEqual(self.client.request(method, path, json={}).status_code, 401)
|
|
for token in ("wrong", "", "\u00e9"):
|
|
with self.client.websocket_connect(self.base + "/session-unknown/terminal") as ws:
|
|
ws.send_json({"type": "auth", "token": token})
|
|
with self.assertRaises(WebSocketDisconnect) as raised:
|
|
ws.receive_json()
|
|
self.assertEqual(raised.exception.code, 4401)
|
|
self.assertEqual(self.application.terminals, [])
|
|
self.assertFalse((Path(self.temporary) / "nest").exists())
|
|
|
|
def test_foreign_browser_origin_is_rejected(self):
|
|
with self.assertRaises(WebSocketDisconnect):
|
|
with self.client.websocket_connect(self.base + "/unknown/terminal", headers={"Origin": "https://foreign.example"}):
|
|
self.fail("foreign origin accepted")
|
|
|
|
def test_create_rejects_unknown_vm_aliases_and_invalid_sizes(self):
|
|
for body in ({"rows": 0}, {"cols": True}, {"cols": 401}, {"command": "id"}):
|
|
self.assertEqual(self.client.post(self.base, headers=self.headers, json=body).status_code, 422)
|
|
self.assertEqual(self.client.post("/vms/name-alias/sessions", headers=self.headers, json={}).status_code, 404)
|
|
self.assertEqual(self.application.terminals, [])
|
|
|
|
def test_sessions_are_nested_under_the_selected_vm(self):
|
|
record = self.create()
|
|
self.assertEqual(record["username"], "worker")
|
|
other = "/vms/vm-other/sessions"
|
|
self.assertEqual(self.client.get(other, headers=self.headers).json(), [])
|
|
self.assertEqual(self.client.get(other + "/" + record["id"], headers=self.headers).status_code, 404)
|
|
self.assertEqual(self.client.post(other + "/" + record["id"] + "/close", headers=self.headers).status_code, 404)
|
|
self.assertEqual(self.client.patch(other + "/" + record["id"], headers=self.headers, json={"name": "Other"}).status_code, 404)
|
|
self.assertEqual(self.client.delete(other + "/" + record["id"], headers=self.headers).status_code, 404)
|
|
ws = self.connect(record, vm_id="vm-other")
|
|
self.assertEqual(ws.receive_json()["type"], "error")
|
|
self.assertFalse(self.application.terminals[0].closed)
|
|
|
|
def test_disconnect_reconnect_input_resize_and_closed_history(self):
|
|
record = self.create()
|
|
ws = self.connect(record)
|
|
output, offset = self.output(ws)
|
|
self.assertEqual(output, b"guest$ ")
|
|
ws.send_json({"type": "resize", "rows": 35, "cols": 120})
|
|
ws.send_json({"type": "input", "data": base64.b64encode(b"pwd\r").decode()})
|
|
output, offset = self.output(ws)
|
|
self.assertEqual(output, b"pwd\r\r\nguest$ ")
|
|
self.assertEqual(self.application.terminals[0].sizes, [(35, 120)])
|
|
ws.close()
|
|
self.assertFalse(self.application.terminals[0].closed)
|
|
reopened = self.connect(record, offset=offset)
|
|
self.assertEqual(reopened.receive_json()["session"]["status"], "active")
|
|
self.assertEqual(len(self.application.terminals), 1)
|
|
closed = self.client.post(self.base + "/" + record["id"] + "/close", headers=self.headers)
|
|
self.assertEqual(closed.json()["status"], "closed")
|
|
self.assertEqual(closed.json()["exit_code"], 0)
|
|
replay = self.connect(record)
|
|
transcript, _end = self.output(replay)
|
|
self.assertEqual(transcript, b"guest$ pwd\r\r\nguest$ ")
|
|
self.assertEqual(self.client.get(self.base, headers=self.headers).json()[0]["status"], "closed")
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
|
|
def test_malformed_input_ends_only_the_browser_attachment(self):
|
|
record = self.create()
|
|
ws = self.connect(record)
|
|
self.output(ws)
|
|
ws.send_json({"type": "input", "data": "not base64!"})
|
|
while (message := ws.receive_json())["type"] == "session":
|
|
pass
|
|
self.assertEqual(message["type"], "error")
|
|
self.assertEqual(self.application.terminals[0].input, [])
|
|
self.assertFalse(self.application.terminals[0].closed)
|
|
self.assertEqual(self.client.get(self.base + "/" + record["id"], headers=self.headers).json()["status"], "active")
|
|
|
|
def test_rename_is_validated_persisted_and_sent_to_attached_terminals(self):
|
|
record = self.create()
|
|
url = self.base + "/" + record["id"]
|
|
for body in ({}, {"name": " "}, {"name": 123}, {"name": "x" * 81},
|
|
{"name": "two\nlines"}, {"name": "bad\x00name"}, {"name": "New", "vm_id": "vm-other"}):
|
|
with self.subTest(body=body):
|
|
self.assertEqual(self.client.patch(url, headers=self.headers, json=body).status_code, 422)
|
|
ws = self.connect(record)
|
|
self.output(ws)
|
|
response = self.client.patch(url, headers=self.headers, json={"name": " Déploiement "})
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual(response.json()["name"], "Déploiement")
|
|
while (message := ws.receive_json()).get("session", {}).get("name") != "Déploiement":
|
|
self.assertEqual(message["type"], "session")
|
|
self.assertEqual(message["session"]["status"], "active")
|
|
self.assertEqual(self.client.get(url, headers=self.headers).json()["name"], "Déploiement")
|
|
self.assertFalse(self.application.terminals[0].closed)
|
|
|
|
def test_delete_live_session_returns_no_content_and_notifies_attached_browser(self):
|
|
record = self.create()
|
|
ws = self.connect(record)
|
|
self.output(ws)
|
|
url = self.base + "/" + record["id"]
|
|
response = self.client.delete(url, headers=self.headers)
|
|
self.assertEqual(response.status_code, 204, response.text)
|
|
self.assertEqual(response.content, b"")
|
|
while (message := ws.receive_json())["type"] == "session":
|
|
pass
|
|
self.assertEqual(message, {"type": "deleted", "session_id": record["id"]})
|
|
self.assertTrue(self.application.terminals[0].closed)
|
|
self.assertEqual(self.client.get(url, headers=self.headers).status_code, 404)
|
|
self.assertEqual(self.client.get(self.base, headers=self.headers).json(), [])
|