feat(instances): migrate a VM with its disks copied into pools of the destination

A migration needed every disk on storage both hosts share: it was refused
when the destination lacked a disk file, and a file of the same name
there was taken for the VM's own disk.

The Migrate tab now lists the VM's disk files for the chosen destination.
For each disk it offers the destination's pools that keep disk files
(directory, filesystem and NFS pools). A disk on storage both hosts share
can also stay where it is.

A directory counts as shared when both hosts have it as a storage pool
and a marker file created on the source exists on the destination. A
directory the probe cannot tell is never taken as shared.

libvirt copies the chosen disks during the migration: migrate3 with
NON_SHARED_DISK for those targets, and destination and persistent
definitions naming the new paths, where libvirt creates the files.

- A shut-off VM is started paused for the copy and shut off on the
  destination; its CPUs never run. Its UEFI variables are copied first,
  as the destination would start from the firmware's template. If the
  migration fails, the VM is shut off on the source again.
- Refused before anything changes, each disk with its reason:
  - a disk neither shared nor given a pool, also one only the next
    start's definition has;
  - CD-ROM images, read-only or encrypted disks, formats other than raw
    and qcow2, and disks with a backing file;
  - a pool that is not of disk files, or that has a file of that name,
    which libvirt would reuse and overwrite;
  - two disks copied to one file.
- The job claims each copy's name in its pool, and the source volume.
- A failed or cancelled migration deletes the copies libvirt made, and
  the UEFI variables copy, unless a VM on the destination has them.
- The source files stay and are named in the log. With "Delete the
  source files of copied disks" they go, except one another VM or volume
  uses; a failed delete is logged and the migration stands.

The NVRAM sharing check uses the same probe: a shut-off move stops when
it cannot be told, and the source file stays.

The API's migrate call takes disk_pools ({target: pool}) and
delete_source.

vrtManager/connection.py

@@ -880,8 +880,11 @@ class wvmConnect(object):
         if os.path.dirname(path) != NVRAM_DIR:
             return False
         with self._nvram_pool() as pool, destination._nvram_pool() as other:
-            if self._nvram_shared(pool, other):
-                return False
+            try:
+                if self.same_directory(pool, other):
+                    return False
+            except libvirtError:
+                return False  # not known to be the source's own file
             vol = self._find_volume(pool, os.path.basename(path))
             if vol is None:
                 return False
@@ -889,11 +892,11 @@ class wvmConnect(object):
             return True
 
     @staticmethod
