From 9205e56e3b360dd8463229057a0858cfd762fbdc Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 2 Oct 2026 23:06:42 +0300 Subject: [PATCH] fix(instances): change a disk's bus without rebuilding it as a file disk The edit_volume bus change detached the disk and attached a new one built as a file disk, so block and rbd disks lost their source and auth, and a failed attach left the disk detached. Add wvmInstance.change_disk_bus, which reuses the disk element and only changes (dropping the old bus address), and re-attaches the original disk if attaching fails. edit_disk also rebuilt every disk with ; it now keeps the existing and of block and network disks. --- instances/test_disk_isolation.py | 12 +++++-- instances/views.py | 43 ++++++++-------------- vrtManager/instance.py | 38 ++++++++++++++++++-- vrtManager/test_xml_safety.py | 62 ++++++++++++++++++++++++++++++++ 4 files changed, 122 insertions(+), 33 deletions(-) diff --git a/instances/test_disk_isolation.py b/instances/test_disk_isolation.py index 3108b921..ae03b011 100644 --- a/instances/test_disk_isolation.py +++ b/instances/test_disk_isolation.py @@ -261,7 +261,7 @@ class DiskViewsTenantIsolationTestCase(TestCase): data.update(extra) return self._post("edit_volume", data) - def test_edit_volume_bus_change_passes_valid_arguments_to_attach_disk(self): + def test_edit_volume_bus_change_keeps_the_disk_and_applies_options(self): # autospec enforces the real attach_disk signature; a plain MagicMock # would accept the invalid driver_type argument. from vrtManager.instance import wvmInstance @@ -273,8 +273,14 @@ class DiskViewsTenantIsolationTestCase(TestCase): 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") + # The disk keeps its type and source; only target dev/bus change, then + # the other options are applied to the disk on its new device. + proxy.change_disk_bus.assert_called_once_with("vda", proxy.change_disk_bus.call_args[0][1], "sata") + new_dev = proxy.change_disk_bus.call_args[0][1] + proxy.attach_disk.assert_not_called() + proxy.edit_disk.assert_called_once() + self.assertEqual(proxy.edit_disk.call_args[0][0], new_dev) + self.assertEqual(proxy.edit_disk.call_args[0][4], "sata") def test_edit_volume_keeps_current_format_when_field_is_none_or_empty(self): self.client.force_login(self.superuser) diff --git a/instances/views.py b/instances/views.py index 17289d4c..ab9122c1 100755 --- a/instances/views.py +++ b/instances/views.py @@ -996,34 +996,21 @@ def edit_volume(request, pk): new_target_dev = utils.get_new_disk_dev(instance.media, instance.disks, new_bus) if new_bus != bus: - instance.proxy.detach_disk(target_dev) - instance.proxy.attach_disk( - new_target_dev, - new_path, - target_bus=new_bus, - format_type=format, - cache_mode=cache, - readonly=readonly, - shareable=shareable, - serial=serial, - io_mode=io, - discard_mode=discard, - detect_zeroes_mode=zeroes, - ) - else: - instance.proxy.edit_disk( - target_dev, - new_path, - readonly, - shareable, - new_bus, - serial, - format, - cache, - io, - discard, - zeroes, - ) + instance.proxy.change_disk_bus(target_dev, new_target_dev, new_bus) + target_dev = new_target_dev + instance.proxy.edit_disk( + target_dev, + new_path, + readonly, + shareable, + new_bus, + serial, + format, + cache, + io, + discard, + zeroes, + ) if not instance.proxy.get_status() == 5: messages.success( diff --git a/vrtManager/instance.py b/vrtManager/instance.py index ea0a2320..cf5162a1 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -844,6 +844,32 @@ class wvmInstance(wvmConnect): if self.get_status() == 5: self.instance.detachDeviceFlags(xml_disk, VIR_DOMAIN_AFFECT_CONFIG) + def change_disk_bus(self, target_dev, new_target_dev, new_bus): + """ + Move a disk to another bus in the persistent config. The disk element + is reused as is (type, source, auth, driver), only changes + and the old bus address is dropped. If attaching fails, the original + disk is attached again. + """ + tree = etree.fromstring(self._XMLDesc(0)) + disk_el = tree.xpath("./devices/disk/target[@dev=$dev]", dev=target_dev)[ + 0 + ].getparent() + old_xml = etree.tostring(disk_el).decode() + + disk_el.find("target").set("dev", new_target_dev) + disk_el.find("target").set("bus", new_bus) + for address in disk_el.findall("address"): + disk_el.remove(address) + new_xml = etree.tostring(disk_el).decode() + + self.instance.detachDeviceFlags(old_xml, VIR_DOMAIN_AFFECT_CONFIG) + try: + self.instance.attachDeviceFlags(new_xml, VIR_DOMAIN_AFFECT_CONFIG) + except libvirtError: + self.instance.attachDeviceFlags(old_xml, VIR_DOMAIN_AFFECT_CONFIG) + raise + def edit_disk( self, target_dev, @@ -894,8 +920,16 @@ class wvmInstance(wvmConnect): f"" ) - xml_disk += f""" - """ + if disk_el.get("type") == "file": + xml_disk += f"" + else: + # Block/network disks: keep their source (and auth) untouched; + # the form only edits the path of file disks. + for name in ("auth", "source"): + el = disk_el.find(name) + if el is not None: + xml_disk += etree.tostring(el).decode() + xml_disk += f"" if readonly: xml_disk += """""" if shareable: diff --git a/vrtManager/test_xml_safety.py b/vrtManager/test_xml_safety.py index 78ac7f83..32b1369a 100644 --- a/vrtManager/test_xml_safety.py +++ b/vrtManager/test_xml_safety.py @@ -242,3 +242,65 @@ class TestDiskXmlEscaping(unittest.TestCase): with self.assertRaises(IndexError): inst.detach_disk("x' or '1'='1") inst.instance.detachDeviceFlags.assert_not_called() + + +class TestChangeDiskBus(unittest.TestCase): + RBD_DOMAIN = ( + "" + "" + "" + "" + "" + "" + "
" + "" + ) + + def _instance(self): + inst = wvmInstance.__new__(wvmInstance) + inst.instance = MagicMock() + inst._XMLDesc = MagicMock(return_value=self.RBD_DOMAIN) + return inst + + def test_keeps_disk_source_and_auth_and_only_changes_target(self): + from lxml import etree + + inst = self._instance() + inst.change_disk_bus("vda", "sda", "sata") + + new_xml = inst.instance.attachDeviceFlags.call_args[0][0] + disk = etree.fromstring(new_xml) + self.assertEqual(disk.get("type"), "network") + self.assertEqual(disk.find("source").get("name"), "pool/vm-disk") + self.assertEqual(disk.find("auth/secret").get("uuid"), "aaaa") + self.assertEqual(disk.find("target").get("dev"), "sda") + self.assertEqual(disk.find("target").get("bus"), "sata") + self.assertIsNone(disk.find("address")) + + def test_reattaches_the_old_disk_when_attach_fails(self): + from libvirt import libvirtError + + inst = self._instance() + inst.instance.attachDeviceFlags.side_effect = [libvirtError("boom"), None] + + with self.assertRaises(libvirtError): + inst.change_disk_bus("vda", "sda", "sata") + + detached_xml = inst.instance.detachDeviceFlags.call_args[0][0] + restored_xml = inst.instance.attachDeviceFlags.call_args_list[1][0][0] + self.assertEqual(restored_xml, detached_xml) + + def test_edit_disk_keeps_source_and_auth_of_a_network_disk(self): + from lxml import etree + + inst = self._instance() + inst.edit_disk( + "vda", "ignored-path", False, False, "virtio", "", "raw", + "default", "default", "default", "default", + ) + + disk = etree.fromstring(inst.instance.updateDeviceFlags.call_args[0][0]) + self.assertEqual(disk.get("type"), "network") + self.assertEqual(disk.find("source").get("name"), "pool/vm-disk") + self.assertEqual(disk.find("source/host").get("name"), "mon1") + self.assertEqual(disk.find("auth/secret").get("uuid"), "aaaa")