""" 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. """ 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 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"] ).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_external_snapshots"): 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) 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"), "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): 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"} ) 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_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_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") as mock_wvm: self.client.post( reverse("instances:snapshot", args=[self.instance.id]), {"name": "s1", "description": ""} ) return mock_wvm.return_value.create_snapshot.called 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))