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 <target> (dropping the old
bus address), and re-attaches the original disk if attaching fails.

edit_disk also rebuilt every disk with <source file=...>; it now keeps the
existing <source> and <auth> of block and network disks.

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)

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(

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 <target> 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"<driver name='{old_driver_name}' type='{format}' {additionals}/>"
             )
 
-        xml_disk += f"""<source file='{source}'/>
-          <target dev='{target_dev}' bus='{target_bus}'/>"""
+        if disk_el.get("type") == "file":
+            xml_disk += f"<source file='{source}'/>"
+        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"<target dev='{target_dev}' bus='{target_bus}'/>"
         if readonly:
             xml_disk += """<readonly/>"""
         if shareable:

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 = (
+        "<domain><devices>"
+        "<disk type='network' device='disk'>"
+        "<driver name='qemu' type='raw'/>"
+        "<auth username='libvirt'><secret type='ceph' uuid='aaaa'/></auth>"
+        "<source protocol='rbd' name='pool/vm-disk'><host name='mon1' port='6789'/></source>"
+        "<target dev='vda' bus='virtio'/>"
+        "<address type='pci' domain='0x0000' bus='0x00' slot='0x05' function='0x0'/>"
+        "</disk></devices></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")