mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
802e90c79f
When a VM action failed because its host could not be reached (the
connection could not be opened, or it dropped during the call), the
error page showed libvirt's connection error to any user: it names the
host and how it is reached, e.g. "ssh: connect to host 10.x port 22".
The lists and the VM page already keep that to administrators.
- Opening a connection now fails with ConnectionFailed (a subclass of
OperationError). A dropped connection is recognized by its error
domain and code (RPC/remote/SSH layer with a transport code, measured:
a keepalive timeout is VIR_FROM_RPC/VIR_ERR_INTERNAL_ERROR); an
operation the daemon does not support (VIR_ERR_NO_SUPPORT) is not
treated as one.
- ExceptionMiddleware (pages and API) and the views that report the
error themselves (snapshot, revert, disk deletion, clone) show it to
superusers and view_instances users, and to everyone else "The host
of this virtual machine cannot be reached right now." Errors of the
operation itself ("domain is not running") are shown as before.
Checked on the test setup: powering on a VM of an unreachable compute
shows the owner the generic text and the administrator the error. The
tests fail on the previous code.
84 lines
4.3 KiB
Python
84 lines
4.3 KiB
Python
"""An error page for a host that cannot be reached shows the connection error
|
|
(host, address) only to administrators; errors of the VM operation itself
|
|
are shown to everyone, as they explain what to do."""
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import libvirt
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.models import Permission
|
|
from django.contrib.messages import get_messages
|
|
from django.contrib.messages.storage.fallback import FallbackStorage
|
|
from django.contrib.sessions.middleware import SessionMiddleware
|
|
from django.test import RequestFactory, TestCase
|
|
from vrtManager.util import ConnectionFailed
|
|
from webvirtcloud.middleware import ExceptionMiddleware, error_text
|
|
|
|
HOST_DETAIL = "ssh: connect to host 10.0.0.9 port 22: No route to host"
|
|
|
|
|
|
def libvirt_error(domain, text, code=libvirt.VIR_ERR_SYSTEM_ERROR):
|
|
err = libvirt.libvirtError.__new__(libvirt.libvirtError)
|
|
Exception.__init__(err, text)
|
|
err.err = (code, domain, text, 2, None, None, None, 0, 0)
|
|
return err
|
|
|
|
|
|
class ConnectionErrorPageTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.owner = User.objects.create_user("ce-owner", password="pw")
|
|
self.admin = User.objects.create_superuser("ce-admin", "ce@example.com", "pw")
|
|
self.viewer = User.objects.create_user("ce-viewer", password="pw")
|
|
self.viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
|
|
self.viewer = User.objects.get(pk=self.viewer.pk)
|
|
|
|
def handle(self, user, exception, path="/instances/1/poweron/"):
|
|
request = RequestFactory().post(path)
|
|
SessionMiddleware(lambda r: None).process_request(request)
|
|
request._messages = FallbackStorage(request)
|
|
request.user = user
|
|
response = ExceptionMiddleware(MagicMock()).process_exception(request, exception)
|
|
return response, [str(m) for m in get_messages(request)]
|
|
|
|
def test_owner_does_not_see_connection_details(self):
|
|
for error in (
|
|
ConnectionFailed(f"Connection Failed: Cannot recv data: {HOST_DETAIL}"),
|
|
libvirt_error(libvirt.VIR_FROM_RPC, f"Cannot recv data: {HOST_DETAIL}"),
|
|
):
|
|
with self.subTest(error=str(error)):
|
|
response, messages = self.handle(self.owner, error)
|
|
self.assertEqual(response.status_code, 500)
|
|
self.assertNotIn("10.0.0.9", " ".join(messages))
|
|
self.assertTrue(any("cannot be reached" in m for m in messages), messages)
|
|
|
|
def test_administrators_see_connection_details(self):
|
|
for user in (self.admin, self.viewer):
|
|
with self.subTest(user=user.username):
|
|
_, messages = self.handle(user, libvirt_error(libvirt.VIR_FROM_RPC, f"Cannot recv data: {HOST_DETAIL}"))
|
|
self.assertIn("10.0.0.9", " ".join(messages))
|
|
|
|
def test_operation_errors_are_shown_to_everyone(self):
|
|
for error, text in (
|
|
(libvirt_error(libvirt.VIR_FROM_QEMU, "Requested operation is not valid: domain is not running",
|
|
libvirt.VIR_ERR_OPERATION_INVALID), "domain is not running"),
|
|
# a reachable daemon without the procedure answers from the RPC layer
|
|
(libvirt_error(libvirt.VIR_FROM_RPC, "this function is not supported by the connection driver: virDomainFoo",
|
|
libvirt.VIR_ERR_NO_SUPPORT), "not supported"),
|
|
):
|
|
with self.subTest(text=text):
|
|
_, messages = self.handle(self.owner, error)
|
|
self.assertIn(text, " ".join(messages))
|
|
|
|
def test_api_answers_the_same_way(self):
|
|
response, _ = self.handle(self.owner, ConnectionFailed(f"Connection Failed: {HOST_DETAIL}"), path="/api/v1/instances/1/")
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertNotIn(b"10.0.0.9", response.content)
|
|
response, _ = self.handle(self.admin, ConnectionFailed(f"Connection Failed: {HOST_DETAIL}"), path="/api/v1/instances/1/")
|
|
self.assertIn(b"10.0.0.9", response.content)
|
|
|
|
def test_views_that_report_the_error_themselves(self):
|
|
request = RequestFactory().post("/")
|
|
request.user = self.owner
|
|
self.assertNotIn("10.0.0.9", error_text(request, libvirt_error(libvirt.VIR_FROM_RPC, HOST_DETAIL)))
|
|
self.assertEqual(error_text(request, ValueError("bad name")), "bad name") # clone also reports these
|