fix(migrate): a disk of an LVM or disk pool does not count as shared with the destination

Each host's logical volumes are its own: a VM whose disk is /dev/vg/x
migrated, uncopied, onto the destination's volume of the same name and
kept running on other data. Such a volume now counts as missing there.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Kk4h2TcUiaWmU5MUszYSea

vrtManager/tests/fakes.py

@@ -88,8 +88,12 @@ def no_volume():
 
 
 class FakeVolume:
-    def __init__(self, backing):
+    def __init__(self, backing, pool_type="dir"):
         self.backing = backing
+        self.pool_type = pool_type
+
+    def storagePoolLookupByVolume(self):
+        return MagicMock(XMLDesc=lambda flags: f"<pool type='{self.pool_type}'/>")
 
     def XMLDesc(self, flags=0):
         if self.backing is None:
@@ -121,7 +125,7 @@ class FakeHost:
     """The destination's libvirt connection. volumes: the paths its pools
     list; unlisted: files its pools show after a refresh."""
 
-    def __init__(self, existing=(), volumes=(), unlisted=(), backing=None, inactive_pools=()):
+    def __init__(self, existing=(), volumes=(), unlisted=(), backing=None, inactive_pools=(), pool_types=None):
         self.existing = set(existing)
         self.defined = []
         self.new = []
@@ -129,12 +133,13 @@ class FakeHost:
         self.unlisted = set(unlisted)
         self.backing = backing or {}  # path: the path of its backing file
         self.inactive_pools = set(inactive_pools)
+        self.pool_types = pool_types or {}  # path: the type of its volume's pool, dir if not named
         self.refreshes = 0
 
     def storageVolLookupByPath(self, path):
         if path not in self.volumes:
             raise no_volume()
-        return FakeVolume(self.backing.get(path))
+        return FakeVolume(self.backing.get(path), self.pool_types.get(path, "dir"))
 
     def storagePoolLookupByName(self, name):
         return FakePool(self, active=name not in self.inactive_pools)

vrtManager/tests/test_migration.py

@@ -246,6 +246,18 @@ class DestinationDisksTestCase(unittest.TestCase):
         with self.assertRaisesRegex(util.OperationError, "vdb \\(dir disk, cannot be checked\\)"):
             self.migrate(FakeHost(), xml, state=5)
 
+    def test_a_volume_of_a_host_s_own_lvm_or_disk_pool_is_missing(self):
+        # another host's logical volume of the same name is other data
+        for pool_type in ("logical", "disk"):
+            with self.subTest(pool_type=pool_type):
+                host = FakeHost(volumes={"/dev/vg/vm"}, pool_types={"/dev/vg/vm": pool_type})
+                xml = with_disks("<disk type='block' device='disk'><source dev='/dev/vg/vm'/><target dev='vda'/></disk>")
+                with self.assertRaisesRegex(util.OperationError, "vda \\(/dev/vg/vm\\)"):
+                    self.migrate(host, xml)
+        host = FakeHost(volumes={"/dev/sdz"}, pool_types={"/dev/sdz": "iscsi"})
+        xml = with_disks("<disk type='block' device='disk'><source dev='/dev/sdz'/><target dev='vda'/></disk>")
+        self.assertEqual(len(self.migrate(host, xml).instance.calls), 1)
+
     def test_a_volume_of_an_inactive_pool_is_missing(self):
         host = FakeHost(volumes={"v"}, inactive_pools={"p"})
         xml = with_disks("<disk type='volume' device='disk'><source pool='p' volume='v'/><target dev='vdc'/></disk>")