From 6846ed79c31f02bbacf48d2bb22e7128549c56f0 Mon Sep 17 00:00:00 2001 From: Subramanian Neelakantan Date: Wed, 5 Feb 2014 17:39:05 +0530 Subject: [PATCH] VMware: changing volumeops tests from mox to mock. Changing unit tests implementation of volumeops to mock. In the process also moving all these unit tests into a separate class for better test organization. Partial-Bug: #1261097 Change-Id: I08b2341b22ed9c8f63930a2c73a9737cbe7ef540 --- cinder/tests/test_vmware_vmdk.py | 729 +---------------------- cinder/tests/test_vmware_volumeops.py | 793 ++++++++++++++++++++++++++ 2 files changed, 799 insertions(+), 723 deletions(-) create mode 100644 cinder/tests/test_vmware_volumeops.py diff --git a/cinder/tests/test_vmware_vmdk.py b/cinder/tests/test_vmware_vmdk.py index 6ff8f286b..9ca3b1438 100644 --- a/cinder/tests/test_vmware_vmdk.py +++ b/cinder/tests/test_vmware_vmdk.py @@ -45,6 +45,12 @@ class FakeVim(object): def Login(self, session_manager, userName, password): return mox.MockAnything() + def Logout(self, session_manager): + pass + + def TerminateSession(self, session_manager, sessionId): + pass + class FakeTaskInfo(object): def __init__(self, state, result=None): @@ -255,83 +261,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): m.UnsetStubs() m.VerifyAll() - def test_continue_retrieval(self): - """Test continue_retrieval.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - self._session.invoke_api(vim_util, 'continue_retrieval', - self._vim, mox.IgnoreArg()) - - m.ReplayAll() - self._volumeops.continue_retrieval(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_cancel_retrieval(self): - """Test cancel_retrieval.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - self._session.invoke_api(vim_util, 'cancel_retrieval', - self._vim, mox.IgnoreArg()) - - m.ReplayAll() - self._volumeops.cancel_retrieval(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_get_backing(self): - """Test get_backing.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - self._session.invoke_api(vim_util, 'get_objects', - self._vim, 'VirtualMachine', - self.MAX_OBJECTS) - - m.ReplayAll() - self._volumeops.get_backing(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_get_backing_multiple_retrieval(self): - """Test get_backing with multiple retrieval.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - retrieve_result = FakeRetrieveResult([], 'my_token') - self._session.invoke_api(vim_util, 'get_objects', - self._vim, 'VirtualMachine', - self.MAX_OBJECTS).AndReturn(retrieve_result) - m.StubOutWithMock(self._volumeops, 'cancel_retrieval') - self._volumeops.continue_retrieval(retrieve_result) - - m.ReplayAll() - self._volumeops.get_backing(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_delete_backing(self): - """Test delete_backing.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_vm') - self._session.invoke_api(self._vim, 'Destroy_Task', backing) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(mox.IgnoreArg()) - - m.ReplayAll() - self._volumeops.delete_backing(backing) - m.UnsetStubs() - m.VerifyAll() - def test_delete_volume_without_backing(self): """Test delete_volume without backing.""" m = self.mox @@ -385,243 +314,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): self._driver.terminate_connection(mox.IgnoreArg(), mox.IgnoreArg(), force=mox.IgnoreArg()) - def test_get_host(self): - """Test get_host.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - instance = FakeObject() - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, instance, 'runtime.host') - - m.ReplayAll() - self._volumeops.get_host(instance) - m.UnsetStubs() - m.VerifyAll() - - def test_get_hosts(self): - """Test get_hosts.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - self._session.invoke_api(vim_util, 'get_objects', self._vim, - 'HostSystem', self.MAX_OBJECTS) - - m.ReplayAll() - self._volumeops.get_hosts() - m.UnsetStubs() - m.VerifyAll() - - def _create_host_mounts(self, access_mode, host, set_accessible=True, - is_accessible=True, mounted=True): - """Create host mount value of datastore with single mount info. - - :param access_mode: string specifying the read/write permission - :param set_accessible: specify whether accessible property - should be set - :param is_accessible: boolean specifying whether the datastore - is accessible to host - :param host: managed object reference of the connected - host - :return: list of host mount info - """ - mntInfo = FakeObject() - mntInfo.accessMode = access_mode - if set_accessible: - mntInfo.accessible = is_accessible - mntInfo.mounted = mounted - - host_mount = FakeObject() - host_mount.key = host - host_mount.mountInfo = mntInfo - host_mounts = FakeObject() - host_mounts.DatastoreHostMount = [host_mount] - - return host_mounts - - def test_is_valid_with_accessible_attr(self): - """Test _is_valid with accessible attribute.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - - datastore = FakeMor('Datastore', 'my_ds') - host = FakeMor('HostSystem', "my_host") - host_mounts = self._create_host_mounts("readWrite", host) - - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, datastore, - 'host').AndReturn(host_mounts) - - m.ReplayAll() - self.assertTrue(self._volumeops._is_valid(datastore, host)) - m.UnsetStubs() - m.VerifyAll() - - def test_is_valid_without_accessible_attr(self): - """Test _is_valid without accessible attribute.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - - datastore = FakeMor('Datastore', 'my_ds') - host = FakeMor('HostSystem', "my_host") - host_mounts = self._create_host_mounts("readWrite", host, False) - - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, datastore, - 'host').AndReturn(host_mounts) - m.StubOutWithMock(self._volumeops, 'get_summary') - summary = FakeObject() - summary.accessible = True - self._volumeops.get_summary(datastore).AndReturn(summary) - - m.ReplayAll() - self.assertTrue(self._volumeops._is_valid(datastore, host)) - m.UnsetStubs() - m.VerifyAll() - - def test_get_dss_rp(self): - """Test get_dss_rp.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - - datastore = FakeMor('Datastore', 'my_ds') - datastore_prop = FakeProp(name='datastore', - val=FakeManagedObjectReference([datastore])) - - compute_resource = FakeMor('ClusterComputeResource', 'my_cluster') - compute_resource_prop = FakeProp(name='parent', val=compute_resource) - - props = [FakeElem(prop_set=[datastore_prop, compute_resource_prop])] - host = FakeMor('HostSystem', "my_host") - self._session.invoke_api(vim_util, 'get_object_properties', - self._vim, host, - ['datastore', 'parent']).AndReturn(props) - - host_mounts = self._create_host_mounts("readWrite", host) - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, datastore, - 'host').AndReturn(host_mounts) - - resource_pool = FakeMor('ResourcePool', 'my_res_pool') - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, compute_resource, - 'resourcePool').AndReturn(resource_pool) - - m.ReplayAll() - (datastores_ret, resource_pool_ret) = self._volumeops.get_dss_rp(host) - self.assertTrue(len(datastores_ret) == 1) - self.assertEqual(datastores_ret[0], datastore) - self.assertEqual(resource_pool_ret, resource_pool) - - m.UnsetStubs() - m.VerifyAll() - - def test_get_dss_rp_without_datastores(self): - """Test get_dss_rp without datastores.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - - host = FakeObject() - self._session.invoke_api(vim_util, 'get_object_properties', - self._vim, host, - ['datastore', 'parent']).AndReturn([]) - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, mox.IgnoreArg(), 'resourcePool') - - m.ReplayAll() - self.assertRaises(error_util.VimException, self._volumeops.get_dss_rp, - host) - m.UnsetStubs() - m.VerifyAll() - - def test_get_parent(self): - """Test get_parent.""" - # Not recursive - child = FakeMor('Parent', 'my_parent') - parent = self._volumeops._get_parent(child, 'Parent') - self.assertEqual(parent, child) - - # Recursive - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - parent = FakeMor('Parent', 'my_parent1') - child = FakeMor('Child', 'my_child') - self._session.invoke_api(vim_util, 'get_object_property', self._vim, - child, 'parent').AndReturn(parent) - - m.ReplayAll() - ret = self._volumeops._get_parent(child, 'Parent') - self.assertEqual(ret, parent) - m.UnsetStubs() - m.VerifyAll() - - def test_get_dc(self): - """Test get_dc.""" - m = self.mox - m.StubOutWithMock(self._volumeops, '_get_parent') - self._volumeops._get_parent(mox.IgnoreArg(), 'Datacenter') - - m.ReplayAll() - self._volumeops.get_dc(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_get_vmfolder(self): - """Test get_vmfolder.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - datacenter = FakeMor('Datacenter', 'my_dc') - self._session.invoke_api(vim_util, 'get_object_property', self._vim, - datacenter, 'vmFolder') - - m.ReplayAll() - dc = self._volumeops.get_vmfolder(datacenter) - m.UnsetStubs() - m.VerifyAll() - - def test_create_backing(self): - """Test create_backing.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - folder = FakeMor('Folder', 'my_fol') - resource_pool = FakeMor('ResourcePool', 'my_rs') - host = FakeMor('HostSystem', 'my_host') - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'CreateVM_Task', folder, - config=mox.IgnoreArg(), pool=resource_pool, - host=host).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - task_info = FakeTaskInfo('success', mox.IgnoreArg()) - self._session.wait_for_task(task).AndReturn(task_info) - name = 'my_vm' - size_kb = 1 * units.MiB - disk_type = 'thick' - ds_name = 'my_ds' - m.StubOutWithMock(self._volumeops, '_get_create_spec') - self._volumeops._get_create_spec(name, size_kb, disk_type, ds_name) - - m.ReplayAll() - self._volumeops.create_backing(name, size_kb, disk_type, folder, - resource_pool, host, ds_name) - m.UnsetStubs() - m.VerifyAll() - def test_create_backing_in_inventory_multi_hosts(self): """Test _create_backing_in_inventory scanning multiple hosts.""" m = self.mox @@ -650,39 +342,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): m.UnsetStubs() m.VerifyAll() - def test_get_datastore(self): - """Test get_datastore.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_back') - datastore = FakeMor('Datastore', 'my_ds') - datastores = FakeManagedObjectReference([datastore]) - self._session.invoke_api(vim_util, 'get_object_property', self._vim, - backing, 'datastore').AndReturn(datastores) - - m.ReplayAll() - result = self._volumeops.get_datastore(backing) - self.assertEqual(result, datastore) - m.UnsetStubs() - m.VerifyAll() - - def test_get_summary(self): - """Test get_summary.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - datastore = FakeMor('Datastore', 'my_ds') - self._session.invoke_api(vim_util, 'get_object_property', self._vim, - datastore, 'summary') - - m.ReplayAll() - self._volumeops.get_summary(datastore) - m.UnsetStubs() - m.VerifyAll() - def test_init_conn_with_instance_and_backing(self): """Test initialize_connection with instance and backing.""" m = self.mox @@ -875,29 +534,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): m.UnsetStubs() m.VerifyAll() - def test_create_snapshot_operation(self): - """Test volumeops.create_snapshot.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - name = 'snapshot_name' - description = 'snapshot_desc' - backing = FakeMor('VirtualMachine', 'my_back') - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'CreateSnapshot_Task', backing, - name=name, description=description, - memory=False, quiesce=False).AndReturn(task) - result = FakeMor('VirtualMachineSnapshot', 'my_snap') - success_task_info = FakeTaskInfo('success', result=result) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task).AndReturn(success_task_info) - - m.ReplayAll() - self._volumeops.create_snapshot(backing, name, description) - m.UnsetStubs() - m.VerifyAll() - def test_create_snapshot_without_backing(self): """Test vmdk.create_snapshot without backing.""" m = self.mox @@ -948,80 +584,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): self.assertRaises(exception.InvalidVolume, self._driver.create_snapshot, snapshot) - def test_get_snapshot_from_tree(self): - """Test _get_snapshot_from_tree.""" - volops = volumeops.VMwareVolumeOps - ret = volops._get_snapshot_from_tree(mox.IgnoreArg(), None) - self.assertIsNone(ret) - name = 'snapshot_name' - snapshot = FakeMor('VirtualMachineSnapshot', 'my_snap') - root = FakeSnapshotTree(name='snapshot_name', snapshot=snapshot) - ret = volops._get_snapshot_from_tree(name, root) - self.assertEqual(ret, snapshot) - snapshot1 = FakeMor('VirtualMachineSnapshot', 'my_snap_1') - root = FakeSnapshotTree(name='snapshot_name_1', snapshot=snapshot1, - childSnapshotList=[root]) - ret = volops._get_snapshot_from_tree(name, root) - self.assertEqual(ret, snapshot) - - def test_get_snapshot(self): - """Test get_snapshot.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - name = 'snapshot_name' - backing = FakeMor('VirtualMachine', 'my_back') - root = FakeSnapshotTree() - tree = FakeSnapshotTree(tree=[root]) - self._session.invoke_api(vim_util, 'get_object_property', - self._session.vim, backing, - 'snapshot').AndReturn(tree) - volops = volumeops.VMwareVolumeOps - m.StubOutWithMock(volops, '_get_snapshot_from_tree') - volops._get_snapshot_from_tree(name, root) - - m.ReplayAll() - self._volumeops.get_snapshot(backing, name) - m.UnsetStubs() - m.VerifyAll() - - def test_delete_snapshot_not_present(self): - """Test volumeops.delete_snapshot, when not present.""" - m = self.mox - m.StubOutWithMock(self._volumeops, 'get_snapshot') - name = 'snapshot_name' - backing = FakeMor('VirtualMachine', 'my_back') - self._volumeops.get_snapshot(backing, name) - - m.ReplayAll() - self._volumeops.delete_snapshot(backing, name) - m.UnsetStubs() - m.VerifyAll() - - def test_delete_snapshot_when_present(self): - """Test volumeops.delete_snapshot, when it is present.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - m.StubOutWithMock(self._volumeops, 'get_snapshot') - name = 'snapshot_name' - backing = FakeMor('VirtualMachine', 'my_back') - snapshot = FakeMor('VirtualMachineSnapshot', 'my_snap') - self._volumeops.get_snapshot(backing, name).AndReturn(snapshot) - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._session.vim, - 'RemoveSnapshot_Task', snapshot, - removeChildren=False).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.delete_snapshot(backing, name) - m.UnsetStubs() - m.VerifyAll() - def test_delete_snapshot_without_backing(self): """Test delete_snapshot without backing.""" m = self.mox @@ -1090,46 +652,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): m.UnsetStubs() m.VerifyAll() - def test_get_path_name(self): - """Test get_path_name.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_back') - - class FakePath(object): - def __init__(self, path=None): - self.vmPathName = path - - path = FakePath() - self._session.invoke_api(vim_util, 'get_object_property', self._vim, - backing, 'config.files').AndReturn(path) - - m.ReplayAll() - self._volumeops.get_path_name(backing) - m.UnsetStubs() - m.VerifyAll() - - def test_delete_file(self): - """Test _delete_file.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - src_path = 'src_path' - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'DeleteDatastoreFile_Task', - mox.IgnoreArg(), name=src_path, - datacenter=mox.IgnoreArg()).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.delete_file(src_path) - m.UnsetStubs() - m.VerifyAll() - def test_clone_backing_by_copying(self): """Test _clone_backing_by_copying.""" m = self.mox @@ -1246,117 +768,6 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase): m.UnsetStubs() m.VerifyAll() - def test_get_entity_name(self): - """Test volumeops get_entity_name.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - entity = FakeMor('VirtualMachine', 'virt') - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, entity, 'name') - - m.ReplayAll() - self._volumeops.get_entity_name(entity) - m.UnsetStubs() - m.VerifyAll() - - def test_get_vmdk_path(self): - """Test volumeops get_vmdk_path.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_back') - vmdk_path = '[datastore 1] folders/myvols/volume-123.vmdk' - - class VirtualDisk: - pass - virtualDisk = VirtualDisk() - - class VirtualDiskFlatVer2BackingInfo: - pass - backingInfo = VirtualDiskFlatVer2BackingInfo() - backingInfo.fileName = vmdk_path - virtualDisk.backing = backingInfo - devices = [FakeObject(), virtualDisk, FakeObject()] - - moxed = self._session.invoke_api(vim_util, 'get_object_property', - self._vim, backing, - 'config.hardware.device') - moxed.AndReturn(devices) - - m.ReplayAll() - actual_vmdk_path = self._volumeops.get_vmdk_path(backing) - self.assertEqual(backingInfo.__class__.__name__, - 'VirtualDiskFlatVer2BackingInfo') - self.assertEqual(virtualDisk.__class__.__name__, 'VirtualDisk') - self.assertEqual(actual_vmdk_path, vmdk_path) - m.UnsetStubs() - m.VerifyAll() - - def test_copy_vmdk_file(self): - """Test copy_vmdk_file.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - dc_ref = FakeMor('Datacenter', 'dc1') - src_path = 'src_path' - dest_path = 'dest_path' - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'CopyVirtualDisk_Task', - mox.IgnoreArg(), sourceName=src_path, - sourceDatacenter=dc_ref, destName=dest_path, - destDatacenter=dc_ref, - force=True).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.copy_vmdk_file(dc_ref, src_path, dest_path) - m.UnsetStubs() - m.VerifyAll() - - def test_delete_vmdk_file(self): - """Test delete_vmdk_file.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - dc_ref = FakeMor('Datacenter', 'dc1') - vmdk_path = 'vmdk_path' - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'DeleteVirtualDisk_Task', - mox.IgnoreArg(), name=vmdk_path, - datacenter=dc_ref).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.delete_vmdk_file(vmdk_path, dc_ref) - m.UnsetStubs() - m.VerifyAll() - - def test_split_datastore_path(self): - """Test volumeops split_datastore_path.""" - test1 = '[datastore1] myfolder/mysubfolder/myvm.vmx' - (datastore, - folder, - file_name) = volumeops.split_datastore_path(test1) - self.assertEqual(datastore, 'datastore1') - self.assertEqual(folder, 'myfolder/mysubfolder/') - self.assertEqual(file_name, 'myvm.vmx') - test2 = '[datastore2 ] myfolder/myvm.vmdk' - (datastore, - folder, - file_name) = volumeops.split_datastore_path(test2) - self.assertEqual(datastore, 'datastore2') - self.assertEqual(folder, 'myfolder/') - self.assertEqual(file_name, 'myvm.vmdk') - test3 = 'myfolder/myvm.vmdk' - self.assertRaises(IndexError, volumeops.split_datastore_path, test3) - def test_copy_image_to_volume_non_vmdk(self): """Test copy_image_to_volume for a non-vmdk disk format.""" m = self.mox @@ -1636,92 +1047,6 @@ class VMwareVcVmdkDriverTestCase(VMwareEsxVmdkDriverTestCase): super(VMwareVcVmdkDriverTestCase, self).setUp() self._driver = vmdk.VMwareVcVmdkDriver(configuration=self._config) - def test_create_folder_not_present(self): - """Test create_folder when not present.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - parent_folder = FakeMor('Folder', 'my_par_fol') - child_entities = FakeManagedObjectReference() - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, parent_folder, - 'childEntity').AndReturn(child_entities) - self._session.invoke_api(self._vim, 'CreateFolder', parent_folder, - name='child_folder_name') - - m.ReplayAll() - dc = self._volumeops.create_folder(parent_folder, 'child_folder_name') - m.UnsetStubs() - m.VerifyAll() - - def test_create_folder_already_present(self): - """Test create_folder when already present.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - parent_folder = FakeMor('Folder', 'my_par_fol') - child_folder = FakeMor('Folder', 'my_child_fol') - child_entities = FakeManagedObjectReference([child_folder]) - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, parent_folder, - 'childEntity').AndReturn(child_entities) - self._session.invoke_api(vim_util, 'get_object_property', - self._vim, child_folder, - 'name').AndReturn('child_folder_name') - - m.ReplayAll() - fol = self._volumeops.create_folder(parent_folder, 'child_folder_name') - self.assertEqual(fol, child_folder) - m.UnsetStubs() - m.VerifyAll() - - def test_relocate_backing(self): - """Test relocate_backing.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._volumeops, '_get_relocate_spec') - datastore = FakeMor('Datastore', 'my_ds') - resource_pool = FakeMor('ResourcePool', 'my_rp') - host = FakeMor('HostSystem', 'my_host') - disk_move_type = 'moveAllDiskBackingsAndAllowSharing' - self._volumeops._get_relocate_spec(datastore, resource_pool, host, - disk_move_type) - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_back') - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'RelocateVM_Task', - backing, spec=mox.IgnoreArg()).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.relocate_backing(backing, datastore, - resource_pool, host) - m.UnsetStubs() - m.VerifyAll() - - def test_move_backing_to_folder(self): - """Test move_backing_to_folder.""" - m = self.mox - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - backing = FakeMor('VirtualMachine', 'my_back') - folder = FakeMor('Folder', 'my_fol') - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'MoveIntoFolder_Task', - folder, list=[backing]).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - self._session.wait_for_task(task) - - m.ReplayAll() - self._volumeops.move_backing_to_folder(backing, folder) - m.UnsetStubs() - m.VerifyAll() - def test_init_conn_with_instance_and_backing(self): """Test initialize_connection with instance and backing.""" m = self.mox @@ -1816,48 +1141,6 @@ class VMwareVcVmdkDriverTestCase(VMwareEsxVmdkDriverTestCase): m.UnsetStubs() m.VerifyAll() - def test_get_folder(self): - """Test _get_folder.""" - m = self.mox - m.StubOutWithMock(self._volumeops, '_get_parent') - self._volumeops._get_parent(mox.IgnoreArg(), 'Folder') - - m.ReplayAll() - self._volumeops._get_folder(mox.IgnoreArg()) - m.UnsetStubs() - m.VerifyAll() - - def test_volumeops_clone_backing(self): - """Test volumeops.clone_backing.""" - m = self.mox - m.StubOutWithMock(self._volumeops, '_get_parent') - backing = FakeMor('VirtualMachine', 'my_back') - folder = FakeMor('Folder', 'my_fol') - self._volumeops._get_folder(backing).AndReturn(folder) - m.StubOutWithMock(self._volumeops, '_get_clone_spec') - name = 'name' - snapshot = FakeMor('VirtualMachineSnapshot', 'my_snap') - datastore = FakeMor('Datastore', 'my_ds') - self._volumeops._get_clone_spec(datastore, mox.IgnoreArg(), snapshot) - m.StubOutWithMock(api.VMwareAPISession, 'vim') - self._session.vim = self._vim - m.StubOutWithMock(self._session, 'invoke_api') - task = FakeMor('Task', 'my_task') - self._session.invoke_api(self._vim, 'CloneVM_Task', backing, - folder=folder, name=name, - spec=mox.IgnoreArg()).AndReturn(task) - m.StubOutWithMock(self._session, 'wait_for_task') - clone = FakeMor('VirtualMachine', 'my_clone') - task_info = FakeTaskInfo('success', clone) - self._session.wait_for_task(task).AndReturn(task_info) - - m.ReplayAll() - ret = self._volumeops.clone_backing(name, backing, snapshot, - mox.IgnoreArg(), datastore) - self.assertEqual(ret, clone) - m.UnsetStubs() - m.VerifyAll() - def test_clone_backing_linked(self): """Test _clone_backing with clone type - linked.""" m = self.mox diff --git a/cinder/tests/test_vmware_volumeops.py b/cinder/tests/test_vmware_volumeops.py new file mode 100644 index 000000000..29925c6f6 --- /dev/null +++ b/cinder/tests/test_vmware_volumeops.py @@ -0,0 +1,793 @@ +# Copyright (c) 2014 VMware, Inc. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Test suite for VMware VMDK driver volumeops module. +""" + +import mock + +from cinder import test +from cinder.volume.drivers.vmware import error_util +from cinder.volume.drivers.vmware import vim_util +from cinder.volume.drivers.vmware import volumeops + + +class VolumeOpsTestCase(test.TestCase): + """Unit tests for volumeops module.""" + + MAX_OBJECTS = 100 + + def setUp(self): + super(VolumeOpsTestCase, self).setUp() + self.session = mock.MagicMock() + self.vops = volumeops.VMwareVolumeOps(self.session, self.MAX_OBJECTS) + + def test_split_datastore_path(self): + test1 = '[datastore1] myfolder/mysubfolder/myvm.vmx' + (datastore, folder, file_name) = volumeops.split_datastore_path(test1) + self.assertEqual(datastore, 'datastore1') + self.assertEqual(folder, 'myfolder/mysubfolder/') + self.assertEqual(file_name, 'myvm.vmx') + + test2 = '[datastore2 ] myfolder/myvm.vmdk' + (datastore, folder, file_name) = volumeops.split_datastore_path(test2) + self.assertEqual(datastore, 'datastore2') + self.assertEqual(folder, 'myfolder/') + self.assertEqual(file_name, 'myvm.vmdk') + + test3 = 'myfolder/myvm.vmdk' + self.assertRaises(IndexError, volumeops.split_datastore_path, test3) + + def vm(self, val): + """Create a mock vm in retrieve result format.""" + vm = mock.MagicMock() + prop = mock.Mock(spec=object) + prop.val = val + vm.propSet = [prop] + return vm + + def test_get_backing(self): + name = 'mock-backing' + + # Test no result + self.session.invoke_api.return_value = None + result = self.vops.get_backing(name) + self.assertIsNone(result) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_objects', + self.session.vim, + 'VirtualMachine', + self.MAX_OBJECTS) + + # Test single result + vm = self.vm(name) + vm.obj = mock.sentinel.vm_obj + retrieve_result = mock.Mock(spec=object) + retrieve_result.objects = [vm] + self.session.invoke_api.return_value = retrieve_result + self.vops.cancel_retrieval = mock.Mock(spec=object) + result = self.vops.get_backing(name) + self.assertEqual(mock.sentinel.vm_obj, result) + self.session.invoke_api.assert_called_with(vim_util, 'get_objects', + self.session.vim, + 'VirtualMachine', + self.MAX_OBJECTS) + self.vops.cancel_retrieval.assert_called_once_with(retrieve_result) + + # Test multiple results + retrieve_result2 = mock.Mock(spec=object) + retrieve_result2.objects = [vm('1'), vm('2'), vm('3')] + self.session.invoke_api.return_value = retrieve_result2 + self.vops.continue_retrieval = mock.Mock(spec=object) + self.vops.continue_retrieval.return_value = retrieve_result + result = self.vops.get_backing(name) + self.assertEqual(mock.sentinel.vm_obj, result) + self.session.invoke_api.assert_called_with(vim_util, 'get_objects', + self.session.vim, + 'VirtualMachine', + self.MAX_OBJECTS) + self.vops.continue_retrieval.assert_called_once_with(retrieve_result2) + self.vops.cancel_retrieval.assert_called_with(retrieve_result) + + def test_delete_backing(self): + backing = mock.sentinel.backing + task = mock.sentinel.task + self.session.invoke_api.return_value = task + self.vops.delete_backing(backing) + self.session.invoke_api.assert_called_once_with(self.session.vim, + "Destroy_Task", + backing) + self.session.wait_for_task(task) + + def test_get_host(self): + instance = mock.sentinel.instance + host = mock.sentinel.host + self.session.invoke_api.return_value = host + result = self.vops.get_host(instance) + self.assertEqual(host, result) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_object_property', + self.session.vim, + instance, + 'runtime.host') + + def test_get_hosts(self): + hosts = mock.sentinel.hosts + self.session.invoke_api.return_value = hosts + result = self.vops.get_hosts() + self.assertEqual(hosts, result) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_objects', + self.session.vim, + 'HostSystem', + self.MAX_OBJECTS) + + def test_continue_retrieval(self): + retrieve_result = mock.sentinel.retrieve_result + self.session.invoke_api.return_value = retrieve_result + result = self.vops.continue_retrieval(retrieve_result) + self.assertEqual(retrieve_result, result) + self.session.invoke_api.assert_called_once_with(vim_util, + 'continue_retrieval', + self.session.vim, + retrieve_result) + + def test_cancel_retrieval(self): + retrieve_result = mock.sentinel.retrieve_result + self.session.invoke_api.return_value = retrieve_result + result = self.vops.cancel_retrieval(retrieve_result) + self.assertIsNone(result) + self.session.invoke_api.assert_called_once_with(vim_util, + 'cancel_retrieval', + self.session.vim, + retrieve_result) + + def test_is_usable(self): + mount_info = mock.Mock(spec=object) + mount_info.accessMode = "readWrite" + mount_info.mounted = True + mount_info.accessible = True + datastore = mock.sentinel.datastore + self.assertTrue(self.vops._is_usable(datastore, mount_info)) + + del mount_info.mounted + self.assertTrue(self.vops._is_usable(datastore, mount_info)) + + mount_info.accessMode = "readonly" + self.assertFalse(self.vops._is_usable(datastore, mount_info)) + + mount_info.accessMode = "readWrite" + mount_info.mounted = False + self.assertFalse(self.vops._is_usable(datastore, mount_info)) + + mount_info.mounted = True + mount_info.accessible = False + self.assertFalse(self.vops._is_usable(datastore, mount_info)) + + with mock.patch.object(self.vops, 'get_summary') as get_summary: + del mount_info.accessible + summary = mock.Mock(spec=object) + summary.accessible = True + get_summary.return_value = summary + self.assertTrue(self.vops._is_usable(datastore, mount_info)) + + summary.accessible = False + self.assertFalse(self.vops._is_usable(datastore, mount_info)) + + def _create_host_mounts(self, access_mode, host, set_accessible=True, + is_accessible=True, mounted=True): + """Create host mount value of datastore with single mount info. + + :param access_mode: string specifying the read/write permission + :param set_accessible: specify whether accessible property + should be set + :param is_accessible: boolean specifying whether the datastore + is accessible to host + :param host: managed object reference of the connected + host + :return: list of host mount info + """ + mntInfo = mock.Mock(spec=object) + mntInfo.accessMode = access_mode + if set_accessible: + mntInfo.accessible = is_accessible + else: + del mntInfo.accessible + mntInfo.mounted = mounted + + host_mount = mock.Mock(spec=object) + host_mount.key = host + host_mount.mountInfo = mntInfo + host_mounts = mock.Mock(spec=object) + host_mounts.DatastoreHostMount = [host_mount] + + return host_mounts + + def test_get_connected_hosts(self): + datastore = mock.sentinel.datastore + host = mock.Mock(spec=object) + host.value = mock.sentinel.host + host_mounts = self._create_host_mounts("readWrite", host) + self.session.invoke_api.return_value = host_mounts + + hosts = self.vops.get_connected_hosts(datastore) + self.assertEqual([mock.sentinel.host], hosts) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_object_property', + self.session.vim, + datastore, + 'host') + + def test_is_valid(self): + datastore = mock.sentinel.datastore + host = mock.Mock(spec=object) + host.value = mock.sentinel.host + + def _is_valid(host_mounts, is_valid): + self.session.invoke_api.return_value = host_mounts + result = self.vops._is_valid(datastore, host) + self.assertEqual(is_valid, result) + self.session.invoke_api.assert_called_with(vim_util, + 'get_object_property', + self.session.vim, + datastore, + 'host') + # Test with accessible attr + _is_valid(self._create_host_mounts("readWrite", host), True) + + # Test without accessible attr, and use summary instead + with mock.patch.object(self.vops, 'get_summary') as get_summary: + summary = mock.Mock(spec=object) + summary.accessible = True + get_summary.return_value = summary + _is_valid(self._create_host_mounts("readWrite", host, False), + True) + + # Test negative cases for is_valid + _is_valid(self._create_host_mounts("Inaccessible", host), False) + _is_valid(self._create_host_mounts("readWrite", host, True, False), + False) + _is_valid(self._create_host_mounts("readWrite", host, True, True, + False), False) + with mock.patch.object(self.vops, 'get_summary') as get_summary: + summary = mock.Mock(spec=object) + summary.accessible = False + get_summary.return_value = summary + _is_valid(self._create_host_mounts("readWrite", host, False), + False) + + def test_get_dss_rp(self): + # build out props to be returned by 1st invoke_api call + datastore_prop = mock.Mock(spec=object) + datastore_prop.name = 'datastore' + datastore_prop.val = mock.Mock(spec=object) + datastore_prop.val.ManagedObjectReference = [mock.sentinel.ds1, + mock.sentinel.ds2] + compute_resource_prop = mock.Mock(spec=object) + compute_resource_prop.name = 'parent' + compute_resource_prop.val = mock.sentinel.compute_resource + elem = mock.Mock(spec=object) + elem.propSet = [datastore_prop, compute_resource_prop] + props = [elem] + # build out host_mounts to be returned by 2nd invoke_api call + host = mock.Mock(spec=object) + host.value = mock.sentinel.host + host_mounts = self._create_host_mounts("readWrite", host) + # build out resource_pool to be returned by 3rd invoke_api call + resource_pool = mock.sentinel.resource_pool + # set return values for each call of invoke_api + self.session.invoke_api.side_effect = [props, + host_mounts, + host_mounts, + resource_pool] + # invoke function and verify results + (dss_actual, rp_actual) = self.vops.get_dss_rp(host) + self.assertEqual([mock.sentinel.ds1, mock.sentinel.ds2], dss_actual) + self.assertEqual(resource_pool, rp_actual) + + # invoke function with no valid datastore and verify exception raised + host_mounts = self._create_host_mounts("inaccessible", host) + self.session.invoke_api.side_effect = [props, + host_mounts, + host_mounts, + resource_pool] + self.assertRaises(error_util.VimException, self.vops.get_dss_rp, host) + + def test_get_parent(self): + # Not recursive + child = mock.Mock(spec=object) + child._type = 'Parent' + ret = self.vops._get_parent(child, 'Parent') + self.assertEqual(ret, child) + + # Recursive + parent = mock.Mock(spec=object) + parent._type = 'Parent' + child = mock.Mock(spec=object) + child._type = 'Child' + self.session.invoke_api.return_value = parent + ret = self.vops._get_parent(child, 'Parent') + self.assertEqual(ret, parent) + self.session.invoke_api.assert_called_with(vim_util, + 'get_object_property', + self.session.vim, child, + 'parent') + + def test_get_dc(self): + # set up hierarchy of objects + dc = mock.Mock(spec=object) + dc._type = 'Datacenter' + o1 = mock.Mock(spec=object) + o1._type = 'mockType1' + o1.parent = dc + o2 = mock.Mock(spec=object) + o2._type = 'mockType2' + o2.parent = o1 + + # mock out invoke_api behaviour to fetch parent + def mock_invoke_api(vim_util, method, vim, the_object, arg): + return the_object.parent + + self.session.invoke_api.side_effect = mock_invoke_api + ret = self.vops.get_dc(o2) + self.assertEqual(dc, ret) + + def test_get_vmfolder(self): + self.session.invoke_api.return_value = mock.sentinel.ret + ret = self.vops.get_vmfolder(mock.sentinel.dc) + self.assertEqual(mock.sentinel.ret, ret) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_object_property', + self.session.vim, + mock.sentinel.dc, + 'vmFolder') + + def test_create_folder_not_present(self): + """Test create_folder when child not present.""" + parent_folder = mock.sentinel.parent_folder + child_name = 'child_folder' + prop_val = mock.Mock(spec=object) + prop_val.ManagedObjectReference = [] + child_folder = mock.sentinel.child_folder + self.session.invoke_api.side_effect = [prop_val, child_folder] + ret = self.vops.create_folder(parent_folder, child_name) + self.assertEqual(child_folder, ret) + expected_invoke_api = [mock.call(vim_util, 'get_object_property', + self.session.vim, parent_folder, + 'childEntity'), + mock.call(self.session.vim, 'CreateFolder', + parent_folder, name=child_name)] + self.assertEqual(expected_invoke_api, + self.session.invoke_api.mock_calls) + + def test_create_folder_already_present(self): + """Test create_folder when child already present.""" + parent_folder = mock.sentinel.parent_folder + child_name = 'child_folder' + prop_val = mock.Mock(spec=object) + child_entity_1 = mock.Mock(spec=object) + child_entity_1._type = 'Folder' + child_entity_1_name = 'SomeOtherName' + child_entity_2 = mock.Mock(spec=object) + child_entity_2._type = 'Folder' + child_entity_2_name = child_name + prop_val.ManagedObjectReference = [child_entity_1, child_entity_2] + self.session.invoke_api.side_effect = [prop_val, child_entity_1_name, + child_entity_2_name] + ret = self.vops.create_folder(parent_folder, child_name) + self.assertEqual(child_entity_2, ret) + expected_invoke_api = [mock.call(vim_util, 'get_object_property', + self.session.vim, parent_folder, + 'childEntity'), + mock.call(vim_util, 'get_object_property', + self.session.vim, child_entity_1, + 'name'), + mock.call(vim_util, 'get_object_property', + self.session.vim, child_entity_2, + 'name')] + self.assertEqual(expected_invoke_api, + self.session.invoke_api.mock_calls) + + def test_get_create_spec(self): + factory = self.session.vim.client.factory + factory.create.return_value = mock.Mock(spec=object) + name = mock.sentinel.name + size_kb = 0.5 + disk_type = 'thin' + ds_name = mock.sentinel.ds_name + ret = self.vops._get_create_spec(name, size_kb, disk_type, ds_name) + self.assertEqual(name, ret.name) + self.assertEqual('[%s]' % ds_name, ret.files.vmPathName) + self.assertEqual(1, ret.deviceChange[1].device.capacityInKB) + expected = [mock.call.create('ns0:VirtualLsiLogicController'), + mock.call.create('ns0:VirtualDeviceConfigSpec'), + mock.call.create('ns0:VirtualDisk'), + mock.call.create('ns0:VirtualDiskFlatVer2BackingInfo'), + mock.call.create('ns0:VirtualDeviceConfigSpec'), + mock.call.create('ns0:VirtualMachineFileInfo'), + mock.call.create('ns0:VirtualMachineConfigSpec')] + factory.create.assert_has_calls(expected, any_order=True) + + @mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.' + '_get_create_spec') + def test_create_backing(self, get_create_spec): + create_spec = mock.sentinel.create_spec + get_create_spec.return_value = create_spec + task = mock.sentinel.task + self.session.invoke_api.return_value = task + task_info = mock.Mock(spec=object) + task_info.result = mock.sentinel.result + self.session.wait_for_task.return_value = task_info + name = 'backing_name' + size_kb = mock.sentinel.size_kb + disk_type = mock.sentinel.disk_type + folder = mock.sentinel.folder + resource_pool = mock.sentinel.resource_pool + host = mock.sentinel.host + ds_name = mock.sentinel.ds_name + ret = self.vops.create_backing(name, size_kb, disk_type, folder, + resource_pool, host, ds_name) + self.assertEqual(mock.sentinel.result, ret) + get_create_spec.assert_called_once_with(name, size_kb, disk_type, + ds_name) + self.session.invoke_api.assert_called_once_with(self.session.vim, + 'CreateVM_Task', + folder, + config=create_spec, + pool=resource_pool, + host=host) + self.session.wait_for_task.assert_called_once_with(task) + + def test_get_datastore(self): + backing = mock.sentinel.backing + datastore = mock.Mock(spec=object) + datastore.ManagedObjectReference = [mock.sentinel.ds] + self.session.invoke_api.return_value = datastore + ret = self.vops.get_datastore(backing) + self.assertEqual(mock.sentinel.ds, ret) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_object_property', + self.session.vim, + backing, 'datastore') + + def test_get_summary(self): + datastore = mock.sentinel.datastore + summary = mock.sentinel.summary + self.session.invoke_api.return_value = summary + ret = self.vops.get_summary(datastore) + self.assertEqual(summary, ret) + self.session.invoke_api.assert_called_once_with(vim_util, + 'get_object_property', + self.session.vim, + datastore, + 'summary') + + def test_get_relocate_spec(self): + factory = self.session.vim.client.factory + spec = mock.Mock(spec=object) + factory.create.return_value = spec + 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) + + @mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.' + '_get_relocate_spec') + def test_relocate_backing(self, get_relocate_spec): + spec = mock.sentinel.relocate_spec + get_relocate_spec.return_value = spec + task = mock.sentinel.task + self.session.invoke_api.return_value = task + backing = mock.sentinel.backing + datastore = mock.sentinel.datastore + resource_pool = mock.sentinel.resource_pool + host = mock.sentinel.host + self.vops.relocate_backing(backing, datastore, resource_pool, host) + # Verify calls + disk_move_type = 'moveAllDiskBackingsAndAllowSharing' + get_relocate_spec.assert_called_once_with(datastore, resource_pool, + host, disk_move_type) + self.session.invoke_api.assert_called_once_with(self.session.vim, + 'RelocateVM_Task', + backing, + spec=spec) + self.session.wait_for_task.assert_called_once_with(task) + + def test_move_backing_to_folder(self): + task = mock.sentinel.task + self.session.invoke_api.return_value = task + backing = mock.sentinel.backing + folder = mock.sentinel.folder + self.vops.move_backing_to_folder(backing, folder) + # Verify calls + self.session.invoke_api.assert_called_once_with(self.session.vim, + 'MoveIntoFolder_Task', + folder, + list=[backing]) + self.session.wait_for_task.assert_called_once_with(task) + + def test_create_snapshot_operation(self): + task = mock.sentinel.task + self.session.invoke_api.return_value = task + task_info = mock.Mock(spec=object) + task_info.result = mock.sentinel.result + self.session.wait_for_task.return_value = task_info + backing = mock.sentinel.backing + name = mock.sentinel.name + desc = mock.sentinel.description + quiesce = True + ret = self.vops.create_snapshot(backing, name, desc, quiesce) + self.assertEqual(mock.sentinel.result, ret) + self.session.invoke_api.assert_called_once_with(self.session.vim, + 'CreateSnapshot_Task', + backing, name=name, + description=desc, + memory=False, + quiesce=quiesce) + self.session.wait_for_task.assert_called_once_with(task) + + def test_get_snapshot_from_tree(self): + volops = volumeops.VMwareVolumeOps + name = mock.sentinel.name + # Test snapshot == 'None' + ret = volops._get_snapshot_from_tree(name, None) + self.assertIsNone(ret) + # Test root == snapshot + snapshot = mock.sentinel.snapshot + node = mock.Mock(spec=object) + node.name = name + node.snapshot = snapshot + ret = volops._get_snapshot_from_tree(name, node) + self.assertEqual(ret, snapshot) + # Test root.childSnapshotList == None + root = mock.Mock(spec=object) + root.name = 'root' + del root.childSnapshotList + ret = volops._get_snapshot_from_tree(name, root) + self.assertIsNone(ret) + # Test root.child == snapshot + root.childSnapshotList = [node] + ret = volops._get_snapshot_from_tree(name, root) + self.assertEqual(ret, snapshot) + + def test_get_snapshot(self): + # build out the root snapshot tree + snapshot_name = mock.sentinel.snapshot_name + snapshot = mock.sentinel.snapshot + root = mock.Mock(spec=object) + root.name = 'root' + node = mock.Mock(spec=object) + node.name = snapshot_name + node.snapshot = snapshot + root.childSnapshotList = [node] + # Test rootSnapshotList is not None + snapshot_tree = mock.Mock(spec=object) + snapshot_tree.rootSnapshotList = [root] + self.session.invoke_api.return_value = snapshot_tree + backing = mock.sentinel.backing + ret = self.vops.get_snapshot(backing, snapshot_name) + self.assertEqual(snapshot, ret) + self.session.invoke_api.assert_called_with(vim_util, + 'get_object_property', + self.session.vim, + backing, + 'snapshot') + # Test rootSnapshotList == None + snapshot_tree.rootSnapshotList = None + ret = self.vops.get_snapshot(backing, snapshot_name) + self.assertIsNone(ret) + self.session.invoke_api.assert_called_with(vim_util, + 'get_object_property', + self.session.vim, + backing, + 'snapshot') + + def test_delete_snapshot(self): + backing = mock.sentinel.backing + snapshot_name = mock.sentinel.snapshot_name + # Test snapshot is None + with mock.patch.object(self.vops, 'get_snapshot') as get_snapshot: + get_snapshot.return_value = None + self.vops.delete_snapshot(backing, snapshot_name) + get_snapshot.assert_called_once_with(backing, snapshot_name) + # Test snapshot is not None + snapshot = mock.sentinel.snapshot + task = mock.sentinel.task + invoke_api = self.session.invoke_api + invoke_api.return_value = task + with mock.patch.object(self.vops, 'get_snapshot') as get_snapshot: + get_snapshot.return_value = snapshot + self.vops.delete_snapshot(backing, snapshot_name) + get_snapshot.assert_called_with(backing, snapshot_name) + invoke_api.assert_called_once_with(self.session.vim, + 'RemoveSnapshot_Task', + snapshot, removeChildren=False) + self.session.wait_for_task.assert_called_once_with(task) + + def test_get_folder(self): + folder = mock.sentinel.folder + backing = mock.sentinel.backing + with mock.patch.object(self.vops, '_get_parent') as get_parent: + get_parent.return_value = folder + ret = self.vops._get_folder(backing) + self.assertEqual(folder, ret) + get_parent.assert_called_once_with(backing, 'Folder') + + def test_get_clone_spec(self): + factory = self.session.vim.client.factory + spec = mock.Mock(spec=object) + factory.create.return_value = 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) + 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) + + @mock.patch('cinder.volume.drivers.vmware.volumeops.VMwareVolumeOps.' + '_get_clone_spec') + def test_clone_backing(self, 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] + task_info = mock.Mock(spec=object) + task_info.result = mock.sentinel.new_backing + self.session.wait_for_task.return_value = task_info + clone_spec = mock.sentinel.clone_spec + get_clone_spec.return_value = clone_spec + # Test non-linked clone_backing + name = mock.sentinel.name + backing = mock.Mock(spec=object) + backing._type = 'VirtualMachine' + snapshot = mock.sentinel.snapshot + clone_type = "anything-other-than-linked" + datastore = mock.sentinel.datstore + ret = self.vops.clone_backing(name, backing, snapshot, clone_type, + datastore) + # verify calls + self.assertEqual(mock.sentinel.new_backing, ret) + disk_move_type = 'moveAllDiskBackingsAndDisallowSharing' + get_clone_spec.assert_called_with(datastore, disk_move_type, snapshot) + 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.assertEqual(expected, self.session.invoke_api.mock_calls) + + # Test linked clone_backing + clone_type = volumeops.LINKED_CLONE_TYPE + 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) + 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', + self.session.vim, backing, 'parent'), + mock.call(self.session.vim, 'CloneVM_Task', backing, + folder=folder, name=name, spec=clone_spec)] + self.assertEqual(expected, self.session.invoke_api.mock_calls) + + def test_delete_file(self): + file_mgr = mock.sentinel.file_manager + self.session.vim.service_content.fileManager = file_mgr + task = mock.sentinel.task + invoke_api = self.session.invoke_api + invoke_api.return_value = task + # Test delete file + file_path = mock.sentinel.file_path + datacenter = mock.sentinel.datacenter + self.vops.delete_file(file_path, datacenter) + # verify calls + invoke_api.assert_called_once_with(self.session.vim, + 'DeleteDatastoreFile_Task', + file_mgr, + name=file_path, + datacenter=datacenter) + self.session.wait_for_task.assert_called_once_with(task) + + def test_get_path_name(self): + path = mock.Mock(spec=object) + path_name = mock.sentinel.vm_path_name + path.vmPathName = path_name + invoke_api = self.session.invoke_api + invoke_api.return_value = path + backing = mock.sentinel.backing + ret = self.vops.get_path_name(backing) + self.assertEqual(path_name, ret) + invoke_api.assert_called_once_with(vim_util, 'get_object_property', + self.session.vim, backing, + 'config.files') + + def test_get_entity_name(self): + entity_name = mock.sentinel.entity_name + invoke_api = self.session.invoke_api + invoke_api.return_value = entity_name + entity = mock.sentinel.entity + ret = self.vops.get_entity_name(entity) + self.assertEqual(entity_name, ret) + invoke_api.assert_called_once_with(vim_util, 'get_object_property', + self.session.vim, entity, 'name') + + def test_get_vmdk_path(self): + # Setup hardware_devices for test + device = mock.Mock() + device.__class__.__name__ = 'VirtualDisk' + backing = mock.Mock() + backing.__class__.__name__ = 'VirtualDiskFlatVer2BackingInfo' + backing.fileName = mock.sentinel.vmdk_path + device.backing = backing + invoke_api = self.session.invoke_api + invoke_api.return_value = [device] + # Test get_vmdk_path + ret = self.vops.get_vmdk_path(backing) + self.assertEqual(mock.sentinel.vmdk_path, ret) + invoke_api.assert_called_once_with(vim_util, 'get_object_property', + self.session.vim, backing, + 'config.hardware.device') + + def test_copy_vmdk_file(self): + task = mock.sentinel.task + invoke_api = self.session.invoke_api + invoke_api.return_value = task + disk_mgr = self.session.vim.service_content.virtualDiskManager + dc_ref = self.session.dc_ref + src_vmdk_file_path = self.session.src + dest_vmdk_file_path = self.session.dest + self.vops.copy_vmdk_file(dc_ref, src_vmdk_file_path, + dest_vmdk_file_path) + invoke_api.assert_called_once_with(self.session.vim, + 'CopyVirtualDisk_Task', + disk_mgr, + sourceName=src_vmdk_file_path, + sourceDatacenter=dc_ref, + destName=dest_vmdk_file_path, + destDatacenter=dc_ref, + force=True) + self.session.wait_for_task.assert_called_once_with(task) + + def test_delete_vmdk_file(self): + task = mock.sentinel.task + invoke_api = self.session.invoke_api + invoke_api.return_value = task + disk_mgr = self.session.vim.service_content.virtualDiskManager + dc_ref = self.session.dc_ref + vmdk_file_path = self.session.vmdk_file + self.vops.delete_vmdk_file(vmdk_file_path, dc_ref) + invoke_api.assert_called_once_with(self.session.vim, + 'DeleteVirtualDisk_Task', + disk_mgr, + name=vmdk_file_path, + datacenter=dc_ref) + self.session.wait_for_task.assert_called_once_with(task) -- 2.45.2