-    def _nvram_shared(pool, other):
-        """Whether pools on NVRAM_DIR of two hosts are one directory, or other
-        is not known to be one of its own: creating the same file in both
-        fails then. The create is exclusive on the file server, unlike a
-        directory listing, which an NFS client may cache."""
+    def same_directory(pool, other):
+        """Whether pools of two hosts are one directory: creating the same
+        file in both fails then, as it exists already. The create is
+        exclusive on the file server, unlike a directory listing, which an
+        NFS client may cache. Any other failure raises: it tells nothing."""
         marker = (
             f"<volume><name>webvirtcloud-check-{uuid.uuid4().hex}</name><capacity>0</capacity>"
             "<target><format type='raw'/></target></volume>"
@@ -901,22 +904,32 @@ class wvmConnect(object):
         mine = pool.createXML(marker, 0)
         try:
             other.createXML(marker, 0).delete(0)
-        except libvirtError:
-            return True
+        except libvirtError as error:
+            # the file exists (OPERATION_INVALID), or the pool lists it already
+            if error.get_error_code() in (libvirt.VIR_ERR_OPERATION_INVALID, libvirt.VIR_ERR_STORAGE_VOL_EXIST):
+                return True
+            raise
         finally:
             mine.delete(0)
         return False
 
+    def pool_at(self, directory):
+        """The active pool whose directory is directory, or None. A pool
+        without a target path (rbd, gluster, ...) holds no files."""
+        for pool in self.wvm.listAllStoragePools(libvirt.VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
+            if (util.get_xml_path(pool.XMLDesc(0), "/pool/target/path") or "").rstrip("/") == directory:
+                return pool
+        return None
+
     @contextlib.contextmanager
     def _nvram_pool(self):
         """An active, refreshed pool on NVRAM_DIR: libvirt reads and deletes
         files only as volumes of a pool. A transient one when none exists."""
-        for pool in self.wvm.listAllStoragePools(libvirt.VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
-            # a pool without a target path (rbd, gluster, ...) holds no files
-            if (util.get_xml_path(pool.XMLDesc(0), "/pool/target/path") or "").rstrip("/") == NVRAM_DIR:
-                pool.refresh(0)
-                yield pool
-                return
+        pool = self.pool_at(NVRAM_DIR)
+        if pool is not None:
+            pool.refresh(0)
+            yield pool
+            return
         pool = self.wvm.storagePoolCreateXML(
             f"<pool type='dir'><name>webvirtcloud-nvram-{uuid.uuid4().hex[:8]}</name>"
             f"<target><path>{NVRAM_DIR}</path></target></pool>",

vrtManager/instance.py

@@ -17,6 +17,7 @@ try:
         VIR_DOMAIN_INTERFACE_ADDRESSES_SRC_ARP,
         VIR_DOMAIN_RUNNING,
         VIR_DOMAIN_SHUTOFF,
+        VIR_DOMAIN_START_PAUSED,
         VIR_DOMAIN_XML_SECURE,
         VIR_DOMAIN_XML_INACTIVE,
         VIR_DOMAIN_XML_MIGRATABLE,
@@ -30,7 +31,12 @@ try:
         VIR_MIGRATE_AUTO_CONVERGE,
         VIR_MIGRATE_COMPRESSED,
         VIR_MIGRATE_LIVE,
+        VIR_MIGRATE_NON_SHARED_DISK,
         VIR_MIGRATE_OFFLINE,
+        VIR_MIGRATE_PARAM_DEST_XML,
+        VIR_MIGRATE_PARAM_MIGRATE_DISKS,
+        VIR_MIGRATE_PARAM_PERSIST_XML,
+        VIR_MIGRATE_PARAM_URI,
         VIR_MIGRATE_PERSIST_DEST,
         VIR_MIGRATE_UNDEFINE_SOURCE,
         VIR_MIGRATE_UNSAFE,
@@ -99,7 +105,8 @@ def _rate(before, after, seconds, scale=1):
 
 
 class wvmInstances(wvmConnect):
-    def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None):
+    def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None,
+               pools=None):
         """Migrates the VM of conn (a wvmInstance on the source) here.
 
         The mode follows the VM's state, read once: a running VM migrates
@@ -108,8 +115,10 @@ class wvmInstances(wvmConnect):
         fit the state is refused. unsafe, autoconverge (live only) and
         compress apply to a running VM. uri: the native migration URI
         (tcp://address) on the destination; None lets libvirt use the
-        destination's own hostname. Returns the mode and options used, for
-        the log."""
+        destination's own hostname. pools: {target: pool here} of the disks
+        libvirt copies into those pools during the migration (disk_copies);
+        the other disks must be on storage both hosts share. Returns the mode
+        and options used, for the log, and the paths of the copied disks."""
         state = conn.get_status()
         if state == 5:
             if live:
@@ -146,13 +155,17 @@ class wvmInstances(wvmConnect):
             raise util.OperationError("A VM with snapshots cannot be migrated")
         if state == 5:
             self.refuse_local_state(dom.XMLDesc(0))
-        definitions = {dom.XMLDesc(0)}
+        # the definition the migration carries first: it has the disks a migration copies
+        definitions = [dom.XMLDesc(0)]
         if state != 5:
-            definitions.add(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE))
+            definitions.append(dom.XMLDesc(VIR_DOMAIN_XML_INACTIVE))
         # a host device is of the source host: the destination may have none, or another one
         if any(etree.fromstring(xml).xpath(HOST_DEVICES) for xml in definitions):
             raise util.OperationError("A VM with host devices cannot be migrated; remove them first")
-        missing = self.missing_disks(definitions)
+        copies = self.disk_copies(conn, definitions, pools or {})
+        if copies:
+            mode.append("disks copied: " + ", ".join(f"{copy['target']} to {copy['pool'].name()}" for copy in copies))
+        missing = self.missing_disks(definitions, copied={copy["path"] for copy in copies})
         if missing:
             raise util.OperationError(
                 "The destination host does not have these disks of the VM: "
@@ -167,7 +180,11 @@ class wvmInstances(wvmConnect):
                 "Destination host emulator is different. Cannot be migrated"
             )
 
-        if state == 5:
+        if copies:
+            note = self.migrate_copying(conn, state, flags, uri, copies)
+            if note:
+                mode.append(note)
+        elif state == 5:
             # Workaround for a libvirt bug (RHEL-156800), see move_definition.
             # Once supported hosts all run a fixed libvirt, this branch and
             # move_definition can go: the flags above are those for
@@ -175,7 +192,185 @@ class wvmInstances(wvmConnect):
             self.move_definition(dom, conn)
         else:
             dom.migrate(self.wvm, flags, None, uri, 0)
-        return ", ".join(mode)
+        return ", ".join(mode), [copy["path"] for copy in copies]
+
+    def migrate_copying(self, conn, state, flags, uri, copies):
+        """Migrates the VM of conn while libvirt copies the disks of copies
+        into their pools here (NON_SHARED_DISK), the definitions naming the
+        new paths. libvirt copies the disks of a running or paused VM only: a
+        shut-off VM is started paused for it, its CPUs never run, and is shut
+        off here after it; its UEFI variables are copied first, as this host
+        would start from the firmware's template. Returns a note for the log
+        when the VM could not be shut off here."""
+        dom = conn.instance
+        nvram = None
+        xml_flags = VIR_DOMAIN_XML_SECURE | VIR_DOMAIN_XML_MIGRATABLE
+        params = {
+            VIR_MIGRATE_PARAM_MIGRATE_DISKS: [copy["target"] for copy in copies],
+            VIR_MIGRATE_PARAM_DEST_XML: self.moved_disks(dom.XMLDesc(xml_flags), copies),
+            VIR_MIGRATE_PARAM_PERSIST_XML: self.moved_disks(dom.XMLDesc(xml_flags | VIR_DOMAIN_XML_INACTIVE), copies),
+        }
+        if uri:
+            params[VIR_MIGRATE_PARAM_URI] = uri
+        try:
+            if state == 5:
+                flags &= ~VIR_MIGRATE_OFFLINE
+                nvram = self.copy_nvram(conn)
+                dom.createWithFlags(VIR_DOMAIN_START_PAUSED)
+            dom.migrate3(self.wvm, params, flags | VIR_MIGRATE_NON_SHARED_DISK)
+        except libvirtError as error:
+            if state == 5:
+                with contextlib.suppress(libvirtError):
+                    dom.destroy()  # shut off on the source again
+            self.remove_partial_copies(dom.UUIDString(), copies, error, nvram)
+            raise
+        if state == 5:
+            try:
+                self.wvm.lookupByUUIDString(dom.UUIDString()).destroy()
+            except libvirtError as error:
+                return f"left paused here, shut it off: {error}"
+        return None
+
+    @staticmethod
+    def moved_disks(xml, copies):
+        """xml with each copied disk's source at its path here; a disk is
+        found by its target and its source path before the copy."""
+        tree = etree.fromstring(xml)
+        sources = [(source, copy["dest"]) for copy in copies for source in tree.xpath(
+            "devices/disk[target/@dev=$target]/source[@file=$path]", target=copy["target"], path=copy["path"])]
+        for source, dest in sources:
+            source.set("file", dest)
+        return etree.tostring(tree, encoding="unicode")
+
+    def disk_plan(self, conn, xml):
+        """How the file disks of the VM of conn (on the source) in xml, the
+        definition the migration carries, can reach this host: a list of
+        {target, path, capacity, shared, refusal}. shared: both hosts see the
+        file at its path, as the directory both have as a storage pool is
+        one (same_directory); False when not, or when it cannot be told.
+        refusal: why the disk cannot be copied, or None."""
+        rows, shared = [], {}
+        for disk in ElementTree.fromstring(xml).iter("disk"):
+            source = disk.find("source")
+            if disk.get("type") != "file" or source is None or not source.get("file"):
+                continue
+            path = source.get("file")
+            directory = os.path.dirname(path)
+            if directory not in shared:
+                here, there = self.pool_at(directory), conn.pool_at(directory)
+                try:
+                    shared[directory] = here is not None and there is not None and self.same_directory(there, here)
+                except libvirtError:
+                    shared[directory] = False  # not known to be shared: copied, never assumed
+            try:
+                volume = conn.wvm.storageVolLookupByPath(path)
+            except libvirtError:
+                volume = None
+            rows.append({"target": disk.find("target").get("dev"), "path": path, "shared": shared[directory],
+                         "capacity": volume.info()[1] if volume else None,
+                         "refusal": self.copy_refusal(disk, volume)})
+        return rows
+
+    @staticmethod
+    def copy_refusal(disk, volume):
+        """Why the disk (its volume on the source, None if no pool has it) cannot be copied, or None."""
+        if disk.get("device") != "disk":
+            return "an image is not copied; eject it first"
+        if disk.find("readonly") is not None or disk.find("encryption") is not None \
+                or disk.find("source/encryption") is not None:
+            return "a read-only or encrypted disk is not copied"
+        driver = disk.find("driver")
+        if driver is None or driver.get("type") not in ("raw", "qcow2"):
+            return "only raw and qcow2 disks are copied"
+        if volume is None:
+            return "no storage pool of the source host has it"
+        if ElementTree.fromstring(volume.XMLDesc(0)).findtext("backingStore/path"):
+            return "a disk with a backing file is not copied"
+        return None
+
+    def file_pools(self):
+        """The active pools of this host that keep disks as files, a copy's
+        destination: {name: (pool, directory)}."""
+        pools = {}
+        for pool in self.wvm.listAllStoragePools(VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE):
+            tree = ElementTree.fromstring(pool.XMLDesc(0))
+            if tree.get("type") in ("dir", "fs", "netfs") and tree.findtext("target/path"):
+                pools[pool.name()] = (pool, tree.findtext("target/path").rstrip("/"))
+        return pools
+
+    def disk_copies(self, conn, definitions, pools):
+        """The copies of the disks pools names a pool here for: {target, path,
+        dest, pool}, from the first of definitions (XML, the one the migration
+        carries). Each other file disk (see disk_plan), also of the next
+        start's definition, must be on shared storage. A copy never replaces a file: one
+        of the same name in the pool refuses it (libvirt would write into it).
+        Every problem is named in one refusal."""
+        here = self.file_pools() if pools else {}
+        copies, problems, targets, carried = [], [], set(), set()
+        for row in self.disk_plan(conn, definitions[0]):
+            targets.add(row["target"])
+            carried.add(row["path"])
+            name = pools.get(row["target"])
+            label = f"{row['target']} ({row['path']})"
+            if not name:
+                if not row["shared"]:
+                    problems.append(f"{label}: the destination host does not share it; choose a pool to copy it to")
+                continue
+            if row["refusal"]:
+                problems.append(f"{label}: {row['refusal']}")
+                continue
+            if name not in here:
+                problems.append(f"{label}: the destination host has no pool {name} for disk files")
+                continue
+            pool, directory = here[name]
+            pool.refresh(0)
+            file = os.path.basename(row["path"])
+            if self._find_volume(pool, file) is not None:
+                problems.append(f"{label}: pool {name} of the destination host already has a file {file}")
+                continue
+            if any(copy["dest"] == f"{directory}/{file}" for copy in copies):
+                problems.append(f"{label}: another disk is copied to {directory}/{file}")
+                continue
+            copies.append({"target": row["target"], "path": row["path"], "dest": f"{directory}/{file}",
+                           "pool": pool})
+        problems += [f"{target}: the VM has no disk file to copy" for target in pools.keys() - targets if pools[target]]
+        for xml in definitions[1:]:
+            problems += [f"{row['target']} ({row['path']}) of the next start: the destination host does not share it"
+                         for row in self.disk_plan(conn, xml) if row["path"] not in carried and not row["shared"]]
+        if problems:
+            raise util.OperationError("The disks cannot be migrated: " + "; ".join(problems))
+        return copies
+
+    def remove_partial_copies(self, uuid, copies, error, nvram=None):
+        """Deletes the copies libvirt made here for a migration that failed,
+        and the copy of its NVRAM file, while this host has no VM of uuid (a
+        VM here has them as its disks)."""
+        try:
+            self.wvm.lookupByUUIDString(uuid)
+            return
+        except libvirtError as lookup_error:
+            if lookup_error.get_error_code() != VIR_ERR_NO_DOMAIN:
+                raise util.OperationError(
+                    f"{error}; remove the partial disk copies from the destination host: "
+                    + ", ".join(copy["dest"] for copy in copies)
+                ) from error
+        left = []
+        for copy in copies:
+            copy["pool"].refresh(0)
+            volume = self._find_volume(copy["pool"], os.path.basename(copy["dest"]))
+            try:
+                if volume is not None:
+                    volume.delete(0)
+            except libvirtError:
+                left.append(copy["dest"])
+        try:
+            self.delete_nvram(nvram)
+        except libvirtError:
+            left.append(f"{NVRAM_DIR}/{nvram}")
+        if left:
+            raise util.OperationError(
+                f"{error}; remove the partial disk copies from the destination host: {', '.join(left)}"
+            ) from error
 
     @staticmethod
     def refuse_local_state(xml):
@@ -201,7 +396,7 @@ class wvmInstances(wvmConnect):
             vol = self._find_volume(source, name)
             if vol is None:
                 return None  # never started: libvirt creates it at the first start
-            if self._nvram_shared(source, target):
+            if self.same_directory(source, target):
                 if self._find_volume(target, name) is None:
                     raise util.OperationError(f"The destination host cannot see the NVRAM file {path}")
                 return None
@@ -218,7 +413,7 @@ class wvmInstances(wvmConnect):
             with self._nvram_pool() as pool:
                 pool.storageVolLookupByName(name).delete(0)
 
-    def missing_disks(self, definitions):
+    def missing_disks(self, definitions, copied=()):
         """(target, source) of the disks in the VM definitions (XML) that are
         not volumes of this host's active storage pools: the VM's own files,
         their backing files (as the definition or this host's volume lists
@@ -240,6 +435,8 @@ class wvmInstances(wvmConnect):
                     if kind == "volume":
                         disks[(source.get("pool"), source.get("volume"))] = target
                     elif kind in ("file", "block") and (source.get("file") or source.get("dev")):
+                        if element is disk and source.get("file") in copied:
+                            continue
                         disks[(None, source.get("file") or source.get("dev"))] = target
                     elif kind not in ("file", "block"):
                         disks[(kind, None)] = target

vrtManager/tests/test_disk_copy_migration.py

@@ -0,0 +1,289 @@
+"""Migration with disks copied into pools of the destination: libvirt copies
+the disks a pool is chosen for (NON_SHARED_DISK, only those targets), the
+definitions name the new paths; every other file disk must be on storage
+both hosts share. A shut-off VM is started paused for the copy and shut off
+on the destination. Disks that cannot be copied are refused before anything
+changes; the copies of a failed migration are removed again."""
+
+import unittest
+from unittest import mock
+
+import libvirt
+from django.conf import settings
+
+if not settings.configured:
+    settings.configure(MAC_OUI="52:54:10")
+
+from vrtManager import util
+from vrtManager.connection import wvmConnect
+from vrtManager.instance import wvmInstances
+
+IMAGES = "/var/lib/libvirt/images"
+FAST = "/srv/fast"
+
+
+def error(code, message="error"):
+    err = libvirt.libvirtError(message)
+    err.err = (code, 0, message, 2, None, None, None, 0, 0)
+    return err
+
+
+def disk(path, target="vda", device="disk", fmt="qcow2", extra=""):
+    return (f"<disk type='file' device='{device}'><driver name='qemu' type='{fmt}'/>"
+            f"<source file='{path}'/><target dev='{target}'/>{extra}</disk>")
+
+
+def domain(*disks):
+    return f"<domain><devices>{''.join(disks)}</devices></domain>"
+
+
+class Pool:
+    def __init__(self, name, path, files=(), kind="dir"):
+        self._name, self.path, self.kind = name, path, kind
+        self.files = set(files)
+        self.deleted = []
+
+    def name(self):
+        return self._name
+
+    def XMLDesc(self, flags=0):
+        return f"<pool type='{self.kind}'><name>{self._name}</name><target><path>{self.path}</path></target></pool>"
+
+    def refresh(self, flags=0):
+        pass
+
+    def storageVolLookupByName(self, name):
+        if name not in self.files:
+            raise error(libvirt.VIR_ERR_NO_STORAGE_VOL)
+        volume = mock.Mock()
+        volume.delete.side_effect = lambda flags: self.deleted.append(name)
+        return volume
+
+
+class Source:
+    """The source host's wvmInstance: a VM in state with the given running
+    (and next-start) definitions; its volumes have capacity and a backing
+    file; pools: the directories it has as pools."""
+
+    def __init__(self, state, xml, inactive=None, capacity=1 << 30, backing=None, pools=(IMAGES,)):
+        self.state = state
+        self.instance = mock.Mock()
+        self.instance.XMLDesc.side_effect = lambda flags: inactive if inactive and flags & 2 else xml
+        self.instance.snapshotNum.return_value = 0
+        self.instance.UUIDString.return_value = "u-1"
+        self.wvm = mock.Mock()
+        volume = self.wvm.storageVolLookupByPath.return_value
+        volume.info.return_value = [0, capacity, capacity]
+        backing_xml = f"<backingStore><path>{backing}</path></backingStore>" if backing else ""
+        volume.XMLDesc.return_value = f"<volume>{backing_xml}</volume>"
+        self.pools = pools
+
+    def get_status(self):
+        return self.state
+
+    def get_arch(self):
+        return "x86_64"
+
+    def get_dom_emulator(self):
+        return "/usr/bin/qemu-kvm"
+
+    def pool_at(self, directory):
+        return Pool("src", directory) if directory in self.pools else None
+
+
+def destination(*pools, shared=False, vm_here=False):
+    dest = wvmInstances.__new__(wvmInstances)
+    dest.wvm = mock.Mock()
+    dest.wvm.listAllStoragePools.return_value = list(pools)
+    if not vm_here:
+        dest.wvm.lookupByUUIDString.side_effect = error(libvirt.VIR_ERR_NO_DOMAIN)
+    dest.get_emulator = lambda arch: "/usr/bin/qemu-kvm"
+    if isinstance(shared, Exception):
+        dest.same_directory = mock.Mock(side_effect=shared)
+    else:
+        dest.same_directory = mock.Mock(return_value=shared)
+    dest.missing_disks = mock.Mock(return_value=[])
+    dest.copy_nvram = mock.Mock(return_value=None)
+    dest.delete_nvram = mock.Mock()
+    return dest
+
+
+def migrate(dest, source, pools, **options):
+    return dest.moveto(source, "vm", live=source.state == 1, unsafe=False, offline=source.state == 5,
+                       uri="tcp://192.0.2.9", pools=pools, **options)
+
+
+class DiskCopyTestCase(unittest.TestCase):
+    def test_disks_on_shared_storage_migrate_as_before(self):
+        source = Source(1, domain(disk(f"{IMAGES}/vm.qcow2")))
+        dest = destination(Pool("default", IMAGES), shared=True)
+        self.assertEqual(migrate(dest, source, {}), ("live", []))
+        source.instance.migrate.assert_called_once()
+        source.instance.migrate3.assert_not_called()
+
+    def test_a_disk_the_destination_does_not_share_needs_a_pool(self):
+        for pools, shared in (((IMAGES,), False), ((), True)):  # not shared; no pool to probe on the source
+            with self.subTest(pools=pools):
+                source = Source(1, domain(disk(f"{IMAGES}/vm.qcow2")), pools=pools)
+                with self.assertRaisesRegex(util.OperationError, "vda .*does not share it; choose a pool"):
+                    migrate(destination(Pool("default", IMAGES), shared=shared), source, {})
+                source.instance.migrate.assert_not_called()
+
+    def test_disks_are_copied_into_the_chosen_pools_under_their_new_paths(self):
+        xml = domain(disk(f"{IMAGES}/vm.qcow2"), disk(f"{IMAGES}/data.raw", "vdb", fmt="raw"),
+                     "<disk type='network' device='disk'><source protocol='rbd' name='p/i'/><target dev='vdc'/></disk>")
+        source = Source(1, xml, inactive=xml)
+        dest = destination(Pool("default", IMAGES), Pool("fast", FAST))
+        mode, copied = migrate(dest, source, {"vda": "fast", "vdb": "default"})
+        self.assertEqual(mode, "live, disks copied: vda to fast, vdb to default")
+        self.assertEqual(copied, [f"{IMAGES}/vm.qcow2", f"{IMAGES}/data.raw"])
+        params, flags = source.instance.migrate3.call_args.args[1:]
+        self.assertEqual(params[libvirt.VIR_MIGRATE_PARAM_MIGRATE_DISKS], ["vda", "vdb"])
+        self.assertEqual(params[libvirt.VIR_MIGRATE_PARAM_URI], "tcp://192.0.2.9")
+        for key in (libvirt.VIR_MIGRATE_PARAM_DEST_XML, libvirt.VIR_MIGRATE_PARAM_PERSIST_XML):
+            self.assertIn(f'<source file="{FAST}/vm.qcow2"/>', params[key])
+            self.assertIn(f'<source file="{IMAGES}/data.raw"/>', params[key])
+        self.assertEqual(flags & (libvirt.VIR_MIGRATE_NON_SHARED_DISK | libvirt.VIR_MIGRATE_LIVE),
+                         libvirt.VIR_MIGRATE_NON_SHARED_DISK | libvirt.VIR_MIGRATE_LIVE)
+        self.assertEqual(dest.missing_disks.call_args.kwargs["copied"], {f"{IMAGES}/vm.qcow2", f"{IMAGES}/data.raw"})
+
+    def test_a_shut_off_vm_runs_paused_for_the_copy_and_is_shut_off_here(self):
+        source = Source(5, domain(disk(f"{IMAGES}/vm.qcow2")))
+        dest = destination(Pool("default", IMAGES), vm_here=True)
+        mode, _copied = migrate(dest, source, {"vda": "default"})
+        self.assertEqual(mode, "offline, disks copied: vda to default")
+        dest.copy_nvram.assert_called_once_with(source)  # this host would start from the firmware's template
+        source.instance.createWithFlags.assert_called_once_with(libvirt.VIR_DOMAIN_START_PAUSED)
+        flags = source.instance.migrate3.call_args.args[2]
+        self.assertFalse(flags & (libvirt.VIR_MIGRATE_OFFLINE | libvirt.VIR_MIGRATE_LIVE))
+        dest.wvm.lookupByUUIDString.return_value.destroy.assert_called_once_with()
+
+    def test_a_shut_off_vm_whose_migration_fails_is_shut_off_on_the_source_again(self):
+        source = Source(5, domain(disk(f"{IMAGES}/vm.qcow2")))
+        source.instance.migrate3.side_effect = libvirt.libvirtError("no route")
+        dest = destination(Pool("default", IMAGES))
+        dest.copy_nvram.return_value = "vm_VARS.fd"
+        with self.assertRaisesRegex(libvirt.libvirtError, "no route"):
+            migrate(dest, source, {"vda": "default"})
+        source.instance.destroy.assert_called_once_with()
+        dest.delete_nvram.assert_called_once_with("vm_VARS.fd")
+
+    def test_a_vm_left_paused_here_is_named_but_the_migration_stands(self):
+        source = Source(5, domain(disk(f"{IMAGES}/vm.qcow2")))
+        dest = destination(Pool("default", IMAGES), vm_here=True)
+        dest.wvm.lookupByUUIDString.return_value.destroy.side_effect = libvirt.libvirtError("busy")
+        mode, _copied = migrate(dest, source, {"vda": "default"})
+        self.assertEqual(mode, "offline, disks copied: vda to default, left paused here, shut it off: busy")
+
+    def test_disks_that_cannot_be_copied_are_refused_with_their_reasons(self):
+        cases = (
+            (disk(f"{IMAGES}/x.iso", "vda", device="cdrom", fmt="raw"), {}, "eject it first"),
+            (disk(f"{IMAGES}/vm.qcow2", extra="<readonly/>"), {}, "read-only or encrypted"),
+            (disk(f"{IMAGES}/vm.vmdk", fmt="vmdk"), {}, "only raw and qcow2"),
+            (disk(f"{IMAGES}/vm.qcow2"), {"backing": f"{IMAGES}/base.qcow2"}, "backing file"),
+        )
+        for xml, volume, reason in cases:
+            with self.subTest(reason=reason):
+                source = Source(1, domain(xml), **volume)
+                with self.assertRaisesRegex(util.OperationError, f"cannot be migrated: vda .*{reason}"):
+                    migrate(destination(Pool("default", IMAGES)), source, {"vda": "default"})
+                source.instance.migrate3.assert_not_called()
+
+    def test_a_file_of_the_same_name_in_the_pool_is_never_overwritten(self):
+        source = Source(1, domain(disk(f"{IMAGES}/vm.qcow2")))
+        with self.assertRaises(util.OperationError) as refused:
+            migrate(destination(Pool("fast", FAST, files={"vm.qcow2"})), source, {"vda": "fast"})
+        self.assertEqual(str(refused.exception), "The disks cannot be migrated: "
+                         f"vda ({IMAGES}/vm.qcow2): pool fast of the destination host already has a file vm.qcow2")
+
+    def test_only_a_pool_of_disk_files_and_a_disk_of_the_vm_can_be_chosen(self):
+        source = Source(1, domain(disk(f"{IMAGES}/vm.qcow2")))
+        dest = destination(Pool("vg", "/dev/vg", kind="logical"))
+        with self.assertRaisesRegex(util.OperationError, "has no pool vg for disk files; vdz: the VM has no disk"):
+            migrate(dest, source, {"vda": "vg", "vdz": "vg"})
+
+    def test_each_disk_keeps_its_own_new_path(self):
+        """Disks of one file name in two directories: neither copy takes the other's path."""
+        xml = domain(disk(f"{IMAGES}/vm.qcow2"), disk(f"{FAST}/vm.qcow2", "vdb"))
+        source = Source(1, xml, pools=(IMAGES, FAST))
+        dest = destination(Pool("default", IMAGES), Pool("fast", FAST), Pool("slow", "/srv/slow"))
+        migrate(dest, source, {"vda": "fast", "vdb": "slow"})
+        dest_xml = source.instance.migrate3.call_args.args[1][libvirt.VIR_MIGRATE_PARAM_DEST_XML]
+        self.assertIn(f'<source file="{FAST}/vm.qcow2"/><target dev="vda"/>', dest_xml)
+        self.assertIn('<source file="/srv/slow/vm.qcow2"/><target dev="vdb"/>', dest_xml)
+
+    def test_a_disk_only_the_next_start_has_must_be_shared_too(self):
+        """It is neither copied nor moved: a file of its name on the
+        destination would be another VM's, used at the next start."""
+        source = Source(1, domain(disk(f"{IMAGES}/live.qcow2")),
+                        inactive=domain(disk(f"{IMAGES}/next.qcow2")))
+        with self.assertRaisesRegex(util.OperationError, f"vda \\({IMAGES}/next.qcow2\\) of the next start: "
+                                                         "the destination host does not share it"):
+            migrate(destination(Pool("default", IMAGES)), source, {"vda": "default"})
+        source.instance.migrate3.assert_not_called()
+
+    def test_two_copies_never_go_to_one_file(self):
+        source = Source(1, domain(disk(f"{IMAGES}/vm.qcow2"), disk(f"{FAST}/vm.qcow2", "vdb")), pools=(IMAGES, FAST))
+        with self.assertRaisesRegex(util.OperationError, "vdb .*another disk is copied to /srv/slow/vm.qcow2"):
+            migrate(destination(Pool("slow", "/srv/slow")), source, {"vda": "slow", "vdb": "slow"})
+
+    @staticmethod
+    def fail_after_copying(source, pool, message):
+        """libvirt made the copy here, then the migration failed."""
+        def migrate3(*args):
+            pool.files.add("vm.qcow2")
+            raise libvirt.libvirtError(message)
+        source.instance.migrate3.side_effect = migrate3
+
+    def test_a_failed_migration_removes_the_partial_copies(self):
+        source, pool = Source(1, domain(disk(f"{IMAGES}/vm.qcow2"))), Pool("fast", FAST)
+        self.fail_after_copying(source, pool, "operation aborted: canceled by client")
+        with self.assertRaisesRegex(libvirt.libvirtError, "canceled by client"):
+            migrate(destination(pool), source, {"vda": "fast"})
+        self.assertEqual(pool.deleted, ["vm.qcow2"])
+
+    def test_copies_a_vm_here_uses_are_kept(self):
+        source, pool = Source(1, domain(disk(f"{IMAGES}/vm.qcow2"))), Pool("fast", FAST)
+        self.fail_after_copying(source, pool, "lost reply")
+        with self.assertRaisesRegex(libvirt.libvirtError, "lost reply"):
+            migrate(destination(pool, vm_here=True), source, {"vda": "fast"})
+        self.assertEqual(pool.deleted, [])
+
+
+class DiskPlanTestCase(unittest.TestCase):
+    def test_each_file_disk_with_its_storage_and_reason(self):
+        xml = domain(disk(f"{IMAGES}/vm.qcow2"), disk(f"{IMAGES}/x.iso", "sda", device="cdrom", fmt="raw"),
+                     disk("/srv/local/data.qcow2", "vdb"))
+        dest = destination(Pool("default", IMAGES), Pool("local", "/srv/local"), shared=True)
+        rows = dest.disk_plan(Source(1, ""), xml)
+        self.assertEqual([(r["target"], r["shared"], r["refusal"], r["capacity"]) for r in rows], [
+            ("vda", True, None, 1 << 30),
+            ("sda", True, "an image is not copied; eject it first", 1 << 30),
+            ("vdb", False, None, 1 << 30),  # no pool on the source: cannot be told
+        ])
+        dest.same_directory.assert_called_once()  # one probe per directory
+
+    def test_a_probe_that_fails_means_not_shared(self):
+        dest = destination(Pool("default", IMAGES), shared=error(libvirt.VIR_ERR_SYSTEM_ERROR))
+        rows = dest.disk_plan(Source(1, ""), domain(disk(f"{IMAGES}/vm.qcow2")))
+        self.assertFalse(rows[0]["shared"])
+
+
+class SameDirectoryTestCase(unittest.TestCase):
+    def check(self, other_create):
+        pool, other = mock.Mock(), mock.Mock()
+        other.createXML.side_effect = other_create
+        try:
+            return wvmConnect.same_directory(pool, other)
+        finally:
+            pool.createXML.return_value.delete.assert_called_once()
+
+    def test_a_file_the_other_host_has_already_means_one_directory(self):
+        for code in (libvirt.VIR_ERR_OPERATION_INVALID, libvirt.VIR_ERR_STORAGE_VOL_EXIST):
+            with self.subTest(code=code):
+                self.assertTrue(self.check(error(code)))
+        self.assertFalse(self.check(None))
+
+    def test_any_other_failure_tells_nothing(self):
+        with self.assertRaises(libvirt.libvirtError):
+            self.check(error(libvirt.VIR_ERR_SYSTEM_ERROR))

vrtManager/tests/test_migration.py

@@ -110,6 +110,9 @@ class FakePool:
     def isActive(self):
         return self.active
 
+    def XMLDesc(self, flags=0):
+        return "<pool><target><path>/elsewhere</path></target></pool>"
+
     def refresh(self, flags=0):
         self.host.refreshes += 1
         self.host.volumes |= self.host.unlisted
@@ -291,7 +294,7 @@ class MigrationModeTestCase(unittest.TestCase):
     def migrate(self, state, **request):
         source = FakeSource(state)
         options = {"live": False, "unsafe": False, "offline": False, **request}
-        mode = destination().moveto(source, "vm", **options)
+        mode, _copied = destination().moveto(source, "vm", **options)
         return mode, source.instance.calls[0][0] if source.instance.calls else None
 
     def test_refused_requests(self):
@@ -344,7 +347,14 @@ FILE_DISK = "<disk type='file' device='disk'><source file='/pool/vm.qcow2'/><tar
 
 class DestinationDisksTestCase(unittest.TestCase):
     """Every disk of the VM must be a volume the destination knows; the
-    VM is refused before anything changes otherwise, naming the disks."""
+    VM is refused before anything changes otherwise, naming the disks.
+    (Which disks are copied, and that the others are shared, is
+    test_disk_copy_migration's.)"""
+
+    def setUp(self):
+        patcher = mock.patch.object(wvmInstances, "disk_copies", return_value=[])
+        patcher.start()
+        self.addCleanup(patcher.stop)
 
     def migrate(self, host, xml, state=1, inactive_xml=None):
         source = FakeSource(state, xml=xml)
@@ -450,7 +460,7 @@ class NvramCopyTestCase(unittest.TestCase):
         self.dest_pool = mock.MagicMock()
         self.dest._nvram_pool = lambda: contextlib.nullcontext(self.dest_pool)
         self.dest.wvm.newStream = mock.MagicMock()
-        self.shared = mock.patch.object(wvmInstances, "_nvram_shared", return_value=False)
+        self.shared = mock.patch.object(wvmInstances, "same_directory", return_value=False)
         self.shared.start()
         self.addCleanup(self.shared.stop)
 
@@ -466,11 +476,19 @@ class NvramCopyTestCase(unittest.TestCase):
 
     def test_a_shared_directory_is_not_copied(self):
         self.shared.stop()
-        with mock.patch.object(wvmInstances, "_nvram_shared", return_value=True):
+        with mock.patch.object(wvmInstances, "same_directory", return_value=True):
             self.move()
         self.dest_pool.createXML.assert_not_called()
         self.shared.start()
 
+    def test_a_directory_whose_sharing_cannot_be_told_stops_the_move(self):
+        self.shared.stop()
+        with mock.patch.object(wvmInstances, "same_directory", side_effect=libvirt.libvirtError("no marker")):
+            with self.assertRaisesRegex(libvirt.libvirtError, "no marker"):
+                self.move()
+        self.shared.start()
+        self.assertEqual(self.host.defined, [])
+
     def test_the_copy_is_deleted_when_the_definition_fails(self):
         # the raised error is this one: its message can be another libvirt
         # error the thread recorded (a closed connection of an earlier live test)

vrtManager/tests/test_remove_nvram.py

@@ -94,6 +94,12 @@ class RemoveNvramTestCase(TestCase):
         self.assertFalse(source.vols["vm_VARS.fd"].delete.called)
         source.created[0].delete.assert_called_once_with(0)  # the marker is gone
 
+    def test_a_directory_whose_sharing_cannot_be_told_is_left_alone(self):
+        source = pool(NVRAM_DIR, ["vm_VARS.fd"])
+        destination = connect([pool(NVRAM_DIR, create_error=error(libvirt.VIR_ERR_SYSTEM_ERROR))])
+        self.assertFalse(connect([source]).remove_nvram(PATH, destination))
+        self.assertFalse(source.vols["vm_VARS.fd"].delete.called)
+
     def test_both_markers_are_removed(self):
         source = pool(NVRAM_DIR, ["vm_VARS.fd"])
         other = pool(NVRAM_DIR)