From d250915e1745a080baf13d2ea8f45f55bd052ccb Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 2 Oct 2026 22:31:25 +0300 Subject: [PATCH] fix(instances): edit_volume bus change and format, full-string validation - edit_volume passed driver_type to attach_disk, which has no such parameter, so every bus change failed after the disk was detached; pass format_type. - A disk without a driver type shows "None" in the edit form; keep the disk's current format instead of rejecting it. - Validate with fullmatch, so a trailing newline is no longer accepted, and use one volume-name pattern for disk views and clone. --- instances/test_disk_isolation.py | 51 ++++++++++++++++++++++++++++++++ instances/views.py | 19 +++++++----- 2 files changed, 62 insertions(+), 8 deletions(-) diff --git a/instances/test_disk_isolation.py b/instances/test_disk_isolation.py index 85e8b553..3108b921 100644 --- a/instances/test_disk_isolation.py +++ b/instances/test_disk_isolation.py @@ -247,3 +247,54 @@ class DiskViewsTenantIsolationTestCase(TestCase): "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_passes_valid_arguments_to_attach_disk(self): + # autospec enforces the real attach_disk signature; a plain MagicMock + # would accept the invalid driver_type argument. + 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 = 5 + res = self._edit_volume_post(vol_bus="sata") + self.assertEqual(res.status_code, 302) + proxy.attach_disk.assert_called_once() + self.assertEqual(proxy.attach_disk.call_args.kwargs["format_type"], "qcow2") + + 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() diff --git a/instances/views.py b/instances/views.py index 25b7893c..e7135cb8 100755 --- a/instances/views.py +++ b/instances/views.py @@ -421,8 +421,8 @@ def invalid_disk_options( "io": (io, lambda v: v in proxy.get_io_modes()), "discard": (discard, lambda v: v in proxy.get_discard_modes()), "detect_zeroes": (zeroes, lambda v: v in proxy.get_detect_zeroes_modes()), - "format": (format, lambda v: bool(DISK_FORMAT_RE.match(v))), - "serial": (serial, lambda v: bool(DISK_SERIAL_RE.match(v))), + "format": (format, lambda v: bool(DISK_FORMAT_RE.fullmatch(v))), + "serial": (serial, lambda v: bool(DISK_SERIAL_RE.fullmatch(v))), } return [ name @@ -823,7 +823,7 @@ def add_new_vol(request, pk): cache = request.POST.get("cache", app_settings.INSTANCE_VOLUME_DEFAULT_CACHE) invalid = invalid_disk_options(instance, bus=bus, cache=cache, format=format) - if not VOLUME_NAME_RE.match(name): + if not VOLUME_NAME_RE.fullmatch(name): invalid.append("name") if invalid: return reject_disk_options(request, invalid) @@ -964,6 +964,10 @@ def edit_volume(request, pk): zeroes = request.POST.get( "vol_detect_zeroes", app_settings.INSTANCE_VOLUME_DEFAULT_DETECT_ZEROES ) + # The form shows "None" (or nothing) for a disk without driver type. + if format in ("", "None"): + current = next((d for d in instance.disks if d["dev"] == target_dev), {}) + format = current.get("format") or "" invalid = invalid_disk_options( instance, @@ -988,7 +992,7 @@ def edit_volume(request, pk): new_target_dev, new_path, target_bus=new_bus, - driver_type=format, + format_type=format, cache_mode=cache, readonly=readonly, shareable=shareable, @@ -1603,7 +1607,6 @@ def del_owner(request, pk): MAC_ADDRESS_RE = re.compile(r"^([0-9A-F]{2})(:?[0-9A-F]{2}){5}$", re.IGNORECASE) -CLONE_DISK_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.+-]*$") CLONE_POST_KEY_RE = re.compile(r"^(clone-net-mac-\d+|disk-[a-z0-9]+|meta-[a-z0-9]+)$") @@ -1633,7 +1636,7 @@ def clone(request, pk): # shows those fields to them. if request.user.is_superuser: for key, value in request.POST.items(): - if CLONE_POST_KEY_RE.match(key): + if CLONE_POST_KEY_RE.fullmatch(key): clone_data[key] = value.strip() if app_settings.CLONE_INSTANCE_AUTO_NAME == "True" and not clone_data["name"]: @@ -1661,12 +1664,12 @@ def clone(request, pk): invalid_macs = [ value for key, value in clone_data.items() - if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.match(value) + if key.startswith("clone-net-mac-") and not MAC_ADDRESS_RE.fullmatch(value) ] invalid_disks = [ value for key, value in clone_data.items() - if key.startswith("disk-") and not CLONE_DISK_NAME_RE.match(value or "") + if key.startswith("disk-") and not VOLUME_NAME_RE.fullmatch(value or "") ] if not request.user.is_superuser and quota_msg: