mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 03:36:48 +00:00
46a611941e
libvirt's backup copies the disks of a running or paused VM from one instant into new volumes while it keeps running; the clone is defined shut off as before. The VM is held for the copy, the copy can be cancelled, and recovery reads the job's records and libvirt.
259 lines
13 KiB
Python
259 lines
13 KiB
Python
"""
|
|
The VM page renders a control only for users the backend lets use it:
|
|
power to superusers and owners; the console, virt-viewer file and VDI URL to
|
|
superusers and owners (can_open_console); root password, SSH key, resize and
|
|
options to superusers and owners with is_change; per-vCPU hotplug, video
|
|
model and guest agent to superusers; clone with clone_instances on a template
|
|
or with is_change; snapshots with snapshot_instances and is_change, on a
|
|
template only for staff. is_staff and the global view_instances permission
|
|
alone are read-only.
|
|
"""
|
|
import shutil
|
|
import subprocess
|
|
import unittest
|
|
from unittest.mock import PropertyMock, patch
|
|
|
|
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 lxml import html as lxml_html
|
|
|
|
from accounts.models import UserInstance
|
|
from appsettings.models import AppSettings
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from operations.models import Operation
|
|
|
|
CONSOLE = {"console", "vv file", "vdi"}
|
|
POWER = {"power"}
|
|
CHANGE = {"root password", "ssh key", "resize", "options"}
|
|
SUPERUSER_ONLY = {"vcpu hotplug", "video model", "guest agent"}
|
|
|
|
|
|
class PageControls:
|
|
def setUp(self):
|
|
AppSettings.objects.filter(
|
|
key__in=["SHOW_ACCESS_ROOT_PASSWORD", "SHOW_ACCESS_SSH_KEYS", "VIEW_INSTANCE_DETAIL_BOTTOM_BAR"]
|
|
).update(value="True")
|
|
self.compute = Compute.objects.create(
|
|
name="ctl-compute", hostname="127.0.0.1:1", login="root", password="", type=1
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute, name="ctl-vm", uuid="dddddddd-eeee-ffff-0000-111111111111"
|
|
)
|
|
|
|
def user(self, name, staff=False, perms=(), owner=None):
|
|
user = get_user_model().objects.create_user(name, password="x", is_staff=staff)
|
|
for codename in perms:
|
|
user.user_permissions.add(Permission.objects.get(codename=codename))
|
|
if owner is not None:
|
|
UserInstance.objects.create(instance=self.instance, user=user, **owner)
|
|
return user
|
|
|
|
def controls(self, user, status=1):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch.object(
|
|
Compute, "status", new_callable=PropertyMock, return_value=True
|
|
):
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_memory.return_value = 1024
|
|
proxy.get_cur_memory.return_value = 1024
|
|
proxy.get_max_memory.return_value = 4096
|
|
proxy.get_vcpu.return_value = 2
|
|
proxy.get_cur_vcpu.return_value = 1
|
|
proxy.get_max_cpus.return_value = [1, 2, 3, 4]
|
|
# a running VM with hotpluggable vCPUs shows the per-vCPU buttons
|
|
proxy.get_vcpus.return_value = {
|
|
0: {"enabled": "yes", "hotpluggable": "no"},
|
|
1: {"enabled": "no", "hotpluggable": "yes"},
|
|
}
|
|
for getter in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices",
|
|
"get_media_devices", "get_net_devices", "get_snapshot",
|
|
"get_snapshot_jobs"):
|
|
getattr(proxy, getter).return_value = []
|
|
proxy.get_console_type.return_value = "vnc"
|
|
proxy.get_status.return_value = status
|
|
response = self.client.get(reverse("instances:instance", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.last_page = response.content.decode()
|
|
doc = lxml_html.fromstring(response.content)
|
|
pk = self.instance.id
|
|
|
|
def form(name):
|
|
return bool(doc.xpath("//form[@action=$a]", a=reverse(f"instances:{name}", args=[pk])))
|
|
|
|
found = {
|
|
"console": bool(doc.xpath("//button[@id='consoleBtnGroup']")),
|
|
"vv file": bool(doc.xpath("//a[@href=$a]", a=reverse("instances:getvvfile", args=[pk]))),
|
|
"vdi": bool(doc.xpath("//*[@id='vdiconsole']")),
|
|
"power": form("poweroff") or form("poweron"),
|
|
"root password": form("rootpasswd"),
|
|
"ssh key": form("add_public_key"),
|
|
# a running VM with hotplug shows per-vCPU buttons instead of the CPU form
|
|
"resize": form("resize_memory"),
|
|
"options": form("change_options"),
|
|
"console settings": form("update_console"),
|
|
# host pages (superuser-only views)
|
|
"host links": bool(doc.xpath("//a[@href=$a]", a=reverse("overview", args=[self.compute.id]))),
|
|
# a link to the destroy confirmation page
|
|
"destroy": bool(doc.xpath("//a[@href=$a]", a=reverse("instances:destroy", args=[pk]))),
|
|
"vcpu hotplug": form("set_vcpu"),
|
|
"video model": form("set_video_model"),
|
|
"guest agent": form("set_guest_agent"),
|
|
"clone": bool(doc.xpath("//*[@id='clone']")),
|
|
"snapshots": bool(doc.xpath("//*[@id='snapshots']")),
|
|
}
|
|
return {name for name, present in found.items() if present}
|
|
|
|
|
|
|
|
class PageControlsTestCase(PageControls, TestCase):
|
|
@unittest.skipUnless(shutil.which("node"), "node is not installed")
|
|
def test_the_page_scripts_parse(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_js", "ctl_js@example.com", "x")
|
|
viewer = self.user("ctl_js_viewer", perms=("view_instances",))
|
|
for user in (superuser, viewer):
|
|
self.controls(user)
|
|
scripts = lxml_html.fromstring(self.last_page).xpath("//script[not(@src)]/text()")
|
|
for number, script in enumerate(scripts):
|
|
with self.subTest(user=user.username, script=number):
|
|
result = subprocess.run(
|
|
["node", "--check", "-"], input=script, capture_output=True, text=True, timeout=30
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_superuser_sees_every_control(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_super", "ctl_super@example.com", "x")
|
|
self.assertEqual(
|
|
self.controls(superuser),
|
|
CONSOLE | POWER | CHANGE | SUPERUSER_ONLY
|
|
| {"clone", "snapshots", "console settings", "destroy", "host links"},
|
|
)
|
|
|
|
def test_the_job_panel_gets_the_jobs_just_submitted_once(self):
|
|
superuser = get_user_model().objects.create_superuser("ctl_jobs", "ctl_jobs@example.com", "x")
|
|
self.client.force_login(superuser)
|
|
session = self.client.session
|
|
other = {"resource": "instance:aaaaaaaa-bbbb-cccc-0000-111111111111", "id": "4343"}
|
|
session["operations_shown"] = [{"resource": f"instance:{self.instance.uuid}", "id": "4242"}, other]
|
|
session.save()
|
|
self.controls(superuser)
|
|
self.assertIn('["4242"]', self.last_page)
|
|
self.assertNotIn('"4343"', self.last_page)
|
|
# a job remembered for another VM waits for that VM's page
|
|
self.assertEqual(self.client.session["operations_shown"], [other])
|
|
self.controls(superuser)
|
|
self.assertNotIn('"4242"', self.last_page)
|
|
|
|
def test_global_viewers_see_nothing_to_act_on(self):
|
|
perms = ("view_instances", "clone_instances", "snapshot_instances")
|
|
for user in (self.user("ctl_viewer", perms=perms), self.user("ctl_staff", staff=True, perms=perms)):
|
|
with self.subTest(user=user.username):
|
|
self.assertEqual(self.controls(user), set())
|
|
|
|
def test_owner_without_change_powers_and_opens_the_console(self):
|
|
owner = self.user("ctl_owner", perms=("clone_instances", "snapshot_instances"), owner={})
|
|
self.assertEqual(self.controls(owner), CONSOLE | POWER)
|
|
|
|
def test_owner_with_change_gets_change_clone_and_snapshots(self):
|
|
changer = self.user("ctl_changer", perms=("clone_instances", "snapshot_instances"),
|
|
owner={"is_change": True})
|
|
self.assertEqual(self.controls(changer), CONSOLE | POWER | CHANGE | {"clone", "snapshots"})
|
|
|
|
def test_a_paused_vm_can_be_destroyed(self):
|
|
# destroy stops a paused VM like a running one
|
|
owner = self.user("paused_deleter", owner={"is_delete": True})
|
|
self.assertIn("destroy", self.controls(owner, status=3))
|
|
|
|
def test_disks_grow_while_running_paused_or_shut_off(self):
|
|
# QEMU grows a running or paused VM's disk itself (blockResize)
|
|
changer = self.user("disk_resizer", owner={"is_change": True})
|
|
for status, enabled in ((1, True), (3, True), (5, True), (7, False)):
|
|
with self.subTest(status=status):
|
|
self.controls(changer, status=status)
|
|
buttons = lxml_html.fromstring(self.last_page).xpath(
|
|
"//form[@action=$a]//button[@type='submit']",
|
|
a=reverse("instances:resize_disk", args=[self.instance.id]),
|
|
)
|
|
self.assertEqual(bool(buttons), enabled)
|
|
|
|
def test_a_running_paused_or_shut_off_vm_can_be_cloned(self):
|
|
changer = self.user("vm_cloner", perms=("clone_instances",), owner={"is_change": True})
|
|
for status, enabled in ((1, True), (3, True), (5, True), (4, False)):
|
|
with self.subTest(status=status):
|
|
self.controls(changer, status=status)
|
|
buttons = lxml_html.fromstring(self.last_page).xpath(
|
|
"//form[@action=$a]//button[@type='submit']",
|
|
a=reverse("instances:clone", args=[self.instance.id]),
|
|
)
|
|
self.assertEqual(bool(buttons), enabled)
|
|
|
|
def test_owner_with_change_and_no_extra_permissions(self):
|
|
changer = self.user("ctl_plain_changer", owner={"is_change": True})
|
|
self.assertEqual(self.controls(changer), CONSOLE | POWER | CHANGE)
|
|
|
|
|
|
class TemplatePageControlsTestCase(PageControls, TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.instance.is_template = True
|
|
self.instance.save()
|
|
|
|
def test_a_template_is_cloned_by_viewing_it(self):
|
|
viewer = self.user("tpl_viewer", perms=("view_instances", "clone_instances"))
|
|
self.assertEqual(self.controls(viewer, status=5), {"clone"})
|
|
|
|
def test_only_staff_owners_change_a_template(self):
|
|
owner = self.user("tpl_plain_changer", owner={"is_change": True})
|
|
staff = self.user("tpl_staff_owner", staff=True, owner={"is_change": True})
|
|
self.assertEqual(self.controls(owner, status=5) & CHANGE, set())
|
|
self.assertEqual(self.controls(staff, status=5) & CHANGE, CHANGE)
|
|
|
|
def test_console_settings_of_a_template_are_for_staff_owners(self):
|
|
flags = {"is_change": True, "is_vnc": True}
|
|
owner = self.user("tpl_vnc_owner", owner=flags)
|
|
staff = self.user("tpl_vnc_staff", staff=True, owner=flags)
|
|
self.assertNotIn("console settings", self.controls(owner, status=5))
|
|
self.assertIn("console settings", self.controls(staff, status=5))
|
|
|
|
def test_only_staff_owners_destroy_a_template(self):
|
|
owner = self.user("tpl_deleter", owner={"is_delete": True})
|
|
staff = self.user("tpl_staff_deleter", staff=True, owner={"is_delete": True})
|
|
self.assertNotIn("destroy", self.controls(owner, status=5))
|
|
self.assertIn("destroy", self.controls(staff, status=5))
|
|
|
|
def test_snapshots_of_a_template_are_for_staff_owners(self):
|
|
perms = ("clone_instances", "snapshot_instances")
|
|
owner = self.user("tpl_changer", perms=perms, owner={"is_change": True})
|
|
staff = self.user("tpl_staff_changer", staff=True, perms=perms, owner={"is_change": True})
|
|
self.assertNotIn("snapshots", self.controls(owner, status=5))
|
|
self.assertIn("snapshots", self.controls(staff, status=5))
|
|
|
|
|
|
class TemplateSnapshotRuleTestCase(PageControls, TestCase):
|
|
"""The backend side of the template rule: only staff owners (and
|
|
superusers) take a snapshot of a template; other owners with is_change
|
|
and snapshot_instances are turned away without a libvirt call."""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.instance.is_template = True
|
|
self.instance.save()
|
|
|
|
def snapshot(self, user):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance"), patch("instances.jobs.wvmInstance") as mock_wvm:
|
|
mock_wvm.return_value.instance.snapshotListNames.return_value = []
|
|
self.client.post(
|
|
reverse("instances:snapshot", args=[self.instance.id]),
|
|
{"name": "s1", "description": "", "submission_key": "key-snap0001"},
|
|
)
|
|
return Operation.objects.filter(kind="snapshot.create").exists()
|
|
|
|
def test_only_staff_owners_snapshot_a_template(self):
|
|
perms = ("snapshot_instances",)
|
|
owner = self.user("snap_changer", perms=perms, owner={"is_change": True})
|
|
staff = self.user("snap_staff_changer", staff=True, perms=perms, owner={"is_change": True})
|
|
self.assertFalse(self.snapshot(owner))
|
|
self.assertTrue(self.snapshot(staff))
|