'summary')
def test_get_relocate_spec(self):
+
+ delete_disk_attribute = True
+
+ def _create_side_effect(type):
+ obj = mock.Mock()
+ if type == "ns0:VirtualDiskFlatVer2BackingInfo":
+ del obj.eagerlyScrub
+ elif (type == "ns0:VirtualMachineRelocateSpec" and
+ delete_disk_attribute):
+ del obj.disk
+ else:
+ pass
+ return obj
+
factory = self.session.vim.client.factory
- spec = mock.Mock(spec=object)
- factory.create.return_value = spec
+ factory.create.side_effect = _create_side_effect
+
datastore = mock.sentinel.datastore
resource_pool = mock.sentinel.resource_pool
host = mock.sentinel.host
disk_move_type = mock.sentinel.disk_move_type
ret = self.vops._get_relocate_spec(datastore, resource_pool, host,
disk_move_type)
- self.assertEqual(spec, ret)
+
self.assertEqual(datastore, ret.datastore)
self.assertEqual(resource_pool, ret.pool)
self.assertEqual(host, ret.host)
self.assertEqual(disk_move_type, ret.diskMoveType)
+ # Test with disk locator.
+ delete_disk_attribute = False
+ disk_type = 'thin'
+ disk_device = mock.Mock()
+ ret = self.vops._get_relocate_spec(datastore, resource_pool, host,
+ disk_move_type, disk_type,
+ disk_device)
+
+ factory.create.side_effect = None
+ self.assertEqual(datastore, ret.datastore)
+ self.assertEqual(resource_pool, ret.pool)
+ self.assertEqual(host, ret.host)
+ self.assertEqual(disk_move_type, ret.diskMoveType)
+ self.assertIsInstance(ret.disk, list)
+ self.assertEqual(1, len(ret.disk))
+ disk_locator = ret.disk[0]
+ self.assertEqual(datastore, disk_locator.datastore)
+ self.assertEqual(disk_device.key, disk_locator.diskId)
+ backing = disk_locator.diskBackingInfo
+ self.assertIsInstance(backing.thinProvisioned, bool)
+ self.assertTrue(backing.thinProvisioned)
+ self.assertEqual('', backing.fileName)
+ self.assertEqual('persistent', backing.diskMode)
+
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_get_relocate_spec')
def test_relocate_backing(self, get_relocate_spec):
self.assertEqual(folder, ret)
get_parent.assert_called_once_with(backing, 'Folder')
- def test_get_clone_spec(self):
+ @mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
+ '_get_relocate_spec')
+ @mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
+ '_get_disk_device')
+ def test_get_clone_spec(self, get_disk_device, get_relocate_spec):
factory = self.session.vim.client.factory
- spec = mock.Mock(spec=object)
- factory.create.return_value = spec
+ factory.create.side_effect = lambda *args: mock.Mock()
+ relocate_spec = mock.sentinel.relocate_spec
+ get_relocate_spec.return_value = relocate_spec
+
datastore = mock.sentinel.datastore
disk_move_type = mock.sentinel.disk_move_type
snapshot = mock.sentinel.snapshot
- ret = self.vops._get_clone_spec(datastore, disk_move_type, snapshot)
- self.assertEqual(spec, ret)
+ disk_type = None
+ backing = mock.sentinel.backing
+ ret = self.vops._get_clone_spec(datastore, disk_move_type, snapshot,
+ backing, disk_type)
+
+ self.assertEqual(relocate_spec, ret.location)
+ self.assertFalse(ret.powerOn)
+ self.assertFalse(ret.template)
+ self.assertEqual(snapshot, ret.snapshot)
+ get_relocate_spec.assert_called_once_with(datastore, None, None,
+ disk_move_type, disk_type,
+ None)
+
+ disk_device = mock.sentinel.disk_device
+ get_disk_device.return_value = disk_device
+
+ disk_type = 'thin'
+ ret = self.vops._get_clone_spec(datastore, disk_move_type, snapshot,
+ backing, disk_type)
+
+ factory.create.side_effect = None
+ self.assertEqual(relocate_spec, ret.location)
+ self.assertFalse(ret.powerOn)
+ self.assertFalse(ret.template)
self.assertEqual(snapshot, ret.snapshot)
- self.assertEqual(spec, ret.location)
- self.assertEqual(datastore, ret.location.datastore)
- self.assertEqual(disk_move_type, ret.location.diskMoveType)
- expected_calls = [mock.call('ns0:VirtualMachineRelocateSpec'),
- mock.call('ns0:VirtualMachineCloneSpec')]
- factory.create.assert_has_calls(expected_calls, any_order=True)
+ get_disk_device.assert_called_once_with(backing)
+ get_relocate_spec.assert_called_with(datastore, None, None,
+ disk_move_type, disk_type,
+ disk_device)
@mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.'
'_get_clone_spec')
folder = mock.Mock(name='folder', spec=object)
folder._type = 'Folder'
task = mock.sentinel.task
- self.session.invoke_api.side_effect = [folder, task, folder, task]
+ self.session.invoke_api.side_effect = [folder, task, folder, task,
+ folder, task]
task_info = mock.Mock(spec=object)
task_info.result = mock.sentinel.new_backing
self.session.wait_for_task.return_value = task_info
# verify calls
self.assertEqual(mock.sentinel.new_backing, ret)
disk_move_type = 'moveAllDiskBackingsAndDisallowSharing'
- get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot)
+ get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot,
+ backing, None)
expected = [mock.call(vim_util, 'get_object_property',
self.session.vim, backing, 'parent'),
mock.call(self.session.vim, 'CloneVM_Task', backing,
# Test linked clone_backing
clone_type = volumeops.LINKED_CLONE_TYPE
+ self.session.invoke_api.reset_mock()
ret = self.vops.clone_backing(name, backing, snapshot, clone_type,
datastore)
# verify calls
self.assertEqual(mock.sentinel.new_backing, ret)
disk_move_type = 'createNewChildDiskBacking'
- get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot)
+ get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot,
+ backing, None)
expected = [mock.call(vim_util, 'get_object_property',
self.session.vim, backing, 'parent'),
mock.call(self.session.vim, 'CloneVM_Task', backing,
- folder=folder, name=name, spec=clone_spec),
- mock.call(vim_util, 'get_object_property',
+ folder=folder, name=name, spec=clone_spec)]
+ self.assertEqual(expected, self.session.invoke_api.mock_calls)
+
+ # Test disk type conversion.
+ clone_type = None
+ disk_type = 'thin'
+ self.session.invoke_api.reset_mock()
+ ret = self.vops.clone_backing(name, backing, snapshot, clone_type,
+ datastore, disk_type)
+
+ self.assertEqual(mock.sentinel.new_backing, ret)
+ disk_move_type = 'moveAllDiskBackingsAndDisallowSharing'
+ get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot,
+ backing, disk_type)
+ expected = [mock.call(vim_util, 'get_object_property',
self.session.vim, backing, 'parent'),
mock.call(self.session.vim, 'CloneVM_Task', backing,
folder=folder, name=name, spec=clone_spec)]
self.session.vim, backing,
'config.hardware.device')
+ backing.__class__.__name__ = ' VirtualDiskSparseVer2BackingInfo'
+ self.assertRaises(AssertionError, self.vops.get_vmdk_path, backing)
+
def test_create_virtual_disk(self):
task = mock.Mock()
invoke_api = self.session.invoke_api
self._session.vim, datastore,
'summary')
+ def _create_relocate_spec_disk_locator(self, datastore, disk_type,
+ disk_device):
+ """Creates spec for disk type conversion during relocate."""
+ cf = self._session.vim.client.factory
+ disk_locator = cf.create("ns0:VirtualMachineRelocateSpecDiskLocator")
+ disk_locator.datastore = datastore
+ disk_locator.diskId = disk_device.key
+ disk_locator.diskBackingInfo = self._create_disk_backing(disk_type,
+ None)
+ return disk_locator
+
def _get_relocate_spec(self, datastore, resource_pool, host,
- disk_move_type):
+ disk_move_type, disk_type=None, disk_device=None):
"""Return spec for relocating volume backing.
:param datastore: Reference to the datastore
:param resource_pool: Reference to the resource pool
:param host: Reference to the host
:param disk_move_type: Disk move type option
+ :param disk_type: Destination disk type
+ :param disk_device: Virtual device corresponding to the disk
:return: Spec for relocation
"""
cf = self._session.vim.client.factory
relocate_spec.host = host
relocate_spec.diskMoveType = disk_move_type
- LOG.debug("Spec for relocating the backing: %s." % relocate_spec)
+ if disk_type is not None and disk_device is not None:
+ disk_locator = self._create_relocate_spec_disk_locator(datastore,
+ disk_type,
+ disk_device)
+ relocate_spec.disk = [disk_locator]
+
+ LOG.debug("Spec for relocating the backing: %s.", relocate_spec)
return relocate_spec
def relocate_backing(self, backing, datastore, resource_pool, host):
"""
return self._get_parent(backing, 'Folder')
- def _get_clone_spec(self, datastore, disk_move_type, snapshot):
+ def _get_clone_spec(self, datastore, disk_move_type, snapshot, backing,
+ disk_type):
"""Get the clone spec.
:param datastore: Reference to datastore
:param disk_move_type: Disk move type
:param snapshot: Reference to snapshot
+ :param backing: Source backing VM
+ :param disk_type: Disk type of clone
:return: Clone spec
"""
+ if disk_type is not None:
+ disk_device = self._get_disk_device(backing)
+ else:
+ disk_device = None
+
relocate_spec = self._get_relocate_spec(datastore, None, None,
- disk_move_type)
+ disk_move_type, disk_type,
+ disk_device)
cf = self._session.vim.client.factory
clone_spec = cf.create('ns0:VirtualMachineCloneSpec')
clone_spec.location = relocate_spec
clone_spec.template = False
clone_spec.snapshot = snapshot
- LOG.debug("Spec for cloning the backing: %s." % clone_spec)
+ LOG.debug("Spec for cloning the backing: %s.", clone_spec)
return clone_spec
- def clone_backing(self, name, backing, snapshot, clone_type, datastore):
+ def clone_backing(self, name, backing, snapshot, clone_type, datastore,
+ disk_type=None):
"""Clone backing.
If the clone_type is 'full', then a full clone of the source volume
:param snapshot: Snapshot point from which the clone should be done
:param clone_type: Whether a full clone or linked clone is to be made
:param datastore: Reference to the datastore entity
+ :param disk_type: Disk type of the clone
"""
LOG.debug("Creating a clone of backing: %(back)s, named: %(name)s, "
"clone type: %(type)s from snapshot: %(snap)s on "
- "datastore: %(ds)s" %
+ "datastore: %(ds)s with disk type: %(disk_type)s.",
{'back': backing, 'name': name, 'type': clone_type,
- 'snap': snapshot, 'ds': datastore})
+ 'snap': snapshot, 'ds': datastore, 'disk_type': disk_type})
folder = self._get_folder(backing)
if clone_type == LINKED_CLONE_TYPE:
disk_move_type = 'createNewChildDiskBacking'
else:
disk_move_type = 'moveAllDiskBackingsAndDisallowSharing'
- clone_spec = self._get_clone_spec(datastore, disk_move_type, snapshot)
+ clone_spec = self._get_clone_spec(datastore, disk_move_type, snapshot,
+ backing, disk_type)
task = self._session.invoke_api(self._session.vim, 'CloneVM_Task',
backing, folder=folder, name=name,
spec=clone_spec)
return self._session.invoke_api(vim_util, 'get_object_property',
self._session.vim, entity, 'name')
- def get_vmdk_path(self, backing):
- """Get the vmdk file name of the backing.
-
- The vmdk file path of the backing returned is of the form:
- "[datastore1] my_folder/my_vm.vmdk"
-
- :param backing: Reference to the backing
- :return: VMDK file path of the backing
- """
+ def _get_disk_device(self, backing):
+ """Get the virtual device corresponding to disk."""
hardware_devices = self._session.invoke_api(vim_util,
'get_object_property',
self._session.vim,
hardware_devices = hardware_devices.VirtualDevice
for device in hardware_devices:
if device.__class__.__name__ == "VirtualDisk":
- bkng = device.backing
- if bkng.__class__.__name__ == "VirtualDiskFlatVer2BackingInfo":
- return bkng.fileName
+ return device
+
+ def get_vmdk_path(self, backing):
+ """Get the vmdk file name of the backing.
+
+ The vmdk file path of the backing returned is of the form:
+ "[datastore1] my_folder/my_vm.vmdk"
+
+ :param backing: Reference to the backing
+ :return: VMDK file path of the backing
+ """
+ disk_device = self._get_disk_device(backing)
+ backing = disk_device.backing
+ if backing.__class__.__name__ != "VirtualDiskFlatVer2BackingInfo":
+ msg = _("Invalid disk backing: %s.") % backing.__class__.__name__
+ LOG.error(msg)
+ raise AssertionError(msg)
+ return backing.fileName
def _get_virtual_disk_create_spec(self, size_in_kb, adapter_type,
disk_type):
Code Review / openstack-build / cinder-build.git / commitdiff