Add test for ignoring vdis when snapshoting a VM

See: xapi-project/xen-api#4563

Signed-off-by: BenjiReis <[email protected]>

conftest.py

@@ -138,6 +138,16 @@ def hostB1(hosts):
     logging.info(">>> hostB1 present: %s" % _hostB1)
     yield _hostB1
 
+@pytest.fixture(scope='session')
+def local_sr_on_hostA1(hostA1):
+    """ A local SR on the pool's second host. """
+    srs = hostA1.local_vm_srs()
+    assert len(srs) > 0, "a local SR is required on the pool's master"
+    # use the first local SR found
+    sr = srs[0]
+    logging.info(">> local SR on hostA1 present : %s" % sr.uuid)
+    yield sr
+
 @pytest.fixture(scope='session')
 def local_sr_on_hostA2(hostA2):
     """ A local SR on the pool's second host. """

lib/common.py

@@ -771,11 +771,12 @@ def migrate(self, target_host, sr=None):
         self.previous_host = self.host
         self.host = target_host
 
-    def snapshot(self):
+    def snapshot(self, ignore_vdis=None):
         logging.info("Snapshot VM")
-        return Snapshot(self.host.xe('vm-snapshot', {'uuid': self.uuid,
-                                                     'new-name-label': 'Snapshot of %s' % self.uuid}),
-                        self.host)
+        args = {'uuid': self.uuid, 'new-name-label': 'Snapshot of %s' % self.uuid}
+        if ignore_vdis:
+            args['ignore-vdi-uuids'] = ','.join(ignore_vdis)
+        return Snapshot(self.host.xe('vm-snapshot', args), self.host)
 
     def checkpoint(self):
         logging.info("Checkpoint VM")

tests/snapshot/conftest.py

@@ -0,0 +1,35 @@
+import logging
+import pytest
+
+@pytest.fixture(scope='module')
+def vdis(host, local_sr_on_hostA1):
+    sr = local_sr_on_hostA1
+
+    logging.info('> Creating VDIs')
+    vdi_A = host.xe('vdi-create', {'name-label': 'VDI_A', 'virtual-size': '64', 'sr-uuid': sr.uuid})
+    vdi_B = host.xe('vdi-create', {'name-label': 'VDI_B', 'virtual-size': '64', 'sr-uuid': sr.uuid})
+    vdi_C = host.xe('vdi-create', {'name-label': 'VDI_C', 'virtual-size': '64', 'sr-uuid': sr.uuid})
+
+    yield vdi_A, vdi_B, vdi_C
+
+    logging.info('< Destroying VDIs')
+    for vdi in [vdi_A, vdi_B, vdi_C]:
+        host.xe('vdi-destroy', {'uuid': vdi})
+
+@pytest.fixture(scope='module')
+def vm_with_vbds(host, vdis, imported_vm):
+    vm = imported_vm
+    vdi_A, vdi_B, vdi_C = vdis
+
+    host.xe('vbd-create', {
+        'vm-uuid': vm.uuid, 'mode': 'RW', 'type': 'Disk', 'device': 'xvda1', 'vdi-uuid': vdi_A
+    })
+    host.xe('vbd-create', {
+        'vm-uuid': vm.uuid, 'mode': 'RW', 'type': 'Disk', 'device': 'xvdb1', 'vdi-uuid': vdi_B
+    })
+    host.xe('vbd-create', {
+        'vm-uuid': vm.uuid, 'mode': 'RW', 'type': 'Disk', 'device': 'xvdc', 'vdi-uuid': vdi_C
+    })
+
+    vm.start()
+    yield vm

tests/snapshot/test_snapshot_ignore_vdis.py

@@ -0,0 +1,81 @@
+import logging
+
+# Requirements:
+# - an XCP-ng host (--hosts) > 8.2
+# - a VM (--vm)
+
+def test_snapshot(host, vdis, vm_with_vbds):
+    vm = vm_with_vbds
+
+    snapshot = vm.snapshot()
+
+    snap_vbds = snapshot.param_get('VBDs')
+    snap_vbds = snap_vbds.split('; ')
+
+    snap_vdis = list(map(
+        lambda vbd: host.xe('vbd-param-get', {'uuid': vbd, 'param-name': 'vdi-uuid'}),
+        snap_vbds
+    ))
+    snap_vdis = [vdi for vdi in snap_vdis if vdi != '<not in database>']
+
+    orig_vdis = list(map(
+        lambda vdi: host.xe('vdi-param-get', {'uuid': vdi, 'param-name': 'snapshot-of'}),
+        snap_vdis
+    ))
+
+    for vdi in vdis:
+        assert vdi in orig_vdis
+
+    snapshot.destroy()
+
+def test_snapshot_ignore_vdi(host, vdis, vm_with_vbds):
+    vdi_A, vdi_B, vdi_C = vdis
+    vm = vm_with_vbds
+
+    snapshot = vm.snapshot(ignore_vdis=[vdi_B])
+
+    snap_vbds = snapshot.param_get('VBDs')
+    snap_vbds = snap_vbds.split('; ')
+
+    snap_vdis = list(map(
+        lambda vbd: host.xe('vbd-param-get', {'uuid': vbd, 'param-name': 'vdi-uuid'}),
+        snap_vbds
+    ))
+    snap_vdis = [vdi for vdi in snap_vdis if vdi != '<not in database>']
+
+    orig_vdis = list(map(
+        lambda vdi: host.xe('vdi-param-get', {'uuid': vdi, 'param-name': 'snapshot-of'}),
+        snap_vdis
+    ))
+
+    assert vdi_A in orig_vdis
+    assert vdi_B not in orig_vdis
+    assert vdi_C in orig_vdis
+
+    snapshot.destroy()
+
+def test_snapshot_ignore_multiple_vdis(host, vdis, vm_with_vbds):
+    vdi_A, vdi_B, vdi_C = vdis
+    vm = vm_with_vbds
+
+    snapshot = vm.snapshot(ignore_vdis=[vdi_B, vdi_C])
+
+    snap_vbds = snapshot.param_get('VBDs')
+    snap_vbds = snap_vbds.split('; ')
+
+    snap_vdis = list(map(
+        lambda vbd: host.xe('vbd-param-get', {'uuid': vbd, 'param-name': 'vdi-uuid'}),
+        snap_vbds
+    ))
+    snap_vdis = [vdi for vdi in snap_vdis if vdi != '<not in database>']
+
+    orig_vdis = list(map(
+        lambda vdi: host.xe('vdi-param-get', {'uuid': vdi, 'param-name': 'snapshot-of'}),
+        snap_vdis
+    ))
+
+    assert vdi_A in orig_vdis
+    assert vdi_B not in orig_vdis
+    assert vdi_C not in orig_vdis
+
+    snapshot.destroy()