mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 03:36:48 +00:00
7cf12b997b
sortable.js 0.8.0 (abandoned since 2014) sorted only a table's first
body: in the grouped VM list the first host's row moved among its VMs
and the other hosts never sorted, and memory sorted as text ("1024 MB"
before "512 MB"). tablesort 5.7.1 sorts each table body on its own and
keeps a row marked data-sort-method="none" in place, so hosts stay in
name order with their row on top and their VMs sort below it. Memory,
vCPUs and volume sizes sort by their numbers (data-sort). Headers are
reachable by keyboard and carry aria-sort.
base.html loads tablesort for every table marked data-sortable; the
per-page sortable includes and sortable-theme-bootstrap.css go, so these
tables take the theme's table style like the others. The grouped list's
row numbers go (meaningless once sorted). The VM owners and VM logs
tables were marked sortable but never loaded the script; they stay
unsorted.
199 lines
10 KiB
Python
199 lines
10 KiB
Python
"""Every libvirt state has a label, and the lists offer the actions the state
|
|
and the user's permissions allow."""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import PropertyMock, patch
|
|
|
|
from accounts.models import UserInstance
|
|
from appsettings.models import AppSettings
|
|
from computes.models import Compute
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.models import Permission
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from instances.models import Instance
|
|
from instances.states import allowed_actions
|
|
from lxml import html
|
|
|
|
LABELS = {0: "No state", 1: "Active", 2: "Blocked", 3: "Suspended", 4: "Shutting down", 5: "Off", 6: "Crashed", 7: "PM suspended"}
|
|
# actions offered per state, for a superuser
|
|
ACTIONS = {
|
|
0: set(),
|
|
1: {"poweroff", "powercycle", "suspend"},
|
|
2: {"poweroff", "powercycle", "suspend"},
|
|
3: {"resume", "force_off"},
|
|
4: {"force_off"},
|
|
5: {"poweron"},
|
|
6: {"force_off"},
|
|
7: {"force_off"},
|
|
}
|
|
ALL = {"poweron", "poweroff", "powercycle", "force_off", "suspend", "resume"}
|
|
|
|
|
|
class Domain:
|
|
def __init__(self, state):
|
|
self.state = state
|
|
|
|
def info(self):
|
|
return [self.state, 4194304, 2097152, 2, 0]
|
|
|
|
|
|
class StateTests(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.admin = User.objects.create_superuser("st-admin", "st@example.com", "pw")
|
|
self.owner = User.objects.create_user("st-owner", password="pw")
|
|
self.viewer = User.objects.create_user("st-viewer", password="pw")
|
|
self.viewer.user_permissions.add(Permission.objects.get(codename="view_instances"))
|
|
self.compute = Compute.objects.create(name="st", hostname="10.0.0.1", login="u", password="p", type=1)
|
|
self.vms = {}
|
|
for state in LABELS:
|
|
vm = Instance.objects.create(compute=self.compute, name=f"st-vm-{state}", uuid=f"{state:08d}-0000-0000-0000-000000000000")
|
|
UserInstance.objects.create(user=self.owner, instance=vm)
|
|
self.vms[state] = vm
|
|
AppSettings.objects.filter(key="VIEW_INSTANCES_LIST_STYLE").update(value="nongrouped")
|
|
|
|
def get(self, user, url):
|
|
by_uuid = {vm.uuid: state for state, vm in self.vms.items()}
|
|
|
|
def wvm(*args, uuid=None, **kwargs):
|
|
return SimpleNamespace(instance=Domain(by_uuid[uuid]), get_title=lambda: "", get_uuid=lambda: uuid)
|
|
|
|
self.client.force_login(user)
|
|
props = {"status": True, "connection_error": None, "cpu_count": 1, "ram_size": 1, "ram_usage": 0}
|
|
patches = [patch.object(Compute, k, new_callable=PropertyMock, return_value=v) for k, v in props.items()]
|
|
patches += [patch("instances.views.utils.refr"), patch("computes.views.utils.refresh_instance_database"),
|
|
patch("instances.models.wvmInstance", side_effect=wvm)]
|
|
for p in patches:
|
|
p.start()
|
|
try:
|
|
response = self.client.get(url)
|
|
finally:
|
|
for p in reversed(patches):
|
|
p.stop()
|
|
self.assertEqual(response.status_code, 200)
|
|
return response.content.decode()
|
|
|
|
def offered(self, page, vm):
|
|
return {a for a in ALL if reverse(f"instances:{a}", args=[vm.id]) in page}
|
|
|
|
def test_every_state_has_a_label(self):
|
|
page = self.get(self.admin, reverse("instances:index"))
|
|
for state, label in LABELS.items():
|
|
with self.subTest(state=state):
|
|
self.assertIn(f">{label}<", page)
|
|
|
|
def test_superuser_actions_follow_the_state(self):
|
|
for url in (reverse("instances:index"), reverse("instances", args=[self.compute.id])):
|
|
page = self.get(self.admin, url)
|
|
for state, vm in self.vms.items():
|
|
with self.subTest(url=url, state=state):
|
|
self.assertEqual(self.offered(page, vm), ACTIONS[state])
|
|
|
|
def test_forced_power_actions_in_the_lists_ask_first(self):
|
|
# one click in a list row would cut the VM off as a power failure does
|
|
for url in (reverse("instances:index"), reverse("instances", args=[self.compute.id])):
|
|
page = html.fromstring(self.get(self.admin, url))
|
|
for state, vm in self.vms.items():
|
|
for action in {"force_off", "powercycle"} & ACTIONS[state]:
|
|
button = page.xpath(f"//form[@action='{reverse(f'instances:{action}', args=[vm.id])}']//button")[0]
|
|
with self.subTest(url=url, vm=vm.name, action=action):
|
|
self.assertEqual(button.get("onclick"), "return confirm(this.dataset.confirm)")
|
|
self.assertIn(vm.name, button.get("data-confirm"))
|
|
self.assertIn("power cut", button.get("data-confirm"))
|
|
|
|
def test_owner_is_not_offered_suspend_or_resume(self):
|
|
# both views are for superusers only; a VM an administrator suspended
|
|
# stays as it is for its owner, as on the VM page
|
|
page = self.get(self.owner, reverse("instances:index"))
|
|
for state, vm in self.vms.items():
|
|
with self.subTest(state=state):
|
|
expected = set() if state == 3 else ACTIONS[state] - {"suspend", "resume"}
|
|
self.assertEqual(self.offered(page, vm), expected)
|
|
|
|
def test_a_viewer_who_owns_nothing_is_offered_no_actions(self):
|
|
page = self.get(self.viewer, reverse("instances:index"))
|
|
for state, vm in self.vms.items():
|
|
with self.subTest(state=state):
|
|
self.assertIn(vm.name, page)
|
|
self.assertEqual(self.offered(page, vm), set())
|
|
|
|
|
|
def test_the_list_offers_each_state_and_marks_each_row_and_host(self):
|
|
# static/js/instance_filter.js filters by these (test_instance_filter_js)
|
|
AppSettings.objects.filter(key="VIEW_INSTANCES_LIST_STYLE").update(value="grouped")
|
|
page = html.fromstring(self.get(self.admin, reverse("instances:index")))
|
|
options = page.xpath("//select[@id='state-filter']/option/text()")
|
|
self.assertEqual(options, ["All states", *LABELS.values(), "Unknown"])
|
|
marks = page.xpath("//tbody[@data-host='st']//tr//@data-state")
|
|
self.assertEqual(marks, list(LABELS.values()))
|
|
# the host row stays on top of its VMs; memory and vCPUs sort as numbers
|
|
rows = page.xpath("//tbody[@data-host='st']/tr")
|
|
self.assertEqual(rows[0].get("data-sort-method"), "none")
|
|
self.assertEqual([r.xpath("td[@data-sort]/@data-sort") for r in rows[1:2]], [["2", "2097152"]])
|
|
|
|
|
|
class DetailStateTests(TestCase):
|
|
"""The VM page names every state and offers a forced power off where only that helps."""
|
|
|
|
def test_states(self):
|
|
owner = get_user_model().objects.create_user("st-detail-owner", password="pw")
|
|
compute = Compute.objects.create(name="st-d", hostname="10.0.0.2", login="u", password="p", type=1)
|
|
vm = Instance.objects.create(compute=compute, name="st-d-vm", uuid="99999999-0000-0000-0000-00000000d000")
|
|
UserInstance.objects.create(user=owner, instance=vm, is_change=True)
|
|
self.client.force_login(owner)
|
|
badge = LABELS
|
|
force_off = reverse("instances:force_off", args=[vm.id])
|
|
poweroff = reverse("instances:poweroff", args=[vm.id])
|
|
for state, label in badge.items():
|
|
with self.subTest(state=state), patch("instances.models.wvmInstance") as wvm, patch.object(
|
|
Compute, "status", new_callable=PropertyMock, return_value=True
|
|
):
|
|
proxy = wvm.return_value
|
|
proxy.get_memory.return_value = 1024
|
|
proxy.get_cur_memory.return_value = 1024
|
|
proxy.get_max_memory.return_value = 4096
|
|
for name in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices", "get_media_devices", "get_net_devices"):
|
|
getattr(proxy, name).return_value = []
|
|
proxy.get_status.return_value = state
|
|
page = self.client.get(reverse("instances:instance", args=[vm.id])).content.decode()
|
|
self.assertIn(f'">{label}</span>', page.replace("\n", "").replace(" ", ""))
|
|
self.assertEqual(poweroff in page, state in (1, 2))
|
|
self.assertEqual(force_off in page, state in (1, 2, 4, 6, 7))
|
|
|
|
|
|
class AllowedActionsTests(TestCase):
|
|
"""The rules themselves, without templates."""
|
|
|
|
def test_rules(self):
|
|
User = get_user_model()
|
|
admin = User.objects.create_superuser("aa-admin", "aa@example.com", "pw")
|
|
owner = User.objects.create_user("aa-owner", password="pw")
|
|
other = User.objects.create_user("aa-other", password="pw")
|
|
compute = Compute.objects.create(name="aa", hostname="10.0.0.3", login="u", password="p", type=1)
|
|
vm = Instance.objects.create(compute=compute, name="aa-vm", uuid="aaaaaaaa-0000-0000-0000-000000000000")
|
|
UserInstance.objects.create(user=owner, instance=vm)
|
|
for state in LABELS:
|
|
with self.subTest(state=state):
|
|
self.assertEqual(allowed_actions(admin, vm, state) - {"console"}, ACTIONS[state])
|
|
expected = set() if state == 3 else ACTIONS[state] - {"suspend", "resume"}
|
|
self.assertEqual(allowed_actions(owner, vm, state) - {"console"}, expected)
|
|
self.assertEqual(allowed_actions(other, vm, state), set())
|
|
|
|
|
|
class TemplateCloneTests(TestCase):
|
|
"""A shut-off template offers Clone to whoever may clone it, owner or not."""
|
|
|
|
def test_clone_link(self):
|
|
User = get_user_model()
|
|
compute = Compute.objects.create(name="tc", hostname="10.0.0.4", login="u", password="p", type=1)
|
|
vm = Instance.objects.create(compute=compute, name="tc-vm", uuid="cccccccc-0000-0000-0000-000000000000", is_template=True)
|
|
cloner = User.objects.create_user("tc-viewer", password="pw")
|
|
cloner.user_permissions.add(*Permission.objects.filter(codename__in=["view_instances", "clone_instances"]))
|
|
cloner = User.objects.get(pk=cloner.pk) # no cached permissions
|
|
owner = User.objects.create_user("tc-owner", password="pw")
|
|
UserInstance.objects.create(user=owner, instance=vm)
|
|
admin = User.objects.create_superuser("tc-admin", "tc@example.com", "pw")
|
|
self.assertEqual(allowed_actions(cloner, vm, 5), {"clone"})
|
|
self.assertEqual(allowed_actions(admin, vm, 5), {"clone"})
|
|
self.assertEqual(allowed_actions(owner, vm, 5), set()) # no clone permission
|