# under the License.
"""
-Test suite for VMware VMDK driver.
+Test suite for VMware vCenter VMDK driver.
"""
from distutils import version as ver
# TODO(vbala) Split test methods handling multiple cases into multiple methods,
# each handling a specific case.
-class VMwareEsxVmdkDriverTestCase(test.TestCase):
- """Test class for VMwareEsxVmdkDriver."""
+class VMwareVcVmdkDriverTestCase(test.TestCase):
+ """Test class for VMwareVcVmdkDriver."""
IP = 'localhost'
PORT = 443
MAX_OBJECTS = 100
TMP_DIR = "/vmware-tmp"
CA_FILE = "/etc/ssl/rui-ca-cert.pem"
- VMDK_DRIVER = vmdk.VMwareEsxVmdkDriver
+ VMDK_DRIVER = vmdk.VMwareVcVmdkDriver
CLUSTERS = ["cls-1", "cls-2"]
+ DEFAULT_VC_VERSION = '5.5'
def setUp(self):
- super(VMwareEsxVmdkDriverTestCase, self).setUp()
+ super(VMwareVcVmdkDriverTestCase, self).setUp()
self._config = mox.MockObject(configuration.Configuration)
self._config.append_config_values(mox.IgnoreArg())
self._config.vmware_host_ip = self.IP
self._config.vmware_ca_file = self.CA_FILE
self._config.vmware_insecure = False
self._config.vmware_cluster_name = self.CLUSTERS
+ self._config.vmware_host_version = self.DEFAULT_VC_VERSION
self._db = mock.Mock()
- self._driver = vmdk.VMwareEsxVmdkDriver(configuration=self._config,
- db=self._db)
+ self._driver = vmdk.VMwareVcVmdkDriver(configuration=self._config,
+ db=self._db)
api_retry_count = self._config.vmware_api_retry_count
task_poll_interval = self._config.vmware_task_poll_interval,
self._session = api.VMwareAPISession(self.IP, self.USERNAME,
self.MAX_OBJECTS)
self._vim = FakeVim()
- def test_do_setup(self):
- """Test do_setup."""
- m = self.mox
- m.StubOutWithMock(self._driver.__class__, 'session')
- self._driver.session = self._session
- m.ReplayAll()
- self._driver.do_setup(mox.IgnoreArg())
- m.UnsetStubs()
- m.VerifyAll()
-
def test_check_for_setup_error(self):
"""Test check_for_setup_error."""
self._driver.check_for_setup_error()
self._driver.terminate_connection(mox.IgnoreArg(), mox.IgnoreArg(),
force=mox.IgnoreArg())
- def test_get_volume_group_folder(self):
- """Test _get_volume_group_folder."""
- m = self.mox
- m.StubOutWithMock(self._driver.__class__, 'volumeops')
- self._driver.volumeops = self._volumeops
- datacenter = FakeMor('Datacenter', 'my_dc')
- m.StubOutWithMock(self._volumeops, 'get_vmfolder')
- self._volumeops.get_vmfolder(datacenter)
-
- m.ReplayAll()
- self._driver._get_volume_group_folder(datacenter)
- m.UnsetStubs()
- m.VerifyAll()
-
@mock.patch('cinder.volume.volume_types.get_volume_type_extra_specs')
def test_get_disk_type(self, get_volume_type_extra_specs):
"""Test _get_disk_type."""
# Test with no volume type.
volume = {'volume_type_id': None}
self.assertEqual(vmdk.THIN_VMDK_TYPE,
- vmdk.VMwareEsxVmdkDriver._get_disk_type(volume))
+ vmdk.VMwareVcVmdkDriver._get_disk_type(volume))
# Test with valid vmdk_type.
volume_type_id = mock.sentinel.volume_type_id
get_volume_type_extra_specs.return_value = vmdk.THICK_VMDK_TYPE
self.assertEqual(vmdk.THICK_VMDK_TYPE,
- vmdk.VMwareEsxVmdkDriver._get_disk_type(volume))
+ vmdk.VMwareVcVmdkDriver._get_disk_type(volume))
get_volume_type_extra_specs.assert_called_once_with(volume_type_id,
'vmware:vmdk_type')
# Test with invalid vmdk_type.
get_volume_type_extra_specs.return_value = 'sparse'
self.assertRaises(vmdk_exceptions.InvalidDiskTypeException,
- vmdk.VMwareEsxVmdkDriver._get_disk_type,
+ vmdk.VMwareVcVmdkDriver._get_disk_type,
volume)
def test_create_snapshot_without_backing(self):
- """Test vmdk.create_snapshot without backing."""
+ """Test create_snapshot without backing."""
m = self.mox
m.StubOutWithMock(self._driver.__class__, 'volumeops')
self._driver.volumeops = self._volumeops
m.VerifyAll()
def test_create_snapshot_with_backing(self):
- """Test vmdk.create_snapshot with backing."""
+ """Test create_snapshot with backing."""
m = self.mox
m.StubOutWithMock(self._driver.__class__, 'volumeops')
self._driver.volumeops = self._volumeops
m.VerifyAll()
def test_create_snapshot_when_attached(self):
- """Test vmdk.create_snapshot when volume is attached."""
+ """Test create_snapshot when volume is attached."""
snapshot = FakeObject()
snapshot['volume'] = FakeObject()
snapshot['volume']['status'] = 'in-use'
self.assertRaises(cinder_exceptions.InvalidVolume,
self._driver.delete_snapshot, snapshot)
- @mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver.'
- 'volumeops', new_callable=mock.PropertyMock)
- def test_create_cloned_volume_without_backing(self, mock_vops):
- """Test create_cloned_volume without a backing."""
- mock_vops = mock_vops.return_value
- driver = self._driver
- volume = {'name': 'mock_vol'}
- src_vref = {'name': 'src_snapshot_name'}
- driver._verify_volume_creation = mock.MagicMock()
- mock_vops.get_backing.return_value = None
-
- # invoke the create_volume_from_snapshot api
- driver.create_cloned_volume(volume, src_vref)
-
- # verify calls
- driver._verify_volume_creation.assert_called_once_with(volume)
- mock_vops.get_backing.assert_called_once_with('src_snapshot_name')
-
- @mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver.'
- 'volumeops', new_callable=mock.PropertyMock)
- def test_create_cloned_volume_with_backing(self, mock_vops):
- """Test create_cloned_volume with a backing."""
- mock_vops = mock_vops.return_value
- driver = self._driver
- volume = mock.sentinel.volume
- fake_size = 1
- src_vref = {'name': 'src_snapshot_name', 'size': fake_size}
- backing = mock.sentinel.backing
- driver._verify_volume_creation = mock.MagicMock()
- mock_vops.get_backing.return_value = backing
- src_vmdk = "[datastore] src_vm/src_vm.vmdk"
- mock_vops.get_vmdk_path.return_value = src_vmdk
- driver._create_backing_by_copying = mock.MagicMock()
-
- # invoke the create_volume_from_snapshot api
- driver.create_cloned_volume(volume, src_vref)
-
- # verify calls
- driver._verify_volume_creation.assert_called_once_with(volume)
- mock_vops.get_backing.assert_called_once_with('src_snapshot_name')
- mock_vops.get_vmdk_path.assert_called_once_with(backing)
- driver._create_backing_by_copying.assert_called_once_with(volume,
- src_vmdk,
- fake_size)
-
- @mock.patch.object(VMDK_DRIVER, '_extend_volumeops_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_create_backing')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_create_backing_by_copying(self, volumeops, create_backing,
- _extend_virtual_disk):
- self._test_create_backing_by_copying(volumeops, create_backing,
- _extend_virtual_disk)
-
- def _test_create_backing_by_copying(self, volumeops, create_backing,
- _extend_virtual_disk):
- """Test _create_backing_by_copying."""
- fake_volume = {'size': 2, 'name': 'fake_volume-0000000000001'}
- fake_size = 1
- fake_src_vmdk_path = "[datastore] src_vm/src_vm.vmdk"
- fake_backing = mock.sentinel.backing
- fake_vmdk_path = mock.sentinel.path
- # "[datastore] dest_vm/dest_vm.vmdk"
- fake_dc = mock.sentinel.datacenter
-
- create_backing.return_value = fake_backing
- volumeops.get_vmdk_path.return_value = fake_vmdk_path
- volumeops.get_dc.return_value = fake_dc
-
- # Test with fake_volume['size'] greater than fake_size
- self._driver._create_backing_by_copying(fake_volume,
- fake_src_vmdk_path,
- fake_size)
- create_backing.assert_called_once_with(fake_volume)
- volumeops.get_vmdk_path.assert_called_once_with(fake_backing)
- volumeops.get_dc.assert_called_once_with(fake_backing)
- volumeops.delete_vmdk_file.assert_called_once_with(fake_vmdk_path,
- fake_dc)
- volumeops.copy_vmdk_file.assert_called_once_with(fake_dc,
- fake_src_vmdk_path,
- fake_vmdk_path)
- _extend_virtual_disk.assert_called_once_with(fake_volume['size'],
- fake_vmdk_path,
- fake_dc)
-
- # Reset all the mocks and test with fake_volume['size']
- # not greater than fake_size
- _extend_virtual_disk.reset_mock()
- fake_size = 2
- self._driver._create_backing_by_copying(fake_volume,
- fake_src_vmdk_path,
- fake_size)
- self.assertFalse(_extend_virtual_disk.called)
-
- @mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver.'
- 'volumeops', new_callable=mock.PropertyMock)
- def test_create_volume_from_snapshot_without_backing(self, mock_vops):
- """Test create_volume_from_snapshot without a backing."""
- mock_vops = mock_vops.return_value
- driver = self._driver
- volume = {'name': 'mock_vol'}
- snapshot = {'volume_name': 'mock_vol', 'name': 'mock_snap'}
- driver._verify_volume_creation = mock.MagicMock()
- mock_vops.get_backing.return_value = None
-
- # invoke the create_volume_from_snapshot api
- driver.create_volume_from_snapshot(volume, snapshot)
-
- # verify calls
- driver._verify_volume_creation.assert_called_once_with(volume)
- mock_vops.get_backing.assert_called_once_with('mock_vol')
-
- @mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver.'
- 'volumeops', new_callable=mock.PropertyMock)
- def test_create_volume_from_snap_without_backing_snap(self, mock_vops):
- """Test create_volume_from_snapshot without a backing snapshot."""
- mock_vops = mock_vops.return_value
- driver = self._driver
- volume = {'volume_type_id': None, 'name': 'mock_vol'}
- snapshot = {'volume_name': 'mock_vol', 'name': 'mock_snap'}
- backing = mock.sentinel.backing
- driver._verify_volume_creation = mock.MagicMock()
- mock_vops.get_backing.return_value = backing
- mock_vops.get_snapshot.return_value = None
-
- # invoke the create_volume_from_snapshot api
- driver.create_volume_from_snapshot(volume, snapshot)
-
- # verify calls
- driver._verify_volume_creation.assert_called_once_with(volume)
- mock_vops.get_backing.assert_called_once_with('mock_vol')
- mock_vops.get_snapshot.assert_called_once_with(backing,
- 'mock_snap')
-
- @mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver.'
- 'volumeops', new_callable=mock.PropertyMock)
- def test_create_volume_from_snapshot(self, mock_vops):
- """Test create_volume_from_snapshot."""
- mock_vops = mock_vops.return_value
- driver = self._driver
- volume = {'volume_type_id': None, 'name': 'mock_vol'}
- snapshot = {'volume_name': 'mock_vol', 'name': 'mock_snap',
- 'volume_size': 1}
- fake_size = snapshot['volume_size']
- backing = mock.sentinel.backing
- snap_moref = mock.sentinel.snap_moref
- driver._verify_volume_creation = mock.MagicMock()
- mock_vops.get_backing.return_value = backing
- mock_vops.get_snapshot.return_value = snap_moref
- src_vmdk = "[datastore] src_vm/src_vm-001.vmdk"
- mock_vops.get_vmdk_path.return_value = src_vmdk
- driver._create_backing_by_copying = mock.MagicMock()
-
- # invoke the create_volume_from_snapshot api
- driver.create_volume_from_snapshot(volume, snapshot)
-
- # verify calls
- driver._verify_volume_creation.assert_called_once_with(volume)
- mock_vops.get_backing.assert_called_once_with('mock_vol')
- mock_vops.get_snapshot.assert_called_once_with(backing,
- 'mock_snap')
- mock_vops.get_vmdk_path.assert_called_once_with(snap_moref)
- driver._create_backing_by_copying.assert_called_once_with(volume,
- src_vmdk,
- fake_size)
-
@mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
@mock.patch.object(VMDK_DRIVER, '_extend_vmdk_virtual_disk')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
'_create_virtual_disk_from_preallocated_image')
@mock.patch.object(VMDK_DRIVER, '_create_virtual_disk_from_sparse_image')
@mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
+ 'cinder.volume.drivers.vmware.vmdk.VMwareVcVmdkDriver._get_disk_type')
@mock.patch.object(VMDK_DRIVER, '_get_ds_name_folder_path')
@mock.patch.object(VMDK_DRIVER, '_create_backing')
def test_copy_image_to_volume_non_stream_optimized(
@mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
@mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
+ 'cinder.volume.drivers.vmware.vmdk.VMwareVcVmdkDriver._get_disk_type')
@mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
@mock.patch.object(VMDK_DRIVER,
'_create_backing_from_stream_optimized_file')
context, name, volume, tmp_file_path, file_size_bytes)
delete_temp_backing.assert_called_once_with(backing)
-
-class VMwareVcVmdkDriverTestCase(VMwareEsxVmdkDriverTestCase):
- """Test class for VMwareVcVmdkDriver."""
- VMDK_DRIVER = vmdk.VMwareVcVmdkDriver
-
- DEFAULT_VC_VERSION = '5.5'
-
- def setUp(self):
- super(VMwareVcVmdkDriverTestCase, self).setUp()
- self._config.vmware_host_version = self.DEFAULT_VC_VERSION
- self._driver = vmdk.VMwareVcVmdkDriver(configuration=self._config,
- db=self._db)
-
@mock.patch('cinder.volume.drivers.vmware.vmdk.VMwareVcVmdkDriver.'
'session', new_callable=mock.PropertyMock)
def test_get_vc_version(self, session):
self.assertEqual(mock.sentinel.cluster_refs, self._driver._clusters)
vops.get_cluster_refs.assert_called_once_with(self.CLUSTERS)
- @mock.patch.object(VMDK_DRIVER, '_extend_volumeops_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_create_backing')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_create_backing_by_copying(self, volumeops, create_backing,
- extend_virtual_disk):
- self._test_create_backing_by_copying(volumeops, create_backing,
- extend_virtual_disk)
-
@mock.patch.object(VMDK_DRIVER, '_get_storage_profile')
@mock.patch.object(VMDK_DRIVER, 'ds_sel')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
profile = self._driver._get_storage_profile(volume)
self.assertIsNone(profile)
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_extend_vmdk_virtual_disk(self, volume_ops):
- """Test vmdk._extend_vmdk_virtual_disk."""
- self._test_extend_vmdk_virtual_disk(volume_ops)
-
- @mock.patch.object(VMDK_DRIVER, '_extend_vmdk_virtual_disk')
- @mock.patch('oslo_utils.uuidutils.generate_uuid')
- @mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch.object(VMDK_DRIVER,
- '_create_virtual_disk_from_preallocated_image')
- @mock.patch.object(VMDK_DRIVER, '_create_virtual_disk_from_sparse_image')
- @mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
- @mock.patch.object(VMDK_DRIVER, '_get_ds_name_folder_path')
- @mock.patch.object(VMDK_DRIVER, '_create_backing')
- def test_copy_image_to_volume_non_stream_optimized(
- self, create_backing, get_ds_name_folder_path, get_disk_type,
- create_disk_from_sparse_image, create_disk_from_preallocated_image,
- vops, select_ds_for_volume, generate_uuid, extend_disk):
- self._test_copy_image_to_volume_non_stream_optimized(
- create_backing,
- get_ds_name_folder_path,
- get_disk_type,
- create_disk_from_sparse_image,
- create_disk_from_preallocated_image,
- vops,
- select_ds_for_volume,
- generate_uuid,
- extend_disk)
-
def _test_copy_image(self, download_flat_image, session, vops,
expected_cacerts=False):
# dowload_flat_image should be called with cacerts=False.
self._test_copy_image(download_flat_image, session, vops)
- @mock.patch.object(VMDK_DRIVER, '_copy_temp_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_get_temp_image_folder')
- @mock.patch(
- 'cinder.volume.drivers.vmware.volumeops.FlatExtentVirtualDiskPath')
- @mock.patch.object(VMDK_DRIVER, '_copy_image')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_create_virtual_disk_from_preallocated_image(
- self, vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk):
- self._test_create_virtual_disk_from_preallocated_image(
- vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk)
-
- @mock.patch.object(VMDK_DRIVER, '_copy_temp_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_get_temp_image_folder')
- @mock.patch(
- 'cinder.volume.drivers.vmware.volumeops.FlatExtentVirtualDiskPath')
- @mock.patch.object(VMDK_DRIVER, '_copy_image')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_create_virtual_disk_from_preallocated_image_with_no_disk_copy(
- self, vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk):
- self._test_create_virtual_disk_from_preallocated_image_with_no_copy(
- vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk)
-
- @mock.patch.object(VMDK_DRIVER, '_copy_temp_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_get_temp_image_folder')
- @mock.patch(
- 'cinder.volume.drivers.vmware.volumeops.FlatExtentVirtualDiskPath')
- @mock.patch.object(VMDK_DRIVER, '_copy_image')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_create_virtual_disk_from_preallocated_image_with_copy_error(
- self, vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk):
- self._test_create_virtual_disk_from_preallocated_image_with_copy_error(
- vops, copy_image, flat_extent_path, get_temp_image_folder,
- copy_temp_virtual_disk)
-
- @mock.patch(
- 'cinder.volume.drivers.vmware.volumeops.'
- 'MonolithicSparseVirtualDiskPath')
- @mock.patch(
- 'cinder.volume.drivers.vmware.volumeops.FlatExtentVirtualDiskPath')
- @mock.patch.object(VMDK_DRIVER, '_copy_temp_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_copy_image')
- def test_create_virtual_disk_from_sparse_image(
- self, copy_image, copy_temp_virtual_disk, flat_extent_path,
- sparse_path):
- self._test_create_virtual_disk_from_sparse_image(
- copy_image, copy_temp_virtual_disk, flat_extent_path, sparse_path)
-
- @mock.patch.object(image_transfer, 'download_stream_optimized_image')
- @mock.patch.object(VMDK_DRIVER, '_extend_vmdk_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
- @mock.patch.object(VMDK_DRIVER, '_get_storage_profile_id')
- @mock.patch.object(VMDK_DRIVER, 'session')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_copy_image_to_volume_stream_optimized(self, volumeops,
- session,
- get_profile_id,
- _select_ds_for_volume,
- _extend_virtual_disk,
- download_image):
- """Test copy_image_to_volume.
-
- Test with an acceptable vmdk disk format and streamOptimized disk type.
- """
- self._test_copy_image_to_volume_stream_optimized(volumeops,
- session,
- get_profile_id,
- _select_ds_for_volume,
- _extend_virtual_disk,
- download_image)
-
def test_copy_image_to_volume_with_ova_container(self):
image_service = mock.Mock(glance.GlanceImageService)
image_size = 2 * units.Gi
self._driver.copy_image_to_volume, context, volume, image_service,
image_id)
- @mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
- @mock.patch('oslo_utils.uuidutils.generate_uuid')
- @mock.patch.object(VMDK_DRIVER, '_get_volume_group_folder')
- @mock.patch('cinder.volume.volume_types.get_volume_type_extra_specs')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch.object(VMDK_DRIVER, 'ds_sel')
- def test_retype(self, ds_sel, vops, get_volume_type_extra_specs,
- get_volume_group_folder, generate_uuid,
- delete_temp_backing):
- self._test_retype(ds_sel, vops, get_volume_type_extra_specs,
- get_volume_group_folder, generate_uuid,
- delete_temp_backing)
-
- @mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
- @mock.patch.object(VMDK_DRIVER, '_extend_vmdk_virtual_disk')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_extend_volume(self, volume_ops, _extend_virtual_disk,
- _select_ds_for_volume):
- """Test extend_volume."""
- self._test_extend_volume(volume_ops, _extend_virtual_disk,
- _select_ds_for_volume)
-
- @mock.patch.object(image_transfer, 'copy_stream_optimized_disk')
- @mock.patch('cinder.volume.drivers.vmware.vmdk.open', create=True)
- @mock.patch.object(VMDK_DRIVER, '_temporary_file')
- @mock.patch('oslo_utils.uuidutils.generate_uuid')
- @mock.patch.object(VMDK_DRIVER, '_create_backing')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch.object(VMDK_DRIVER, 'session')
- def test_backup_volume(self, session, vops, create_backing, generate_uuid,
- temporary_file, file_open, copy_disk):
- self._test_backup_volume(session, vops, create_backing, generate_uuid,
- temporary_file, file_open, copy_disk)
-
- @mock.patch.object(VMDK_DRIVER, 'extend_volume')
- @mock.patch.object(VMDK_DRIVER, '_restore_backing')
- @mock.patch('cinder.volume.drivers.vmware.vmdk.open', create=True)
- @mock.patch.object(VMDK_DRIVER, '_temporary_file')
- @mock.patch('oslo_utils.uuidutils.generate_uuid')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- def test_restore_backup(self, vops, generate_uuid, temporary_file,
- file_open, restore_backing, extend_volume):
- self._test_restore_backup(vops, generate_uuid, temporary_file,
- file_open, restore_backing, extend_volume)
-
- @mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
- @mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
- @mock.patch.object(VMDK_DRIVER,
- '_create_backing_from_stream_optimized_file')
- @mock.patch('oslo_utils.uuidutils.generate_uuid')
- def test_restore_backing(
- self, generate_uuid, create_backing, select_ds, get_disk_type,
- vops, delete_temp_backing):
- self._test_restore_backing(
- generate_uuid, create_backing, select_ds, get_disk_type, vops,
- delete_temp_backing)
-
- @mock.patch.object(VMDK_DRIVER, '_delete_temp_backing')
- @mock.patch.object(image_transfer, 'download_stream_optimized_data')
- @mock.patch('cinder.volume.drivers.vmware.vmdk.open', create=True)
- @mock.patch.object(VMDK_DRIVER, 'volumeops')
- @mock.patch(
- 'cinder.volume.drivers.vmware.vmdk.VMwareEsxVmdkDriver._get_disk_type')
- @mock.patch.object(VMDK_DRIVER, '_get_storage_profile_id')
- @mock.patch.object(VMDK_DRIVER, 'session')
- @mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
- def test_create_backing_from_stream_optimized_file(
- self, select_ds, session, get_storage_profile_id, get_disk_type,
- vops, file_open, download_data, delete_temp_backing):
- self._test_create_backing_from_stream_optimized_file(
- select_ds, session, get_storage_profile_id, get_disk_type, vops,
- file_open, download_data, delete_temp_backing)
-
@mock.patch.object(VMDK_DRIVER, '_select_ds_for_volume')
@mock.patch.object(VMDK_DRIVER, 'volumeops')
def test_create_backing_with_params(self, vops, select_ds_for_volume):
# under the License.
"""
-Volume driver for VMware vCenter/ESX managed datastores.
+Volume driver for VMware vCenter managed datastores.
The volumes created by this driver are backed by VMDK (Virtual Machine
Disk) files stored in datastores. For ease of managing the VMDKs, the
vmdk_opts = [
cfg.StrOpt('vmware_host_ip',
default=None,
- help='IP address for connecting to VMware ESX/vCenter server.'),
+ help='IP address for connecting to VMware vCenter server.'),
cfg.StrOpt('vmware_host_username',
default=None,
- help='Username for authenticating with VMware ESX/vCenter '
+ help='Username for authenticating with VMware vCenter '
'server.'),
cfg.StrOpt('vmware_host_password',
default=None,
- help='Password for authenticating with VMware ESX/vCenter '
+ help='Password for authenticating with VMware vCenter '
'server.',
secret=True),
cfg.StrOpt('vmware_wsdl_location',
'to default location for bug work-arounds.'),
cfg.IntOpt('vmware_api_retry_count',
default=10,
- help='Number of times VMware ESX/vCenter server API must be '
+ help='Number of times VMware vCenter server API must be '
'retried upon connection related issues.'),
cfg.FloatOpt('vmware_task_poll_interval',
default=0.5,
help='The interval (in seconds) for polling remote tasks '
- 'invoked on VMware ESX/vCenter server.'),
+ 'invoked on VMware vCenter server.'),
cfg.StrOpt('vmware_volume_folder',
default='Volumes',
help='Name of the vCenter inventory folder that will '
extra_spec_disk_type)
-class VMwareEsxVmdkDriver(driver.VolumeDriver):
- """Manage volumes on VMware ESX server."""
+class VMwareVcVmdkDriver(driver.VolumeDriver):
+ """Manage volumes on VMware vCenter server."""
# 1.0 - initial version of driver
# 1.1.0 - selection of datastore based on number of host mounts
# 1.5.0 - restrict volume placement to specific vCenter clusters
VERSION = '1.5.0'
- def _do_deprecation_warning(self):
- LOG.warning(_LW('The VMware ESX VMDK driver is now deprecated '
- 'and will be removed in the Juno release. The VMware '
- 'vCenter VMDK driver will remain and continue to be '
- 'supported.'))
+ # Minimum supported vCenter version.
+ MIN_SUPPORTED_VC_VERSION = dist_version.LooseVersion('5.1')
- def __init__(self, *args, **kwargs):
- super(VMwareEsxVmdkDriver, self).__init__(*args, **kwargs)
+ # PBM is enabled only for vCenter versions 5.5 and above
+ PBM_ENABLED_VC_VERSION = dist_version.LooseVersion('5.5')
- self._do_deprecation_warning()
+ def __init__(self, *args, **kwargs):
+ super(VMwareVcVmdkDriver, self).__init__(*args, **kwargs)
self.configuration.append_config_values(vmdk_opts)
self._session = None
self._stats = None
self._volumeops = None
- # No storage policy based placement possible when connecting
- # directly to ESX
self._storage_policy_enabled = False
self._ds_sel = None
self._clusters = None
- @property
- def session(self):
- if not self._session:
- ip = self.configuration.vmware_host_ip
- username = self.configuration.vmware_host_username
- password = self.configuration.vmware_host_password
- api_retry_count = self.configuration.vmware_api_retry_count
- task_poll_interval = self.configuration.vmware_task_poll_interval
- wsdl_loc = self.configuration.safe_get('vmware_wsdl_location')
- self._session = api.VMwareAPISession(ip, username,
- password, api_retry_count,
- task_poll_interval,
- wsdl_loc=wsdl_loc)
- return self._session
-
@property
def volumeops(self):
if not self._volumeops:
self.session)
return self._ds_sel
- def do_setup(self, context):
- """Perform validations and establish connection to server.
-
- :param context: Context information
- """
-
+ def _validate_params(self):
# Throw error if required parameters are not set.
required_params = ['vmware_host_ip',
'vmware_host_username',
if not getattr(self.configuration, param, None):
raise exception.InvalidInput(_("%s not set.") % param)
- # Create the session object for the first time for ESX driver
- driver = self.__class__.__name__
- if driver == 'VMwareEsxVmdkDriver':
- max_objects = self.configuration.vmware_max_objects_retrieval
- self._volumeops = volumeops.VMwareVolumeOps(self.session,
- max_objects)
- LOG.info(_LI("Successfully setup driver: %(driver)s for "
- "server: %(ip)s."),
- {'driver': driver,
- 'ip': self.configuration.vmware_host_ip})
-
def check_for_setup_error(self):
pass
"""
self._delete_volume(volume)
- def _get_volume_group_folder(self, datacenter):
- """Return vmFolder of datacenter as we cannot create folder in ESX.
-
- :param datacenter: Reference to the datacenter
- :return: vmFolder reference of the datacenter
- """
- return self.volumeops.get_vmfolder(datacenter)
-
def _get_extra_spec_storage_profile(self, type_id):
"""Get storage profile name in the given volume type's extra spec.
:param volume: Volume object
:return: Disk type
"""
- return VMwareEsxVmdkDriver._get_extra_spec_disk_type(
+ return VMwareVcVmdkDriver._get_extra_spec_disk_type(
volume['volume_type_id'])
def _get_storage_profile_id(self, volume):
extra_config=extra_config)
# create a backing with single disk
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
size_kb = volume['size'] * units.Mi
adapter_type = create_params.get(CREATE_PARAM_ADAPTER_TYPE,
'lsiLogic')
self.volumeops.update_backing_disk_uuid(backing, volume['id'])
return backing
- def _relocate_backing(self, volume, backing, host):
- pass
-
def _get_hosts(self, clusters):
hosts = []
if clusters:
"""
self._delete_snapshot(snapshot)
- def _create_backing_by_copying(self, volume, src_vmdk_path,
- src_size_in_gb):
- """Create volume backing.
-
- Creates a backing for the input volume and replaces its VMDK file
- with the input VMDK file copy.
-
- :param volume: New Volume object
- :param src_vmdk_path: VMDK file path of the source volume backing
- :param src_size_in_gb: The size of the original volume to be cloned
- in GB. The size of the target volume is saved in volume['size'].
- This parameter is used to check if the size specified by the user is
- greater than the original size. If so, the target volume should extend
- its size.
- """
-
- # Create a backing
- backing = self._create_backing(volume)
- dest_vmdk_path = self.volumeops.get_vmdk_path(backing)
- datacenter = self.volumeops.get_dc(backing)
- # Deleting the current VMDK file
- self.volumeops.delete_vmdk_file(dest_vmdk_path, datacenter)
- # Copying the source VMDK file
- self.volumeops.copy_vmdk_file(datacenter, src_vmdk_path,
- dest_vmdk_path)
- # If the target volume has a larger size than the source
- # volume/snapshot, we need to resize/extend the size of the
- # vmdk virtual disk to the value specified by the user.
- if volume['size'] > src_size_in_gb:
- self._extend_volumeops_virtual_disk(volume['size'], dest_vmdk_path,
- datacenter)
- LOG.info(_LI("Successfully cloned new backing: %(back)s from "
- "source VMDK file: %(vmdk)s."),
- {'back': backing, 'vmdk': src_vmdk_path})
-
- def _create_cloned_volume(self, volume, src_vref):
- """Creates volume clone.
-
- If source volume's backing does not exist, then pass.
- Creates a backing and replaces its VMDK file with a copy of the
- source backing's VMDK file.
-
- :param volume: New Volume object
- :param src_vref: Volume object that must be cloned
- """
- self._verify_volume_creation(volume)
- backing = self.volumeops.get_backing(src_vref['name'])
- if not backing:
- LOG.info(_LI("There is no backing for the source volume: "
- "%(svol)s. Not creating any backing for the "
- "volume: %(vol)s."),
- {'svol': src_vref['name'],
- 'vol': volume['name']})
- return
- src_vmdk_path = self.volumeops.get_vmdk_path(backing)
- self._create_backing_by_copying(volume, src_vmdk_path,
- src_vref['size'])
-
- def create_cloned_volume(self, volume, src_vref):
- """Creates volume clone.
-
- :param volume: New Volume object
- :param src_vref: Volume object that must be cloned
- """
- self._create_cloned_volume(volume, src_vref)
-
- def _create_volume_from_snapshot(self, volume, snapshot):
- """Creates a volume from a snapshot.
-
- If the snapshot does not exist or source volume's backing does not
- exist, then pass.
- Else creates clone of source volume backing by copying its VMDK file.
-
- :param volume: Volume object
- :param snapshot: Snapshot object
- """
- self._verify_volume_creation(volume)
- backing = self.volumeops.get_backing(snapshot['volume_name'])
- if not backing:
- LOG.info(_LI("There is no backing for the source snapshot: "
- "%(snap)s. Not creating any backing for the "
- "volume: %(vol)s."),
- {'snap': snapshot['name'],
- 'vol': volume['name']})
- return
- snapshot_moref = self.volumeops.get_snapshot(backing,
- snapshot['name'])
- if not snapshot_moref:
- LOG.info(_LI("There is no snapshot point for the snapshotted "
- "volume: %(snap)s. Not creating any backing for "
- "the volume: %(vol)s."),
- {'snap': snapshot['name'], 'vol': volume['name']})
- return
- src_vmdk_path = self.volumeops.get_vmdk_path(snapshot_moref)
- self._create_backing_by_copying(volume, src_vmdk_path,
- snapshot['volume_size'])
-
- def create_volume_from_snapshot(self, volume, snapshot):
- """Creates a volume from a snapshot.
-
- :param volume: Volume object
- :param snapshot: Snapshot object
- """
- self._create_volume_from_snapshot(volume, snapshot)
-
def _get_ds_name_folder_path(self, backing):
"""Get datastore name and folder path of the given backing.
"""
# We should use the disk type in volume type for backing's virtual
# disk.
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
# First, create a disk-less backing.
create_params = {CREATE_PARAM_DISK_LESS: True}
"""Creates volume from image using HttpNfc VM import.
Uses Nfc API to download the VMDK file from Glance. Nfc creates the
- backing VM that wraps the VMDK in the ESX/vCenter inventory.
+ backing VM that wraps the VMDK in the vCenter inventory.
This method assumes glance image is VMDK disk format and its
vmware_disktype is 'streamOptimized'.
"""
# prepare create spec for backing vm
profile_id = self._get_storage_profile_id(volume)
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
# The size of stream optimized glance image is often suspect,
# so better let vCenter figure out the disk capacity during import.
This method only supports Glance image of VMDK disk format.
Uses flat vmdk file copy for "sparse" and "preallocated" disk types
Uses HttpNfc import API for "streamOptimized" disk types. This API
- creates a backing VM that wraps the VMDK in the ESX/vCenter inventory.
+ creates a backing VM that wraps the VMDK in the vCenter inventory.
:param context: context
:param volume: Volume object
# Verify glance image is vmdk disk format
metadata = image_service.show(context, image_id)
- VMwareEsxVmdkDriver._validate_disk_format(metadata['disk_format'])
+ VMwareVcVmdkDriver._validate_disk_format(metadata['disk_format'])
# Validate container format; only 'bare' is supported currently.
container_format = metadata.get('container_format')
# validate disk format is vmdk
LOG.debug("Copy Volume: %s to new image.", volume['name'])
- VMwareEsxVmdkDriver._validate_disk_format(image_meta['disk_format'])
+ VMwareVcVmdkDriver._validate_disk_format(image_meta['disk_format'])
# get backing vm of volume and its vmdk path
backing = self.volumeops.get_backing(volume['name'])
return True
# Check whether we need disk type conversion.
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
- new_disk_type = VMwareEsxVmdkDriver._get_extra_spec_disk_type(
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
+ new_disk_type = VMwareVcVmdkDriver._get_extra_spec_disk_type(
new_type['id'])
need_disk_type_conversion = disk_type != new_disk_type
vm_import_spec = cf.create('ns0:VirtualMachineImportSpec')
profile_id = self._get_storage_profile_id(volume)
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
extra_config = self._get_extra_config(volume)
# We cannot determine the size of a virtual disk created from
# streamOptimized disk image. Set size to 0 and let vCenter
(host, rp, _folder, summary) = self._select_ds_for_volume(volume)
datastore = summary.datastore
- disk_type = VMwareEsxVmdkDriver._get_disk_type(volume)
+ disk_type = VMwareVcVmdkDriver._get_disk_type(volume)
dest = self.volumeops.clone_backing(dest_name, src, None,
volumeops.FULL_CLONE_TYPE,
datastore, disk_type, host, rp)
{'backup_id': backup['id'],
'name': volume['name']})
-
-class VMwareVcVmdkDriver(VMwareEsxVmdkDriver):
- """Manage volumes on VMware vCenter server."""
-
- # Minimum supported vCenter version.
- MIN_SUPPORTED_VC_VERSION = dist_version.LooseVersion('5.1')
-
- # PBM is enabled only for vCenter versions 5.5 and above
- PBM_ENABLED_VC_VERSION = dist_version.LooseVersion('5.5')
-
- def _do_deprecation_warning(self):
- # no deprecation warning for vCenter vmdk driver
- pass
-
- def __init__(self, *args, **kwargs):
- super(VMwareVcVmdkDriver, self).__init__(*args, **kwargs)
- self._session = None
-
@property
def session(self):
if not self._session:
def do_setup(self, context):
"""Any initialization the volume driver does while starting."""
- super(VMwareVcVmdkDriver, self).do_setup(context)
- # vCenter specific setup is done here
+ self._validate_params()
# Validate vCenter version.
vc_version = self._get_vc_version()
Code Review / openstack-build / cinder-build.git / commitdiff