mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
3d632f1592
Settings > Users was a bold title, an unlabelled "+" and a table of names with a trash icon. The owner chose this layout from three rendered options: one list of owners, each with what it may do (change settings, delete the VM, change console settings, or power and console only), Edit rights and Remove, an empty state, and a labelled "Add owner". - Remove and Edit rights come back to the VM's Users tab: the return address carried #users unencoded, so it was lost, and Edit rights went to the user's account page. Both accounts views now go to a ?next page of this site. - The add-owner list offers only users who are not owners yet and not administrators, who can use every VM anyway; a superuser owner also took the only place when a VM may have one owner. - The unused del_owner view and URL are removed: the list uses the accounts page, which asks before removing an owner. XML: - An XML edit not saved yet is kept: the page no longer reloads itself when the VM's state changes, and the browser asks before leaving it. A state change lost the edit. - The button that opens the review dialog reads "Review changes"; it did not save, as "Change" suggested. Closing the dialog gives focus back to the editor.
139 lines
6.9 KiB
Python
139 lines
6.9 KiB
Python
"""Failures reach the user: safe redirects, failure statuses, no silent success."""
|
|
|
|
import inspect
|
|
from unittest.mock import patch
|
|
|
|
from computes.models import Compute
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.messages import get_messages
|
|
from django.test import TestCase
|
|
from django.urls import URLPattern, get_resolver, reverse
|
|
from logs.models import Logs
|
|
from vrtManager.util import OperationError
|
|
|
|
from instances.models import Instance
|
|
|
|
EXTERNAL = "https://evil.example/instances/1/"
|
|
|
|
|
|
def back_views():
|
|
"""URL names of the instance views that return with _back(), also through _snapshot_job()."""
|
|
names = []
|
|
for pattern in get_resolver().url_patterns:
|
|
if getattr(pattern, "namespace", None) != "instances":
|
|
continue
|
|
for sub in pattern.url_patterns:
|
|
if isinstance(sub, URLPattern) and any(
|
|
call in inspect.getsource(inspect.unwrap(sub.callback))
|
|
for call in ("_back(request, pk", "_snapshot_job(request, pk")
|
|
):
|
|
names.append(sub.name)
|
|
return names
|
|
|
|
|
|
class ErrorHandlingTestCase(TestCase):
|
|
def setUp(self):
|
|
admin = get_user_model().objects.create_superuser("err_admin", "e@example.com", "pw")
|
|
self.client.force_login(admin)
|
|
self.compute = Compute.objects.create(name="err-compute", hostname="127.0.0.1", login="", password="", type=4)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute, name="err-vm", uuid="11111111-2222-3333-4444-555555555555"
|
|
)
|
|
self.page = f"/instances/{self.instance.id}/"
|
|
|
|
def url(self, name):
|
|
args = [self.instance.id, "sda"] if name == "detach_cdrom" else [self.instance.id]
|
|
return reverse(f"instances:{name}", args=args)
|
|
|
|
# Required fields for the views that 404 without them
|
|
DATA = {
|
|
"add_owner": lambda self: {"user_id": get_user_model().objects.get(username="err_admin").id},
|
|
"add_network": lambda self: {"add-net-network": "net:default"},
|
|
"set_qos": lambda self: {"qos_direction": "inbound", "net-mac-0": "52:54:00:00:00:01"},
|
|
}
|
|
|
|
def test_redirects_without_or_with_a_foreign_referer_go_to_the_vm_page(self):
|
|
names = back_views()
|
|
self.assertGreaterEqual(len(names), 29)
|
|
for name in names:
|
|
for referer in (None, EXTERNAL):
|
|
with self.subTest(view=name, referer=referer), patch("instances.models.wvmInstance"), \
|
|
patch("instances.jobs.wvmInstance"):
|
|
headers = {"HTTP_REFERER": referer} if referer else {}
|
|
data = self.DATA[name](self) if name in self.DATA else {}
|
|
response = self.client.post(self.url(name), data, **headers)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertTrue(response.url.startswith(self.page + "#"), response.url)
|
|
|
|
def test_add_owner_rejects_a_missing_user(self):
|
|
self.assertEqual(self.client.post(self.url("add_owner"), {}).status_code, 404)
|
|
|
|
def test_same_site_referer_is_kept(self):
|
|
with patch("instances.models.wvmInstance"):
|
|
response = self.client.post(self.url("set_autostart"), {}, HTTP_REFERER="http://testserver/instances/?x=1")
|
|
self.assertEqual(response.url, "http://testserver/instances/?x=1#boot_opt")
|
|
|
|
def migrate(self, target_id, up=True):
|
|
with patch("instances.models.wvmInstance"), patch.object(Compute, "status", up):
|
|
return self.client.post(self.url("migrate"), {"compute_id": target_id, "submission_key": "key-migrate01"})
|
|
|
|
def test_migrate_to_a_down_host_is_reported_and_not_logged_as_done(self):
|
|
target = Compute.objects.create(name="down", hostname="10.0.0.9", login="", password="", type=4)
|
|
response = self.migrate(str(target.id), up=False)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("not reachable", " ".join(str(m) for m in get_messages(response.wsgi_request)))
|
|
self.assertFalse(Logs.objects.filter(message__contains="migrated").exists())
|
|
|
|
def test_migrate_to_the_same_host_is_reported(self):
|
|
response = self.migrate(str(self.compute.id))
|
|
self.assertIn("already on", " ".join(str(m) for m in get_messages(response.wsgi_request)))
|
|
|
|
def test_migrate_to_an_invalid_compute_is_404(self):
|
|
self.assertEqual(self.migrate("").status_code, 404)
|
|
self.assertEqual(self.migrate("999").status_code, 404)
|
|
|
|
|
|
class ApiErrorTestCase(TestCase):
|
|
def setUp(self):
|
|
admin = get_user_model().objects.create_superuser("api_admin", "a@example.com", "pw")
|
|
self.client.force_login(admin)
|
|
self.compute = Compute.objects.create(name="api-compute", hostname="127.0.0.1", login="", password="", type=4)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute, name="api-vm", uuid="11111111-2222-3333-4444-666666666666"
|
|
)
|
|
|
|
def test_libvirt_error_is_json_with_a_failure_status(self):
|
|
url = reverse("compute-instance-poweron", kwargs={"compute_pk": self.compute.id, "pk": self.instance.id})
|
|
with patch("instances.models.wvmInstance") as wvm:
|
|
wvm.return_value.start.side_effect = OperationError("cannot start")
|
|
response = self.client.post(url, {})
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertEqual(response.json(), {"detail": "cannot start"})
|
|
|
|
def test_migrate_to_a_down_host_is_not_started(self):
|
|
target = Compute.objects.create(name="api-down", hostname="10.0.0.9", login="", password="", type=4)
|
|
with patch("instances.models.wvmInstance"), patch.object(Compute, "status", False):
|
|
response = self.client.post(
|
|
reverse("instance-migrate-list"), {"instance": self.instance.id, "target_compute": target.id,
|
|
"submission_key": "key-migrate02"}
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("not reachable", response.json()["detail"])
|
|
|
|
def volume(self, method, active=True, data=None):
|
|
kwargs = {"compute_pk": self.compute.id, "storage_pk": "default"}
|
|
with patch("storages.api.viewsets.wvmStorage") as storage:
|
|
storage.return_value.is_active.return_value = active
|
|
if method == "post":
|
|
return self.client.post(reverse("compute-storage-volumes-list", kwargs=kwargs), data or {})
|
|
return self.client.delete(reverse("compute-storage-volumes-detail", kwargs={**kwargs, "pk": "disk.qcow2"}))
|
|
|
|
def test_volume_calls_on_an_inactive_pool_fail(self):
|
|
data = {"name": "v", "size": 1, "type": "qcow2", "meta_prealloc": False}
|
|
self.assertEqual(self.volume("post", active=False, data=data).status_code, 409)
|
|
self.assertEqual(self.volume("delete", active=False).status_code, 409)
|
|
|
|
def test_invalid_volume_data_is_400_with_the_errors(self):
|
|
response = self.volume("post", data={})
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("name", response.json())
|