diff --git a/vrtManager/instance.py b/vrtManager/instance.py
index 74310c7f..f4ed1366 100644
--- a/vrtManager/instance.py
+++ b/vrtManager/instance.py
@@ -40,6 +40,12 @@ try:
VIR_ERR_OPERATION_INVALID,
VIR_DOMAIN_SNAPSHOT_LIST_INTERNAL,
VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL,
+ VIR_DOMAIN_BACKUP_BEGIN_REUSE_EXTERNAL,
+ VIR_DOMAIN_JOB_NONE,
+ VIR_DOMAIN_JOB_OPERATION_BACKUP,
+ VIR_DOMAIN_JOB_STATS_COMPLETED,
+ VIR_DOMAIN_JOB_STATS_KEEP_COMPLETED,
+ VIR_ERR_NO_DOMAIN_BACKUP,
libvirtError,
)
from libvirt_qemu import VIR_DOMAIN_QEMU_AGENT_COMMAND_DEFAULT, qemuAgentCommand
@@ -65,6 +71,12 @@ PERSISTENT_XML = VIR_DOMAIN_XML_INACTIVE | VIR_DOMAIN_XML_SECURE
SNAPSHOT_NAME = re.compile(r"[A-Za-z0-9_-][A-Za-z0-9._-]{0,63}")
# the longest window wvmInstance.usage() averages over
USAGE_MAX_WINDOW = 30
+# A running VM is cloned with a push backup that keeps the VM alive when the
+# guest shuts down during the copy: libvirt 11.10.
+LIVE_CLONE_MIN_LIBVIRT = 11_010_000
+# libvirt 11.10's value; bindings built against older libvirt lack the name,
+# and the flag is sent to the host, whose version is checked.
+VIR_DOMAIN_BACKUP_BEGIN_PRESERVE_SHUTDOWN_DOMAIN = 2
def _rate(before, after, seconds, scale=1):
@@ -1894,13 +1906,11 @@ class wvmInstance(wvmConnect):
mac_tuples = [mac[i : i + n] for i in range(0, len(mac), n)]
return ":".join(mac_tuples)
- def clone_instance(self, clone_data, uuid):
- """Copy a shut-off VM, with its disks and UEFI variables (NVRAM).
- Inputs are checked before anything is allocated, and on failure only
- the volumes this call created are removed. Returns the clone's UUID."""
- if self.get_status() != 5:
- # A running guest keeps writing while its disks are copied.
- raise util.OperationError("Shut the VM down before cloning it")
+ def _clone_definition(self, clone_data, uuid):
+ """The VM's persistent definition renamed to the clone, with its UUID,
+ title, description and MACs; returns it and the (source, dev) of each
+ disk to copy. Disks without a source are dropped. Raises before
+ anything is allocated."""
if clone_data["name"] in self.get_instances():
raise ValueError("An instance with the clone name already exists")
@@ -1941,15 +1951,90 @@ class wvmInstance(wvmConnect):
raise ValueError(f"No MAC address for network interface {num}")
net.find("mac").set("address", self.fix_mac(mac))
+ nvram = tree.find("os/nvram")
+ if nvram is not None and nvram.get("type", "file") != "file":
+ raise util.OperationError("Cannot clone a VM whose NVRAM is not a file")
+ return tree, disks
+
+ def _clone_volume_xml(self, name, capacity, vol_format, owner_uid, owner_gid):
+ return f"""
+ {util.xml_escape(name)}
+ {capacity}
+ 0
+
+
+
+ {owner_uid}
+ {owner_gid}
+ 0644
+
+
+ 1.1
+
+
+ """
+
+ def define_clone(self, clone_data, uuid, paths):
+ """Defines the clone with each copied disk's source set from paths
+ ({dev: new path}) and its own copy of the VM's UEFI variables (NVRAM).
+ The NVRAM copy is removed if the definition fails. Returns the
+ clone's UUID."""
+ tree, disks = self._clone_definition(clone_data, uuid)
+ for source, dev in disks:
+ source.set("file" if source.get("file") else "dev" if source.get("dev") else "name", paths[dev])
+
# the UEFI variables: a path as text, or a since libvirt 8.5
nvram = tree.find("os/nvram")
nvram_source = nvram.find("source") if nvram is not None else None
nvram_path = None
if nvram is not None:
- if nvram.get("type", "file") != "file":
- raise util.OperationError("Cannot clone a VM whose NVRAM is not a file")
nvram_path = nvram_source.get("file") if nvram_source is not None else (nvram.text or "").strip()
+ if not nvram_path:
+ return self._defineXML(etree.tostring(tree).decode()).UUIDString()
+ # the clone starts with the VM's UEFI variables: boot entries, enrolled keys
+ if os.path.dirname(nvram_path) == NVRAM_DIR:
+ pools = self._nvram_pool()
+ else: # a volume of a pool; libvirtError when no pool holds it
+ pools = contextlib.nullcontext(self.get_volume_by_path(nvram_path).storagePoolLookupByVolume())
+ with pools as pool:
+ original = pool.storageVolLookupByName(os.path.basename(nvram_path))
+ # same format, owner and mode as the original: the domain names the format,
+ # and qemu may run without libvirt changing the file's owner
+ target = etree.fromstring(original.XMLDesc(0)).find("target")
+ volume = etree.Element("volume")
+ etree.SubElement(volume, "name").text = clone_data["name"] + "_VARS" + os.path.splitext(nvram_path)[1]
+ etree.SubElement(volume, "capacity").text = "0"
+ copy_target = etree.SubElement(volume, "target")
+ for tag in ("format", "permissions"):
+ if target.find(tag) is not None:
+ copy_target.append(target.find(tag))
+ copy = pool.createXMLFrom(etree.tostring(volume).decode(), original, 0)
+ if nvram_source is not None:
+ nvram_source.set("file", copy.path())
+ else:
+ nvram.text = copy.path()
+ try:
+ return self._defineXML(etree.tostring(tree).decode()).UUIDString()
+ except Exception as err:
+ try:
+ copy.delete(0)
+ except libvirtError:
+ raise util.OperationError(
+ f"Clone failed ({err}); remove these copies by hand: {copy.path()}"
+ ) from err
+ raise
+
+ def clone_instance(self, clone_data, uuid):
+ """Copy a shut-off VM, with its disks and UEFI variables (NVRAM).
+ Inputs are checked before anything is allocated, and on failure only
+ the volumes this call created are removed. Returns the clone's UUID."""
+ if self.get_status() != 5:
+ # A running guest keeps writing while its disks are copied.
+ raise util.OperationError("Shut the VM down before cloning it")
+ _, disks = self._clone_definition(clone_data, uuid)
+
+ paths = {} # dev: path of its copy
created = [] # paths of volumes this call created
try:
for source, dev in disks:
@@ -1959,27 +2044,14 @@ class wvmInstance(wvmConnect):
vol = self.get_volume_by_path(source.get("file"))
vol_format = util.get_xml_path(vol.XMLDesc(0), "/volume/target/format/@type")
new_vol = vol.storagePoolLookupByVolume().createXMLFrom(
- f"""
- {target_file}
- 0
- 0
-
-
-
- {clone_data['disk_owner_uid']}
- {clone_data['disk_owner_gid']}
- 0644
-
-
- 1.1
-
-
- """,
+ self._clone_volume_xml(
+ target_file, 0, vol_format, clone_data["disk_owner_uid"], clone_data["disk_owner_gid"]
+ ),
vol,
meta_prealloc and vol_format == "qcow2",
)
created.append(new_vol.path())
- source.set("file", new_vol.path())
+ paths[dev] = new_vol.path()
elif source.get("protocol") == "rbd":
vol = self.get_volume_by_path(source.get("name"))
vol_format = util.get_xml_path(vol.XMLDesc(0), "/volume/target/format/@type")
@@ -1994,7 +2066,7 @@ class wvmInstance(wvmConnect):
meta_prealloc,
)
created.append(new_vol.path())
- source.set("name", f"{os.path.dirname(source.get('name'))}/{target_file}")
+ paths[dev] = f"{os.path.dirname(source.get('name'))}/{target_file}"
else:
vol = self.get_volume_by_path(source.get("dev"))
pool_name = util.get_xml_path(vol.storagePoolLookupByVolume().XMLDesc(0), "/pool/name")
@@ -2002,41 +2074,9 @@ class wvmInstance(wvmConnect):
name = storage.clone_volume(vol.name(), target_file)
new_path = storage.get_volume(name).path()
created.append(new_path)
- source.set("dev", new_path)
+ paths[dev] = new_path
- if not nvram_path:
- clone = self._defineXML(etree.tostring(tree).decode())
- else:
- # the clone starts with the VM's UEFI variables: boot entries, enrolled keys
- if os.path.dirname(nvram_path) == NVRAM_DIR:
- pools = self._nvram_pool()
- else: # a volume of a pool; libvirtError when no pool holds it
- pools = contextlib.nullcontext(self.get_volume_by_path(nvram_path).storagePoolLookupByVolume())
- with pools as pool:
- original = pool.storageVolLookupByName(os.path.basename(nvram_path))
- # same format, owner and mode as the original: the domain names the format,
- # and qemu may run without libvirt changing the file's owner
- target = etree.fromstring(original.XMLDesc(0)).find("target")
- volume = etree.Element("volume")
- etree.SubElement(volume, "name").text = clone_data["name"] + "_VARS" + os.path.splitext(nvram_path)[1]
- etree.SubElement(volume, "capacity").text = "0"
- copy_target = etree.SubElement(volume, "target")
- for tag in ("format", "permissions"):
- if target.find(tag) is not None:
- copy_target.append(target.find(tag))
- copy = pool.createXMLFrom(etree.tostring(volume).decode(), original, 0)
- if nvram_source is not None:
- nvram_source.set("file", copy.path())
- else:
- nvram.text = copy.path()
- try:
- clone = self._defineXML(etree.tostring(tree).decode())
- except Exception:
- try:
- copy.delete(0)
- except libvirtError:
- created.append(copy.path())
- raise
+ return self.define_clone(clone_data, uuid, paths)
except Exception as err:
leftovers = []
for path in reversed(created):
@@ -2048,7 +2088,119 @@ class wvmInstance(wvmConnect):
raise util.OperationError(f"Clone failed ({err}); remove these copies by hand: {', '.join(leftovers)}") from err
raise
- return clone.UUIDString()
+ def live_clone_targets(self, clone_data, free_percent, free_min_gib):
+ """Refuses a live clone that cannot work, before anything is
+ allocated; returns the new volume of each copied disk, in disk order:
+ in the source volume's pool, at its virtual capacity and format. Each
+ pool keeps the copies plus free_percent of them, at least
+ free_min_gib GiB, free (the snapshots' rule); checked only, nothing
+ is reserved."""
+ if self.wvm.getLibVersion() < LIVE_CLONE_MIN_LIBVIRT:
+ raise util.OperationError("Live clone needs libvirt 11.10 or later; shut the VM down to clone it")
+ tree, disks = self._clone_definition(clone_data, "")
+ if tree.find("devices/tpm") is not None:
+ raise util.OperationError("A VM with a TPM is cloned only shut down")
+ for source, dev in disks:
+ if not source.get("file"):
+ raise util.OperationError(f"Disk {dev} is not a local file; shut the VM down to clone it")
+ if source.getparent().find("shareable") is not None:
+ raise util.OperationError(f"Disk {dev} is shareable; shut the VM down to clone it")
+ if self.instance.jobInfo()[0] != VIR_DOMAIN_JOB_NONE:
+ raise util.OperationError("Another job is running on the VM; clone it when that ends")
+
+ targets, pools, copied = [], {}, {}
+ for source, dev in disks:
+ vol = self.get_volume_by_path(source.get("file"))
+ pool = vol.storagePoolLookupByVolume()
+ name = clone_data["disk-" + dev]
+ try:
+ pool.storageVolLookupByName(name)
+ except libvirtError as err:
+ if err.get_error_code() != VIR_ERR_NO_STORAGE_VOL:
+ raise
+ else:
+ raise util.OperationError(f"Volume {name} already exists in storage pool {pool.name()}")
+ capacity = vol.info()[1]
+ pools[pool.name()] = pool
+ copied[pool.name()] = copied.get(pool.name(), 0) + capacity
+ targets.append(
+ {
+ "dev": dev,
+ "pool": pool.name(),
+ "name": name,
+ "path": os.path.join(util.get_xml_path(pool.XMLDesc(0), "/pool/target/path"), name),
+ "capacity": capacity,
+ "format": util.get_xml_path(vol.XMLDesc(0), "/volume/target/format/@type"),
+ }
+ )
+ for name, size in copied.items():
+ needed = size + max(size * free_percent / 100, free_min_gib * 1024**3)
+ free = pools[name].info()[3]
+ if free < needed:
+ raise util.OperationError(
+ f"Storage pool {name} has {free / 1024**3:.1f} GiB free; the clone needs {needed / 1024**3:.1f} GiB "
+ f"(the copied disks, plus {free_percent:g}% of them, at least {free_min_gib:g} GiB)"
+ )
+ return targets
+
+ def create_clone_target(self, target, owner_uid, owner_gid):
+ """An empty volume for a copied disk; libvirt refuses one that exists.
+ Returns its path."""
+ pool = self.wvm.storagePoolLookupByName(target["pool"])
+ xml = self._clone_volume_xml(target["name"], target["capacity"], target["format"], owner_uid, owner_gid)
+ return pool.createXML(xml, 0).path()
+
+ def begin_clone_backup(self, targets):
+ """Starts libvirt's push backup of the targets' disks into their
+ volumes. The VM keeps running when the guest shuts down meanwhile."""
+ disks = "".join(
+ f""
+ f""
+ for t in targets
+ )
+ self.instance.backupBegin(
+ f"{disks}",
+ None,
+ VIR_DOMAIN_BACKUP_BEGIN_REUSE_EXTERNAL | VIR_DOMAIN_BACKUP_BEGIN_PRESERVE_SHUTDOWN_DOMAIN,
+ )
+
+ def backup_targets(self):
+ """The target paths of the running backup; None when none runs."""
+ try:
+ xml = self.instance.backupGetXMLDesc(0)
+ except libvirtError as err:
+ if err.get_error_code() == VIR_ERR_NO_DOMAIN_BACKUP:
+ return None
+ raise
+ return set(etree.fromstring(xml).xpath("disks/disk/target/@file"))
+
+ def backup_progress(self):
+ """(bytes copied, bytes to copy) of the running backup; None when
+ none runs."""
+ stats = self.instance.jobStats(0)
+ if (
+ stats.get("type", VIR_DOMAIN_JOB_NONE) == VIR_DOMAIN_JOB_NONE
+ or stats.get("operation") != VIR_DOMAIN_JOB_OPERATION_BACKUP
+ ):
+ return None
+ return stats.get("disk_processed", 0), stats.get("disk_total", 0)
+
+ def backup_result(self):
+ """(success, libvirt's error) of the completed backup, kept for later
+ reads; None when libvirt has no completed backup."""
+ stats = self.instance.jobStats(VIR_DOMAIN_JOB_STATS_COMPLETED | VIR_DOMAIN_JOB_STATS_KEEP_COMPLETED)
+ if stats.get("operation") != VIR_DOMAIN_JOB_OPERATION_BACKUP or "success" not in stats:
+ return None
+ return bool(stats["success"]), stats.get("errmsg") or ""
+
+ def abort_backup(self):
+ self.instance.abortJob()
+
+ def fs_freeze(self):
+ self.instance.fsFreeze(None, 0)
+
+ def fs_thaw(self):
+ self.instance.fsThaw(None, 0)
def add_network(
self, mac_address, source, source_type="net", model="virtio", nwfilter=None
diff --git a/vrtManager/tests/test_live_clone.py b/vrtManager/tests/test_live_clone.py
new file mode 100644
index 00000000..c3e0f1e8
--- /dev/null
+++ b/vrtManager/tests/test_live_clone.py
@@ -0,0 +1,263 @@
+"""wvmInstance's live clone primitives: the refusals, the push backup of the
+copied disks into new volumes, and the clone definition both paths share."""
+
+import unittest
+from unittest.mock import MagicMock
+
+import libvirt
+from lxml import etree
+
+from vrtManager import util
+from vrtManager.instance import LIVE_CLONE_MIN_LIBVIRT, VIR_DOMAIN_BACKUP_BEGIN_PRESERVE_SHUTDOWN_DOMAIN, wvmInstance
+
+GIB = 1024**3
+UUID = "abababab-0000-1111-2222-333333333333"
+FORM = {
+ "name": "copy",
+ "disk-vda": "copy-vda.qcow2",
+ "clone-net-mac-0": "52:54:00:00:00:02",
+ "disk_owner_uid": 107,
+ "disk_owner_gid": 107,
+}
+
+
+def error(code):
+ err = libvirt.libvirtError("x")
+ err.err = (code,)
+ return err
+
+
+def domain(disk="", extra=""):
+ return (
+ f"srcu{disk}"
+ ""
+ f"{extra}"
+ )
+
+
+def proxy(xml=None, free=100 * GIB, capacity=10 * GIB, existing=False):
+ p = wvmInstance.__new__(wvmInstance) # no connection: only these calls
+ p.wvm = MagicMock()
+ p.wvm.getLibVersion.return_value = LIVE_CLONE_MIN_LIBVIRT
+ p.instance = MagicMock()
+ p.instance.jobInfo.return_value = [libvirt.VIR_DOMAIN_JOB_NONE] + [0] * 11
+ p.get_instances = lambda: []
+ p._XMLDesc = lambda flags: xml or domain()
+ p.pool = MagicMock()
+ p.pool.name.return_value = "default"
+ p.pool.XMLDesc.return_value = "default/pool"
+ p.pool.info.return_value = [1, 200 * GIB, 100 * GIB, free]
+ if not existing:
+ p.pool.storageVolLookupByName.side_effect = error(libvirt.VIR_ERR_NO_STORAGE_VOL)
+ p.vol = MagicMock()
+ p.vol.info.return_value = [0, capacity, 1 * GIB]
+ p.vol.XMLDesc.return_value = ""
+ p.vol.storagePoolLookupByVolume.return_value = p.pool
+ p.get_volume_by_path = MagicMock(return_value=p.vol)
+ return p
+
+
+class LiveCloneTargetsTestCase(unittest.TestCase):
+ def refused(self, p, text):
+ with self.assertRaisesRegex(util.OperationError, text):
+ p.live_clone_targets(FORM, 10, 5)
+
+ def test_targets_of_the_copied_disks(self):
+ p = proxy()
+ self.assertEqual(
+ p.live_clone_targets(FORM, 10, 5),
+ [
+ {
+ "dev": "vda",
+ "pool": "default",
+ "name": "copy-vda.qcow2",
+ "path": "/pool/copy-vda.qcow2",
+ "capacity": 10 * GIB,
+ "format": "qcow2",
+ }
+ ],
+ )
+ p.pool.storageVolLookupByName.assert_called_once_with("copy-vda.qcow2")
+
+ def test_an_older_libvirt_is_refused(self):
+ p = proxy()
+ p.wvm.getLibVersion.return_value = 11_009_000
+ self.refused(p, "^Live clone needs libvirt 11.10 or later; shut the VM down to clone it$")
+
+ def test_a_block_disk_is_refused(self):
+ p = proxy(domain(""))
+ self.refused(p, "vda is not a local file")
+
+ def test_a_shareable_disk_is_refused(self):
+ p = proxy(
+ domain("")
+ )
+ self.refused(p, "vda is shareable")
+
+ def test_a_tpm_is_refused(self):
+ self.refused(proxy(domain(extra="")), "TPM")
+
+ def test_a_running_libvirt_job_is_refused(self):
+ p = proxy()
+ p.instance.jobInfo.return_value = [libvirt.VIR_DOMAIN_JOB_UNBOUNDED] + [0] * 11
+ self.refused(p, "Another job is running")
+
+ def test_an_existing_target_volume_is_refused(self):
+ self.refused(proxy(existing=True), "copy-vda.qcow2 already exists in storage pool default")
+
+ def test_a_lookup_error_other_than_missing_is_raised(self):
+ p = proxy()
+ p.pool.storageVolLookupByName.side_effect = error(libvirt.VIR_ERR_INTERNAL_ERROR)
+ with self.assertRaises(libvirt.libvirtError):
+ p.live_clone_targets(FORM, 10, 5)
+
+ def test_a_pool_without_room_is_refused(self):
+ # 10 GiB copied plus max(10% of it, 5 GiB) = 15 GiB
+ self.refused(proxy(free=14 * GIB), "Storage pool default has 14.0 GiB free; the clone needs 15.0 GiB")
+ self.assertEqual(len(proxy(free=15 * GIB).live_clone_targets(FORM, 10, 5)), 1)
+
+ def test_the_percentage_counts_when_it_is_larger(self):
+ # 100 GiB copied plus max(10 GiB, 5 GiB) = 110 GiB
+ self.refused(proxy(free=110 * GIB - 1, capacity=100 * GIB), "the clone needs 110.0 GiB")
+
+
+class BackupTestCase(unittest.TestCase):
+ def test_begin_lists_only_the_targets(self):
+ p = proxy()
+ p.begin_clone_backup([{"dev": "vda", "path": "/pool/copy-vda.qcow2", "format": "qcow2"}])
+ xml, checkpoint, flags = p.instance.backupBegin.call_args.args
+ self.assertEqual(
+ xml,
+ ""
+ "",
+ )
+ self.assertIsNone(checkpoint)
+ self.assertEqual(
+ flags,
+ libvirt.VIR_DOMAIN_BACKUP_BEGIN_REUSE_EXTERNAL | VIR_DOMAIN_BACKUP_BEGIN_PRESERVE_SHUTDOWN_DOMAIN,
+ )
+
+ def test_targets_of_the_running_backup(self):
+ p = proxy()
+ p.instance.backupGetXMLDesc.return_value = (
+ ""
+ ""
+ ""
+ ""
+ )
+ self.assertEqual(p.backup_targets(), {"/pool/a.qcow2", "/pool/b.qcow2"})
+
+ def test_no_backup_running(self):
+ p = proxy()
+ p.instance.backupGetXMLDesc.side_effect = error(libvirt.VIR_ERR_NO_DOMAIN_BACKUP)
+ self.assertIsNone(p.backup_targets())
+ p.instance.backupGetXMLDesc.side_effect = error(libvirt.VIR_ERR_INTERNAL_ERROR)
+ with self.assertRaises(libvirt.libvirtError):
+ p.backup_targets()
+
+ def test_progress(self):
+ p = proxy()
+ p.instance.jobStats.return_value = {
+ "type": libvirt.VIR_DOMAIN_JOB_UNBOUNDED,
+ "operation": libvirt.VIR_DOMAIN_JOB_OPERATION_BACKUP,
+ "disk_processed": 3,
+ "disk_total": 10,
+ }
+ self.assertEqual(p.backup_progress(), (3, 10))
+ p.instance.jobStats.assert_called_once_with(0)
+ p.instance.jobStats.return_value = {"type": libvirt.VIR_DOMAIN_JOB_NONE}
+ self.assertIsNone(p.backup_progress())
+ p.instance.jobStats.return_value = {
+ "type": libvirt.VIR_DOMAIN_JOB_UNBOUNDED,
+ "operation": libvirt.VIR_DOMAIN_JOB_OPERATION_MIGRATION_OUT,
+ }
+ self.assertIsNone(p.backup_progress())
+
+ def test_result(self):
+ p = proxy()
+ backup = {"type": libvirt.VIR_DOMAIN_JOB_COMPLETED, "operation": libvirt.VIR_DOMAIN_JOB_OPERATION_BACKUP}
+ p.instance.jobStats.return_value = {**backup, "success": True}
+ self.assertEqual(p.backup_result(), (True, ""))
+ p.instance.jobStats.assert_called_with(
+ libvirt.VIR_DOMAIN_JOB_STATS_COMPLETED | libvirt.VIR_DOMAIN_JOB_STATS_KEEP_COMPLETED
+ )
+ p.instance.jobStats.return_value = {**backup, "success": False, "errmsg": "Operation canceled"}
+ self.assertEqual(p.backup_result(), (False, "Operation canceled"))
+ p.instance.jobStats.return_value = {}
+ self.assertIsNone(p.backup_result())
+ p.instance.jobStats.return_value = {
+ "type": libvirt.VIR_DOMAIN_JOB_COMPLETED,
+ "operation": libvirt.VIR_DOMAIN_JOB_OPERATION_MIGRATION_OUT,
+ "success": True,
+ }
+ self.assertIsNone(p.backup_result())
+
+ def test_freeze_thaw_abort(self):
+ p = proxy()
+ p.fs_freeze()
+ p.instance.fsFreeze.assert_called_once_with(None, 0)
+ p.fs_thaw()
+ p.instance.fsThaw.assert_called_once_with(None, 0)
+ p.abort_backup()
+ p.instance.abortJob.assert_called_once_with()
+
+
+class CreateCloneTargetTestCase(unittest.TestCase):
+ def test_an_empty_volume_at_the_source_capacity(self):
+ p = proxy()
+ p.wvm.storagePoolLookupByName.return_value = p.pool
+ p.pool.createXML.return_value.path.return_value = "/pool/copy-vda.qcow2"
+ target = {"pool": "default", "name": "copy-vda.qcow2", "capacity": 10 * GIB, "format": "qcow2"}
+ self.assertEqual(p.create_clone_target(target, 107, 108), "/pool/copy-vda.qcow2")
+ p.wvm.storagePoolLookupByName.assert_called_once_with("default")
+ xml, flags = p.pool.createXML.call_args.args
+ volume = etree.fromstring(xml)
+ self.assertEqual(volume.findtext("name"), "copy-vda.qcow2")
+ self.assertEqual(volume.findtext("capacity"), str(10 * GIB))
+ self.assertEqual(volume.find("target/format").get("type"), "qcow2")
+ self.assertEqual(volume.findtext("target/permissions/owner"), "107")
+ self.assertEqual(volume.findtext("target/permissions/group"), "108")
+ self.assertEqual(flags, 0)
+
+
+class DefineCloneTestCase(unittest.TestCase):
+ def test_the_definition_names_the_copies(self):
+ p = proxy()
+ p._defineXML = MagicMock()
+ p._defineXML.return_value.UUIDString.return_value = UUID
+ self.assertEqual(p.define_clone(FORM, UUID, {"vda": "/pool/copy-vda.qcow2"}), UUID)
+ (xml,), _ = p._defineXML.call_args
+ tree = etree.fromstring(xml)
+ self.assertEqual(tree.findtext("name"), "copy")
+ self.assertEqual(tree.findtext("uuid"), UUID)
+ self.assertEqual(tree.find("devices/disk[@device='disk']/source").get("file"), "/pool/copy-vda.qcow2")
+ self.assertEqual(tree.find("devices/disk[@device='cdrom']/source").get("file"), "/iso/x.iso")
+ self.assertEqual(tree.find("devices/interface/mac").get("address"), "52:54:00:00:00:02")
+ p.vol.storagePoolLookupByVolume.return_value.createXMLFrom.assert_not_called()
+
+ def test_a_missing_mac_is_refused(self):
+ p = proxy()
+ p._defineXML = MagicMock()
+ with self.assertRaisesRegex(ValueError, "No MAC address"):
+ p.define_clone({**FORM, "clone-net-mac-0": ""}, UUID, {"vda": "/pool/copy-vda.qcow2"})
+ p._defineXML.assert_not_called()
+
+ def test_clone_instance_copies_then_defines(self):
+ p = proxy()
+ p.get_status = lambda: 5
+ p.pool.createXMLFrom.return_value.path.return_value = "/pool/copy-vda.qcow2"
+ p.define_clone = MagicMock(return_value=UUID)
+ self.assertEqual(p.clone_instance(FORM, UUID), UUID)
+ p.pool.createXMLFrom.assert_called_once()
+ p.define_clone.assert_called_once_with(FORM, UUID, {"vda": "/pool/copy-vda.qcow2"})
+
+ def test_clone_instance_checks_the_macs_before_copying(self):
+ p = proxy()
+ p.get_status = lambda: 5
+ with self.assertRaisesRegex(ValueError, "No MAC address"):
+ p.clone_instance({**FORM, "clone-net-mac-0": ""}, UUID)
+ p.pool.createXMLFrom.assert_not_called()
+
+
+if __name__ == "__main__":
+ unittest.main()