mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
b04f656d2b
Super-linter ran from the unpinned :latest image and, because of branches-ignore: [master], never on pull requests to master. Ruff now checks syntax errors and pyflakes rules (E9, F) on every push and PR. Fixes the existing findings: unused imports and variables, an unused duplicate import in instances/tests.py, and the star import in accounts/views.py. settings-dev.py keeps its star import of the base settings.
312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""
|
|
Regression tests for ROADMAP Wave 0 item 1 (S-02, S-03, S-04):
|
|
disk and media mutation views must be superuser-only (matching the UI, where
|
|
the Disk tab is only rendered for superusers), and even for superusers the
|
|
views must not act on volumes or option values taken blindly from the client.
|
|
"""
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
|
|
from accounts.models import UserInstance
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
|
|
DISK_ENDPOINTS = [
|
|
("add_new_vol", []),
|
|
("add_existing_vol", []),
|
|
("edit_volume", []),
|
|
("delete_vol", []),
|
|
("detach_vol", []),
|
|
("add_cdrom", []),
|
|
("detach_cdrom", ["hda"]),
|
|
("mount_iso", []),
|
|
("unmount_iso", []),
|
|
]
|
|
|
|
VM_DISK = {
|
|
"dev": "vda",
|
|
"bus": "virtio",
|
|
"image": "vm-disk.qcow2",
|
|
"storage": "default",
|
|
"path": "/var/lib/libvirt/images/vm-disk.qcow2",
|
|
"format": "qcow2",
|
|
}
|
|
|
|
|
|
class DiskViewsTenantIsolationTestCase(TestCase):
|
|
def setUp(self):
|
|
User = get_user_model()
|
|
self.superuser = User.objects.create_superuser(
|
|
username="disk_super", email="disk_super@example.com", password="password"
|
|
)
|
|
self.change_user = User.objects.create_user(
|
|
username="disk_change_user", password="password"
|
|
)
|
|
self.staff_change_user = User.objects.create_user(
|
|
username="disk_staff_change_user", password="password", is_staff=True
|
|
)
|
|
self.compute = Compute.objects.create(
|
|
name="disk-compute", hostname="127.0.0.1", login="root", password="", type=1
|
|
)
|
|
self.instance = Instance.objects.create(
|
|
compute=self.compute,
|
|
name="disk-test-vm",
|
|
uuid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
)
|
|
for user in (self.change_user, self.staff_change_user):
|
|
UserInstance.objects.create(
|
|
instance=self.instance, user=user, is_change=True, is_delete=False
|
|
)
|
|
self.referer = f"/instances/{self.instance.id}/"
|
|
|
|
def _url(self, name, extra=()):
|
|
return reverse(f"instances:{name}", args=[self.instance.id, *extra])
|
|
|
|
def _post(self, name, data, extra=()):
|
|
return self.client.post(self._url(name, extra), data, HTTP_REFERER=self.referer)
|
|
|
|
def test_owner_with_is_change_cannot_mutate_disks(self):
|
|
for user in (self.change_user, self.staff_change_user):
|
|
self.client.force_login(user)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage, patch("instances.views.wvmCreate") as mock_create:
|
|
for name, extra in DISK_ENDPOINTS:
|
|
res = self._post(
|
|
name,
|
|
{"storage": "default", "name": "victim.qcow2", "dev": "vda"},
|
|
extra,
|
|
)
|
|
self.assertEqual(
|
|
res.status_code, 403, f"{user.username} POST {name} should be 403"
|
|
)
|
|
mock_wvm.assert_not_called()
|
|
mock_storage.assert_not_called()
|
|
mock_create.assert_not_called()
|
|
|
|
def test_delete_vol_deletes_the_volume_attached_to_dev_not_the_posted_one(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
mock_wvm.return_value.get_disk_devices.return_value = [VM_DISK]
|
|
res = self._post(
|
|
"delete_vol",
|
|
{"storage": "other-pool", "name": "victim.qcow2", "dev": "vda"},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_wvm.return_value.detach_disk.assert_called_once_with("vda")
|
|
self.assertEqual(mock_storage.call_args[0][-1], "default")
|
|
mock_storage.return_value.del_volume.assert_called_once_with("vm-disk.qcow2")
|
|
|
|
def test_delete_vol_refuses_dev_not_attached_to_vm(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
mock_wvm.return_value.get_disk_devices.return_value = [VM_DISK]
|
|
res = self._post(
|
|
"delete_vol",
|
|
{"storage": "default", "name": "victim.qcow2", "dev": "vdz"},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_wvm.return_value.detach_disk.assert_not_called()
|
|
mock_storage.return_value.del_volume.assert_not_called()
|
|
|
|
def test_edit_volume_rejects_unknown_cache_mode(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
proxy.get_io_modes.return_value = {"default": "", "native": ""}
|
|
proxy.get_discard_modes.return_value = {"default": "", "unmap": ""}
|
|
proxy.get_detect_zeroes_modes.return_value = {"default": "", "on": ""}
|
|
res = self._post(
|
|
"edit_volume",
|
|
{
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_cache": "none' io='native",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.edit_disk.assert_not_called()
|
|
proxy.attach_disk.assert_not_called()
|
|
proxy.detach_disk.assert_not_called()
|
|
|
|
def test_add_new_vol_rejects_unknown_bus(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
res = self._post(
|
|
"add_new_vol",
|
|
{
|
|
"storage": "default",
|
|
"name": "new-disk",
|
|
"format": "qcow2",
|
|
"size": "10",
|
|
"bus": "virtio'/><disk",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
proxy.attach_disk.assert_not_called()
|
|
|
|
def test_add_existing_vol_rejects_volume_not_in_pool(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmStorage"
|
|
) as mock_storage:
|
|
proxy = mock_wvm.return_value
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
mock_storage.return_value.get_volumes.return_value = ["vm-disk.qcow2"]
|
|
res = self._post(
|
|
"add_existing_vol",
|
|
{
|
|
"selected_storage": "default",
|
|
"vols": "../../../etc/shadow",
|
|
"bus": "virtio",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.attach_disk.assert_not_called()
|
|
|
|
def _mock_disk_options(self, proxy):
|
|
proxy.get_disk_devices.return_value = [VM_DISK]
|
|
proxy.get_media_devices.return_value = []
|
|
proxy.get_disk_bus_types.return_value = ["virtio", "sata", "scsi"]
|
|
proxy.get_cache_modes.return_value = {"default": "", "none": ""}
|
|
proxy.get_io_modes.return_value = {"default": "", "native": ""}
|
|
proxy.get_discard_modes.return_value = {"default": "", "unmap": ""}
|
|
proxy.get_detect_zeroes_modes.return_value = {"default": "", "on": ""}
|
|
|
|
def test_add_new_vol_rejects_unsafe_volume_name(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
self._mock_disk_options(mock_wvm.return_value)
|
|
res = self._post(
|
|
"add_new_vol",
|
|
{
|
|
"storage": "default",
|
|
"name": "x</name><key>/etc/shadow",
|
|
"format": "qcow2",
|
|
"size": "10",
|
|
"bus": "virtio",
|
|
"cache": "default",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
|
|
def test_edit_volume_accepts_valid_options(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
res = self._post(
|
|
"edit_volume",
|
|
{
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_serial": "SER-01",
|
|
"vol_cache": "none",
|
|
"vol_io_mode": "native",
|
|
"vol_discard_mode": "unmap",
|
|
"vol_detect_zeroes": "on",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.edit_disk.assert_called_once_with(
|
|
"vda", VM_DISK["path"], False, False, "virtio", "SER-01", "qcow2",
|
|
"none", "native", "unmap", "on",
|
|
)
|
|
|
|
def _edit_volume_post(self, **extra):
|
|
data = {
|
|
"edit_volume": "1",
|
|
"dev": "vda",
|
|
"vol_path": VM_DISK["path"],
|
|
"vol_bus_old": "virtio",
|
|
"vol_bus": "virtio",
|
|
"vol_format": "qcow2",
|
|
"vol_cache": "none",
|
|
}
|
|
data.update(extra)
|
|
return self._post("edit_volume", data)
|
|
|
|
def test_edit_volume_bus_change_keeps_the_disk_and_applies_options(self):
|
|
from vrtManager.instance import wvmInstance
|
|
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance", autospec=wvmInstance) as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
proxy.get_status.return_value = 1
|
|
res = self._edit_volume_post(vol_bus="sata")
|
|
self.assertEqual(res.status_code, 302)
|
|
proxy.attach_disk.assert_not_called()
|
|
proxy.detach_disk.assert_not_called()
|
|
proxy.edit_disk.assert_called_once()
|
|
self.assertEqual(proxy.edit_disk.call_args.args[0], "vda")
|
|
self.assertEqual(proxy.edit_disk.call_args.args[4], "sata")
|
|
|
|
def test_edit_volume_keeps_current_format_when_field_is_none_or_empty(self):
|
|
self.client.force_login(self.superuser)
|
|
for value in ("None", ""):
|
|
with self.subTest(vol_format=value), patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
self._edit_volume_post(vol_format=value)
|
|
proxy.edit_disk.assert_called_once()
|
|
self.assertEqual(proxy.edit_disk.call_args[0][6], VM_DISK["format"])
|
|
|
|
def test_disk_names_with_trailing_newline_are_rejected(self):
|
|
self.client.force_login(self.superuser)
|
|
with patch("instances.models.wvmInstance") as mock_wvm, patch(
|
|
"instances.views.wvmCreate"
|
|
) as mock_create, patch("instances.views.wvmStorage"):
|
|
self._mock_disk_options(mock_wvm.return_value)
|
|
self._post(
|
|
"add_new_vol",
|
|
{"storage": "default", "name": "disk\n", "format": "qcow2", "size": "1",
|
|
"bus": "virtio", "cache": "default"},
|
|
)
|
|
mock_create.return_value.create_volume.assert_not_called()
|
|
|
|
def test_edit_volume_accepts_a_disk_without_any_driver_format(self):
|
|
self.client.force_login(self.superuser)
|
|
no_format_disk = dict(VM_DISK, format=None)
|
|
with patch("instances.models.wvmInstance") as mock_wvm:
|
|
proxy = mock_wvm.return_value
|
|
self._mock_disk_options(proxy)
|
|
proxy.get_disk_devices.return_value = [no_format_disk]
|
|
self._edit_volume_post(vol_format="None")
|
|
proxy.edit_disk.assert_called_once()
|
|
self.assertEqual(proxy.edit_disk.call_args[0][6], "")
|