mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-10 12:16:49 +00:00
feat(console): a working serial console
The xterm console never connected: socketiod expected a different token format, its worker had no compute credentials, one global console was shared by every client and its output broadcast to all of them, and the Docker image did not run socketiod at all. - console/tokens.py: one authorize() for noVNC and socketiod; the signed token is the only credential, checked at connect time. - socketiod: one pty worker per client (consolecallback with the instance id), one session per VM, a session cap, input and resize checked; the parent never opens libvirt. The engine.io same-origin check is on. - consolecallback: opens the console with the compute's credentials and exits with a message when it cannot. - Docker runs socketiod as a runit service; nginx proxies /socket.io/ with Host and the scheme for the origin check. Supervisor and systemd start it as well; WEBVIRTCLOUD_SERIAL_CONSOLE=1 enables it. - Tests: token and session unit tests, socketiod as a process, and a live test that reads SeaBIOS from a wvc-test VM on the TCP and SSH hosts.
This commit is contained in:
catborise
parent
9b0c6e2d8e
commit
d37aa42392
20 changed files with 592 additions and 569 deletions
79
console/tests/test_live_serial.py
Normal file
79
console/tests/test_live_serial.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
"""Live serial console: socketiod opens a running wvc-test VM's console on the
|
||||
TCP host (TEST_LIBVIRT_HOST, with its password) and on the SSH host
|
||||
(TEST_LIBVIRT_SSH_HOST). The VM has no graphics and no OS; SeaBIOS writes its
|
||||
banner to the serial port. Runs only with TEST_LIBVIRT_HOST set."""
|
||||
|
||||
import os
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import libvirt
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TransactionTestCase
|
||||
from vrtManager.connection import connection_manager
|
||||
|
||||
from accounts.models import UserInstance
|
||||
from computes.models import Compute
|
||||
from console import tokens
|
||||
from console.tests.test_socketiod_child import children, console, start_socketiod, socketio
|
||||
from instances.models import Instance
|
||||
from instances.tests import livetest
|
||||
|
||||
P = livetest.PREFIX
|
||||
|
||||
|
||||
def serial_vm(conn, name):
|
||||
"""A paused VM without graphics whose BIOS writes to the serial console once resumed."""
|
||||
dom = conn.defineXML(
|
||||
f"<domain type='kvm'><name>{name}</name><memory unit='MiB'>64</memory><vcpu>1</vcpu>"
|
||||
"<os><type arch='x86_64' machine='q35'>hvm</type><bios useserial='yes'/></os>"
|
||||
"<devices><serial type='pty'><target port='0'/></serial><console type='pty'/></devices></domain>"
|
||||
)
|
||||
dom.createWithFlags(libvirt.VIR_DOMAIN_START_PAUSED)
|
||||
return dom
|
||||
|
||||
|
||||
@unittest.skipIf(socketio is None, "python-socketio is not installed")
|
||||
class LiveSerialConsoleTestCase(TransactionTestCase):
|
||||
def setUp(self):
|
||||
if not livetest.enabled():
|
||||
self.skipTest("Set TEST_LIBVIRT_HOST to run the live libvirt tests")
|
||||
self.user = get_user_model().objects.create_user("serial-live", password="pw")
|
||||
self.hosts = [(os.environ["TEST_LIBVIRT_HOST"], os.environ.get("TEST_LIBVIRT_LOGIN", ""),
|
||||
os.environ.get("TEST_LIBVIRT_PASSWORD", ""), int(os.environ.get("TEST_LIBVIRT_TYPE", 4)))]
|
||||
if os.environ.get("TEST_LIBVIRT_SSH_HOST"):
|
||||
self.hosts.append((os.environ["TEST_LIBVIRT_SSH_HOST"], os.environ.get("TEST_LIBVIRT_SSH_LOGIN", "root"), "", 2))
|
||||
self.server, self.port = start_socketiod(self)
|
||||
|
||||
def token(self, host, login, password, conn_type):
|
||||
compute = Compute.objects.create(name=P + host, hostname=host, login=login, password=password, type=conn_type)
|
||||
conn = connection_manager.get_connection(host, login, password, conn_type)
|
||||
before = livetest.inventory(conn)
|
||||
self.addCleanup(lambda: self.assertEqual(livetest.inventory(conn), before))
|
||||
self.addCleanup(livetest.cleanup, conn)
|
||||
dom = serial_vm(conn, P + "serial")
|
||||
vm = Instance.objects.create(compute=compute, name=dom.name(), uuid=dom.UUIDString())
|
||||
UserInstance.objects.create(user=self.user, instance=vm)
|
||||
return tokens.SIGNER.sign(f"{compute.id}:{vm.uuid}:{self.user.id}"), dom
|
||||
|
||||
def test_the_console_of_a_running_vm_opens_once(self):
|
||||
for args in self.hosts:
|
||||
with self.subTest(host=args[0], type=args[3]):
|
||||
token, dom = self.token(*args)
|
||||
client, received = console(self.port, token)
|
||||
time.sleep(1) # the console is open on the paused VM; its BIOS writes once it runs
|
||||
dom.resume()
|
||||
deadline = time.monotonic() + 30
|
||||
while "SeaBIOS" not in "".join(received["output"]) and time.monotonic() < deadline:
|
||||
time.sleep(0.2)
|
||||
self.assertIn("SeaBIOS", "".join(received["output"]))
|
||||
# one client per VM
|
||||
with self.assertRaises(socketio.exceptions.ConnectionError):
|
||||
console(self.port, token)
|
||||
client.disconnect()
|
||||
deadline = time.monotonic() + 5
|
||||
while children(self.server.pid) and time.monotonic() < deadline:
|
||||
time.sleep(0.05)
|
||||
self.assertEqual(children(self.server.pid), [])
|
||||
dom.destroy()
|
||||
137
console/tests/test_socketiod.py
Normal file
137
console/tests/test_socketiod.py
Normal file
|
|
@ -0,0 +1,137 @@
|
|||
"""socketiod: who may connect, and each session's console kept to itself."""
|
||||
|
||||
import os
|
||||
import time
|
||||
from importlib.machinery import SourceFileLoader
|
||||
from importlib.util import module_from_spec, spec_from_loader
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase, override_settings
|
||||
|
||||
from accounts.models import UserInstance
|
||||
from computes.models import Compute
|
||||
from console import tokens
|
||||
from instances.models import Instance
|
||||
|
||||
SOCKETIOD = Path(__file__).resolve().parent.parent / "socketiod"
|
||||
|
||||
|
||||
def load_socketiod():
|
||||
loader = SourceFileLoader("socketiod_under_test", str(SOCKETIOD))
|
||||
module = module_from_spec(spec_from_loader(loader.name, loader))
|
||||
loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class Base(TestCase):
|
||||
def setUp(self):
|
||||
self.compute = Compute.objects.create(name="c", hostname="192.0.2.1", login="u", password="p", type=1)
|
||||
self.vm = Instance.objects.create(compute=self.compute, name="vm-a", uuid="11111111-0000-0000-0000-000000000001")
|
||||
self.user = get_user_model().objects.create_user("owner", password="pw")
|
||||
UserInstance.objects.create(user=self.user, instance=self.vm)
|
||||
|
||||
def token(self, user=None, vm=None):
|
||||
vm = vm or self.vm
|
||||
return tokens.SIGNER.sign(f"{vm.compute_id}:{vm.uuid}:{(user or self.user).id}")
|
||||
|
||||
|
||||
class TokenTestCase(Base):
|
||||
def test_a_token_opens_its_vm_for_a_user_who_may_open_it(self):
|
||||
self.assertEqual(tokens.authorize(self.token()), self.vm)
|
||||
|
||||
def test_refused_tokens(self):
|
||||
stranger = get_user_model().objects.create_user("stranger", password="pw")
|
||||
inactive = get_user_model().objects.create_user("inactive", password="pw", is_active=False)
|
||||
UserInstance.objects.create(user=inactive, instance=self.vm)
|
||||
cases = {
|
||||
"none": None,
|
||||
"legacy host-uuid": f"{self.compute.id}-{self.vm.uuid}",
|
||||
"tampered": self.token()[:-2] + "xx",
|
||||
"no access": self.token(stranger),
|
||||
"inactive user": self.token(inactive),
|
||||
}
|
||||
for name, token in cases.items():
|
||||
with self.subTest(name), self.assertRaises(PermissionError):
|
||||
tokens.authorize(token)
|
||||
token = self.token()
|
||||
with patch("django.core.signing.time.time", return_value=time.time() + tokens.MAX_AGE + 1):
|
||||
with self.assertRaises(PermissionError):
|
||||
tokens.authorize(token)
|
||||
|
||||
|
||||
class SocketiodTestCase(Base):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.d = load_socketiod()
|
||||
self.started = []
|
||||
self.forks = iter(range(1000, 2000))
|
||||
patches = [
|
||||
patch.object(self.d.pty, "fork", side_effect=self.fork),
|
||||
patch.object(self.d.sio, "start_background_task", side_effect=lambda *a: self.started.append(a)),
|
||||
patch.object(self.d.os, "kill"),
|
||||
# the daemon closes its connection after each check; the test's transaction keeps it
|
||||
patch.object(self.d.db, "close_old_connections"),
|
||||
]
|
||||
for p in patches:
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
|
||||
def fork(self):
|
||||
read, write = os.pipe()
|
||||
self.addCleanup(lambda: [os.close(fd) for fd in (read, write) if _open(fd)])
|
||||
return next(self.forks), read
|
||||
|
||||
@override_settings(SERIAL_CONSOLE_ENABLED=True)
|
||||
def test_a_valid_token_starts_a_console_for_its_session(self):
|
||||
self.assertIsNone(self.d.connect("sid-1", {}, {"token": self.token()}))
|
||||
self.assertEqual(self.d.sessions["sid-1"]["vm"], self.vm.pk)
|
||||
self.assertEqual(self.started[0][0], self.d.forward)
|
||||
|
||||
def test_refused_connections_start_nothing(self):
|
||||
other = Instance.objects.create(compute=self.compute, name="vm-b", uuid="11111111-0000-0000-0000-000000000002")
|
||||
UserInstance.objects.create(user=self.user, instance=other)
|
||||
with override_settings(SERIAL_CONSOLE_ENABLED=False):
|
||||
self.assertIs(self.d.connect("off", {}, {"token": self.token()}), False)
|
||||
with override_settings(SERIAL_CONSOLE_ENABLED=True):
|
||||
self.assertIs(self.d.connect("no-auth", {}, None), False)
|
||||
self.assertIs(self.d.connect("bad", {}, {"token": "x"}), False)
|
||||
self.assertIsNone(self.d.connect("first", {}, {"token": self.token()}))
|
||||
# one session per VM
|
||||
self.assertIs(self.d.connect("second", {}, {"token": self.token()}), False)
|
||||
with patch.object(self.d, "MAX_SESSIONS", 1):
|
||||
self.assertIs(self.d.connect("over", {}, {"token": self.token(vm=other)}), False)
|
||||
self.assertEqual(list(self.d.sessions), ["first"])
|
||||
|
||||
@override_settings(SERIAL_CONSOLE_ENABLED=True)
|
||||
def test_input_and_resize_go_to_the_session_and_are_checked(self):
|
||||
self.d.connect("sid-1", {}, {"token": self.token()})
|
||||
fd = self.d.sessions["sid-1"]["fd"]
|
||||
with patch.object(self.d.os, "write") as write, patch.object(self.d.fcntl, "ioctl") as ioctl:
|
||||
self.d.pty_input("sid-1", {"input": "ls\r"})
|
||||
for bad in (None, "ls", {"input": 7}):
|
||||
self.d.pty_input("sid-1", bad)
|
||||
self.d.pty_input("unknown", {"input": "x"})
|
||||
self.d.resize("sid-1", {"rows": 24, "cols": 80})
|
||||
for bad in ({"rows": 0, "cols": 80}, {"rows": 24, "cols": 501}, {"rows": "24", "cols": 80}, "x"):
|
||||
self.d.resize("sid-1", bad)
|
||||
write.assert_called_once_with(fd, b"ls\r")
|
||||
self.assertEqual(ioctl.call_count, 1)
|
||||
|
||||
@override_settings(SERIAL_CONSOLE_ENABLED=True)
|
||||
def test_a_session_is_stopped_once(self):
|
||||
self.d.connect("sid-1", {}, {"token": self.token()})
|
||||
pid = self.d.sessions["sid-1"]["pid"]
|
||||
self.d.disconnect("sid-1")
|
||||
self.d.stop("sid-1")
|
||||
self.assertNotIn("sid-1", self.d.sessions)
|
||||
self.d.os.kill.assert_called_once_with(pid, self.d.signal.SIGKILL)
|
||||
|
||||
|
||||
def _open(fd):
|
||||
try:
|
||||
os.fstat(fd)
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
|
@ -1,5 +1,7 @@
|
|||
"""socketiod's console process ending on its own: the clients are told and
|
||||
disconnected, the process is reaped and the next connection starts a new one."""
|
||||
"""socketiod as a process: each client gets the output of its own console
|
||||
worker, the worker is reaped, and a client without a valid token is refused.
|
||||
The workers cannot reach libvirt here (the VMs' compute is the local socket),
|
||||
so each ends with "Cannot open the console of <its VM>"."""
|
||||
|
||||
import os
|
||||
import socket
|
||||
|
|
@ -11,6 +13,15 @@ import time
|
|||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.db import connection
|
||||
from django.test import TransactionTestCase
|
||||
|
||||
from accounts.models import UserInstance
|
||||
from computes.models import Compute
|
||||
from console import tokens
|
||||
from instances.models import Instance
|
||||
|
||||
try:
|
||||
import socketio
|
||||
except ImportError: # pragma: no cover
|
||||
|
|
@ -40,79 +51,78 @@ def children(pid):
|
|||
return found
|
||||
|
||||
|
||||
def start_socketiod(test):
|
||||
"""socketiod on a free port, reading test's database; stopped at test's cleanup."""
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
test.addCleanup(tmp.cleanup)
|
||||
db = {k: connection.settings_dict[k] for k in ("ENGINE", "USER", "PASSWORD", "HOST", "PORT", "OPTIONS")}
|
||||
db["NAME"] = str(connection.settings_dict["NAME"])
|
||||
(Path(tmp.name) / "socketiod_test_settings.py").write_text(
|
||||
"from webvirtcloud.settings import * # noqa\n"
|
||||
"SERIAL_CONSOLE_ENABLED = True\n"
|
||||
"DATABASES = {'default': %r}\n" % db
|
||||
)
|
||||
port = free_port()
|
||||
env = dict(os.environ, DJANGO_SETTINGS_MODULE="socketiod_test_settings",
|
||||
PYTHONPATH=os.pathsep.join([tmp.name, str(ROOT)]))
|
||||
server = subprocess.Popen([sys.executable, str(SOCKETIOD), "-H", "127.0.0.1", "-p", str(port)],
|
||||
cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
test.addCleanup(lambda: (server.kill(), server.wait()))
|
||||
end = time.monotonic() + 20
|
||||
while time.monotonic() < end:
|
||||
with socket.socket() as s:
|
||||
if s.connect_ex(("127.0.0.1", port)) == 0:
|
||||
return server, port
|
||||
test.assertIsNone(server.poll(), "socketiod exited at start")
|
||||
time.sleep(0.1)
|
||||
test.fail("socketiod did not listen within 20 s")
|
||||
|
||||
|
||||
def console(port, token):
|
||||
"""A connected client and what it receives until the server disconnects it."""
|
||||
client = socketio.Client(reconnection=False)
|
||||
received = {"output": [], "gone": threading.Event()}
|
||||
client.on("pty_output", lambda data: received["output"].append(data["output"]))
|
||||
client.on("disconnect", lambda *args: received["gone"].set())
|
||||
# the same origin as the server, as the page through nginx
|
||||
url = "http://127.0.0.1:%d" % port
|
||||
client.connect(url, auth={"token": token}, transports=["polling"], headers={"Origin": url})
|
||||
return client, received
|
||||
|
||||
|
||||
@unittest.skipIf(socketio is None, "python-socketio is not installed")
|
||||
@unittest.skipUnless(sys.platform.startswith("linux"), "reads /proc")
|
||||
class SocketiodChildExitTestCase(unittest.TestCase):
|
||||
class SocketiodProcessTestCase(TransactionTestCase):
|
||||
def setUp(self):
|
||||
tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(tmp.cleanup)
|
||||
(Path(tmp.name) / "socketiod_child_settings.py").write_text(
|
||||
"from webvirtcloud.settings import * # noqa\n"
|
||||
"SERIAL_CONSOLE_ENABLED = True\n"
|
||||
"DATABASES = {'default': {'ENGINE': 'django.db.backends.sqlite3', 'NAME': %r}}\n"
|
||||
% str(Path(tmp.name) / "db.sqlite3")
|
||||
)
|
||||
self.port = free_port()
|
||||
env = dict(
|
||||
os.environ,
|
||||
DJANGO_SETTINGS_MODULE="socketiod_child_settings",
|
||||
PYTHONPATH=os.pathsep.join([tmp.name, str(ROOT)]),
|
||||
)
|
||||
self.server = subprocess.Popen(
|
||||
[sys.executable, str(SOCKETIOD), "-H", "127.0.0.1", "-p", str(self.port)],
|
||||
cwd=ROOT,
|
||||
env=env,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
self.addCleanup(self.stop_server)
|
||||
end = time.monotonic() + 20
|
||||
while time.monotonic() < end:
|
||||
with socket.socket() as s:
|
||||
if s.connect_ex(("127.0.0.1", self.port)) == 0:
|
||||
return
|
||||
self.assertIsNone(self.server.poll(), "socketiod exited at start")
|
||||
time.sleep(0.1)
|
||||
self.fail("socketiod did not listen within 20 s")
|
||||
compute = Compute.objects.create(name="local", hostname="localhost", login="", password="", type=4)
|
||||
user = get_user_model().objects.create_user("serial-owner", password="pw")
|
||||
self.tokens = {}
|
||||
for n, name in enumerate(("vm-a", "vm-b"), 1):
|
||||
vm = Instance.objects.create(compute=compute, name=name, uuid=f"22222222-0000-0000-0000-00000000000{n}")
|
||||
UserInstance.objects.create(user=user, instance=vm)
|
||||
self.tokens[name] = tokens.SIGNER.sign(f"{compute.id}:{vm.uuid}:{user.id}")
|
||||
self.server, self.port = start_socketiod(self)
|
||||
|
||||
def stop_server(self):
|
||||
self.server.kill()
|
||||
self.server.wait()
|
||||
def console(self, token):
|
||||
return console(self.port, token)
|
||||
|
||||
def open_console(self):
|
||||
"""Connect with a malformed token, so the console process fails at once;
|
||||
returns the output and whether the server disconnected the client."""
|
||||
client = socketio.Client(reconnection=False)
|
||||
output = []
|
||||
gone = threading.Event()
|
||||
client.on("pty_output", lambda data: output.append(data["output"]))
|
||||
client.on("disconnect", lambda *args: gone.set())
|
||||
client.connect(
|
||||
"http://127.0.0.1:%d" % self.port,
|
||||
headers={"Cookie": "token=no-such-vm"},
|
||||
transports=["polling"],
|
||||
)
|
||||
try:
|
||||
gone.wait(15)
|
||||
finally:
|
||||
if not gone.is_set():
|
||||
client.disconnect()
|
||||
return "".join(output), gone.is_set()
|
||||
|
||||
def assert_no_child_left(self):
|
||||
def test_each_client_gets_its_own_console(self):
|
||||
sessions = {name: self.console(token) for name, token in self.tokens.items()}
|
||||
for name, (client, received) in sessions.items():
|
||||
with self.subTest(name):
|
||||
received["gone"].wait(20)
|
||||
output = "".join(received["output"])
|
||||
self.assertIn("Cannot open the console of %s" % name, output)
|
||||
self.assertNotIn("vm-a" if name == "vm-b" else "vm-b", output)
|
||||
self.assertIn("console closed", output)
|
||||
end = time.monotonic() + 5
|
||||
while children(self.server.pid) and time.monotonic() < end:
|
||||
time.sleep(0.05)
|
||||
self.assertEqual(children(self.server.pid), [])
|
||||
self.assertIsNone(self.server.poll())
|
||||
|
||||
def test_the_client_is_told_and_the_next_connection_starts_again(self):
|
||||
for attempt in (1, 2):
|
||||
with self.subTest(attempt=attempt):
|
||||
output, disconnected = self.open_console()
|
||||
self.assertIn("console closed", output)
|
||||
# no traceback of socketiod's own code (on Python 3.13 eventlet
|
||||
# prints an ignored _after_fork error in every forked child)
|
||||
self.assertNotIn('socketiod", line', output)
|
||||
self.assertTrue(disconnected)
|
||||
self.assert_no_child_left()
|
||||
self.assertIsNone(self.server.poll())
|
||||
def test_a_client_without_a_valid_token_is_refused(self):
|
||||
for token in ("", "22-no-such-vm"):
|
||||
with self.subTest(token=token), self.assertRaises(socketio.exceptions.ConnectionError):
|
||||
self.console(token)
|
||||
self.assertEqual(children(self.server.pid), [])
|
||||
|
|
|
|||
|
|
@ -1,42 +0,0 @@
|
|||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent.parent
|
||||
SOCKETIOD = ROOT / "console" / "socketiod"
|
||||
|
||||
|
||||
class SocketiodDisabledTestCase(unittest.TestCase):
|
||||
"""socketiod must refuse to start unless explicitly enabled."""
|
||||
|
||||
def test_socketiod_exits_when_serial_console_disabled(self):
|
||||
# The settings module lives in a temporary directory, so the test does
|
||||
# not write into the source tree (read-only checkouts, --parallel).
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
(Path(tmp) / "socketiod_disabled_settings.py").write_text(
|
||||
"from webvirtcloud.settings import * # noqa\n"
|
||||
"SERIAL_CONSOLE_ENABLED = False\n"
|
||||
"SOCKETIO_HOST = '127.0.0.1'\n"
|
||||
"SOCKETIO_PORT = 0\n"
|
||||
)
|
||||
env = dict(
|
||||
os.environ,
|
||||
DJANGO_SETTINGS_MODULE="socketiod_disabled_settings",
|
||||
PYTHONPATH=os.pathsep.join([tmp, str(ROOT)]),
|
||||
)
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(SOCKETIOD)],
|
||||
cwd=ROOT,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=20,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.fail("socketiod kept running although SERIAL_CONSOLE_ENABLED=False")
|
||||
self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
|
||||
self.assertIn("SERIAL_CONSOLE_ENABLED", result.stderr)
|
||||
|
|
@ -140,8 +140,6 @@ class ConsoleViewsTestCase(TestCase):
|
|||
@override_settings(SERIAL_CONSOLE_ENABLED=False)
|
||||
@patch("console.views.wvmInstance")
|
||||
def test_console_pty_refused_when_serial_console_disabled(self, mock_wvm):
|
||||
# socketiod is unauthenticated, so the serial console is
|
||||
# disabled unless an admin explicitly enables it.
|
||||
mock_conn = MagicMock()
|
||||
mock_conn.get_console_type.return_value = "pty"
|
||||
mock_wvm.return_value = mock_conn
|
||||
|
|
@ -519,7 +517,7 @@ class NovncdDaemonLogicTestCase(TestCase):
|
|||
"No console token provided in cookie or query parameters"
|
||||
)
|
||||
|
||||
@patch("instances.models.Instance.objects.get")
|
||||
@patch("console.tokens.authorize")
|
||||
@patch("vrtManager.instance.wvmInstance")
|
||||
def test_get_connection_infos_valid_signed_token(self, mock_wvm, mock_inst_get):
|
||||
User = get_user_model()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue