376 lines
16 KiB
Python
376 lines
16 KiB
Python
# Copyright (c) 2014 VMware, Inc.
|
|
# 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 images.
|
|
"""
|
|
|
|
import os
|
|
import tarfile
|
|
|
|
import mock
|
|
from oslo_utils import units
|
|
from oslo_vmware import rw_handles
|
|
|
|
from nova import exception
|
|
from nova import objects
|
|
from nova import test
|
|
import nova.tests.unit.image.fake
|
|
from nova.tests import uuidsentinel
|
|
from nova.virt.vmwareapi import constants
|
|
from nova.virt.vmwareapi import images
|
|
from nova.virt.vmwareapi import vm_util
|
|
|
|
|
|
class VMwareImagesTestCase(test.NoDBTestCase):
|
|
"""Unit tests for Vmware API connection calls."""
|
|
|
|
def test_fetch_image(self):
|
|
"""Test fetching images."""
|
|
|
|
dc_name = 'fake-dc'
|
|
file_path = 'fake_file'
|
|
ds_name = 'ds1'
|
|
host = mock.MagicMock()
|
|
port = 7443
|
|
context = mock.MagicMock()
|
|
|
|
image_data = {
|
|
'id': nova.tests.unit.image.fake.get_valid_image_id(),
|
|
'disk_format': 'vmdk',
|
|
'size': 512,
|
|
}
|
|
read_file_handle = mock.MagicMock()
|
|
write_file_handle = mock.MagicMock()
|
|
read_iter = mock.MagicMock()
|
|
instance = objects.Instance(id=1,
|
|
uuid=uuidsentinel.foo,
|
|
image_ref=image_data['id'])
|
|
|
|
def fake_read_handle(read_iter):
|
|
return read_file_handle
|
|
|
|
def fake_write_handle(host, port, dc_name, ds_name, cookies,
|
|
file_path, file_size):
|
|
return write_file_handle
|
|
|
|
with test.nested(
|
|
mock.patch.object(rw_handles, 'ImageReadHandle',
|
|
side_effect=fake_read_handle),
|
|
mock.patch.object(rw_handles, 'FileWriteHandle',
|
|
side_effect=fake_write_handle),
|
|
mock.patch.object(images, 'image_transfer'),
|
|
mock.patch.object(images.IMAGE_API, 'get',
|
|
return_value=image_data),
|
|
mock.patch.object(images.IMAGE_API, 'download',
|
|
return_value=read_iter),
|
|
) as (glance_read, http_write, image_transfer, image_show,
|
|
image_download):
|
|
images.fetch_image(context, instance,
|
|
host, port, dc_name,
|
|
ds_name, file_path)
|
|
|
|
glance_read.assert_called_once_with(read_iter)
|
|
http_write.assert_called_once_with(host, port, dc_name, ds_name, None,
|
|
file_path, image_data['size'])
|
|
image_transfer.assert_called_once_with(read_file_handle,
|
|
write_file_handle)
|
|
image_download.assert_called_once_with(context, instance['image_ref'])
|
|
image_show.assert_called_once_with(context, instance['image_ref'])
|
|
|
|
def _setup_mock_get_remote_image_service(self,
|
|
mock_get_remote_image_service,
|
|
metadata):
|
|
mock_image_service = mock.MagicMock()
|
|
mock_image_service.show.return_value = metadata
|
|
mock_get_remote_image_service.return_value = [mock_image_service, 'i']
|
|
|
|
def test_get_vmdk_name_from_ovf(self):
|
|
ovf_path = os.path.join(os.path.dirname(__file__), 'ovf.xml')
|
|
with open(ovf_path) as f:
|
|
ovf_descriptor = f.read()
|
|
vmdk_name = images.get_vmdk_name_from_ovf(ovf_descriptor)
|
|
self.assertEqual("Damn_Small_Linux-disk1.vmdk", vmdk_name)
|
|
|
|
@mock.patch('oslo_vmware.rw_handles.ImageReadHandle')
|
|
@mock.patch('oslo_vmware.rw_handles.VmdkWriteHandle')
|
|
@mock.patch.object(tarfile, 'open')
|
|
def test_fetch_image_ova(self, mock_tar_open, mock_write_class,
|
|
mock_read_class):
|
|
session = mock.MagicMock()
|
|
ovf_descriptor = None
|
|
ovf_path = os.path.join(os.path.dirname(__file__), 'ovf.xml')
|
|
with open(ovf_path) as f:
|
|
ovf_descriptor = f.read()
|
|
|
|
with test.nested(
|
|
mock.patch.object(images.IMAGE_API, 'get'),
|
|
mock.patch.object(images.IMAGE_API, 'download'),
|
|
mock.patch.object(images, 'image_transfer'),
|
|
mock.patch.object(images, '_build_shadow_vm_config_spec'),
|
|
mock.patch.object(session, '_call_method'),
|
|
mock.patch.object(vm_util, 'get_vmdk_info')
|
|
) as (mock_image_api_get,
|
|
mock_image_api_download,
|
|
mock_image_transfer,
|
|
mock_build_shadow_vm_config_spec,
|
|
mock_call_method,
|
|
mock_get_vmdk_info):
|
|
image_data = {'id': 'fake-id',
|
|
'disk_format': 'vmdk',
|
|
'size': 512}
|
|
instance = mock.MagicMock()
|
|
instance.image_ref = image_data['id']
|
|
mock_image_api_get.return_value = image_data
|
|
|
|
vm_folder_ref = mock.MagicMock()
|
|
res_pool_ref = mock.MagicMock()
|
|
context = mock.MagicMock()
|
|
|
|
mock_read_handle = mock.MagicMock()
|
|
mock_read_class.return_value = mock_read_handle
|
|
mock_write_handle = mock.MagicMock()
|
|
mock_write_class.return_value = mock_write_handle
|
|
mock_write_handle.get_imported_vm.return_value = \
|
|
mock.sentinel.vm_ref
|
|
|
|
mock_ovf = mock.MagicMock()
|
|
mock_ovf.name = 'dsl.ovf'
|
|
mock_vmdk = mock.MagicMock()
|
|
mock_vmdk.name = "Damn_Small_Linux-disk1.vmdk"
|
|
|
|
def fake_extract(name):
|
|
if name == mock_ovf:
|
|
m = mock.MagicMock()
|
|
m.read.return_value = ovf_descriptor
|
|
return m
|
|
elif name == mock_vmdk:
|
|
return mock_read_handle
|
|
|
|
mock_tar = mock.MagicMock()
|
|
mock_tar.__iter__ = mock.Mock(return_value = iter([mock_ovf,
|
|
mock_vmdk]))
|
|
mock_tar.extractfile = fake_extract
|
|
mock_tar_open.return_value.__enter__.return_value = mock_tar
|
|
|
|
images.fetch_image_ova(
|
|
context, instance, session, 'fake-vm', 'fake-datastore',
|
|
vm_folder_ref, res_pool_ref)
|
|
|
|
mock_tar_open.assert_called_once_with(mode='r|',
|
|
fileobj=mock_read_handle)
|
|
mock_image_transfer.assert_called_once_with(mock_read_handle,
|
|
mock_write_handle)
|
|
mock_get_vmdk_info.assert_called_once_with(
|
|
session, mock.sentinel.vm_ref, 'fake-vm')
|
|
mock_call_method.assert_called_once_with(
|
|
session.vim, "UnregisterVM", mock.sentinel.vm_ref)
|
|
|
|
@mock.patch('oslo_vmware.rw_handles.ImageReadHandle')
|
|
@mock.patch('oslo_vmware.rw_handles.VmdkWriteHandle')
|
|
def test_fetch_image_stream_optimized(self,
|
|
mock_write_class,
|
|
mock_read_class):
|
|
"""Test fetching streamOptimized disk image."""
|
|
session = mock.MagicMock()
|
|
|
|
with test.nested(
|
|
mock.patch.object(images.IMAGE_API, 'get'),
|
|
mock.patch.object(images.IMAGE_API, 'download'),
|
|
mock.patch.object(images, 'image_transfer'),
|
|
mock.patch.object(images, '_build_shadow_vm_config_spec'),
|
|
mock.patch.object(session, '_call_method'),
|
|
mock.patch.object(vm_util, 'get_vmdk_info')
|
|
) as (mock_image_api_get,
|
|
mock_image_api_download,
|
|
mock_image_transfer,
|
|
mock_build_shadow_vm_config_spec,
|
|
mock_call_method,
|
|
mock_get_vmdk_info):
|
|
image_data = {'id': 'fake-id',
|
|
'disk_format': 'vmdk',
|
|
'size': 512}
|
|
instance = mock.MagicMock()
|
|
instance.image_ref = image_data['id']
|
|
mock_image_api_get.return_value = image_data
|
|
|
|
vm_folder_ref = mock.MagicMock()
|
|
res_pool_ref = mock.MagicMock()
|
|
context = mock.MagicMock()
|
|
|
|
mock_read_handle = mock.MagicMock()
|
|
mock_read_class.return_value = mock_read_handle
|
|
mock_write_handle = mock.MagicMock()
|
|
mock_write_class.return_value = mock_write_handle
|
|
mock_write_handle.get_imported_vm.return_value = \
|
|
mock.sentinel.vm_ref
|
|
|
|
images.fetch_image_stream_optimized(
|
|
context, instance, session, 'fake-vm', 'fake-datastore',
|
|
vm_folder_ref, res_pool_ref)
|
|
|
|
mock_image_transfer.assert_called_once_with(mock_read_handle,
|
|
mock_write_handle)
|
|
mock_call_method.assert_called_once_with(
|
|
session.vim, "UnregisterVM", mock.sentinel.vm_ref)
|
|
mock_get_vmdk_info.assert_called_once_with(
|
|
session, mock.sentinel.vm_ref, 'fake-vm')
|
|
|
|
def test_from_image_with_image_ref(self):
|
|
raw_disk_size_in_gb = 83
|
|
raw_disk_size_in_bytes = raw_disk_size_in_gb * units.Gi
|
|
image_id = nova.tests.unit.image.fake.get_valid_image_id()
|
|
mdata = {'size': raw_disk_size_in_bytes,
|
|
'disk_format': 'vmdk',
|
|
'properties': {
|
|
"vmware_ostype": constants.DEFAULT_OS_TYPE,
|
|
"vmware_adaptertype": constants.DEFAULT_ADAPTER_TYPE,
|
|
"vmware_disktype": constants.DEFAULT_DISK_TYPE,
|
|
"hw_vif_model": constants.DEFAULT_VIF_MODEL,
|
|
"vmware_linked_clone": True}}
|
|
mdata = objects.ImageMeta.from_dict(mdata)
|
|
with mock.patch.object(images, 'get_vsphere_location',
|
|
return_value=None):
|
|
img_props = images.VMwareImage.from_image(None, image_id, mdata)
|
|
|
|
image_size_in_kb = raw_disk_size_in_bytes / units.Ki
|
|
|
|
# assert that defaults are set and no value returned is left empty
|
|
self.assertEqual(constants.DEFAULT_OS_TYPE, img_props.os_type)
|
|
self.assertEqual(constants.DEFAULT_ADAPTER_TYPE,
|
|
img_props.adapter_type)
|
|
self.assertEqual(constants.DEFAULT_DISK_TYPE, img_props.disk_type)
|
|
self.assertEqual(constants.DEFAULT_VIF_MODEL, img_props.vif_model)
|
|
self.assertTrue(img_props.linked_clone)
|
|
self.assertEqual(image_size_in_kb, img_props.file_size_in_kb)
|
|
|
|
def _image_build(self, image_lc_setting, global_lc_setting,
|
|
disk_format=constants.DEFAULT_DISK_FORMAT,
|
|
os_type=constants.DEFAULT_OS_TYPE,
|
|
adapter_type=constants.DEFAULT_ADAPTER_TYPE,
|
|
disk_type=constants.DEFAULT_DISK_TYPE,
|
|
vif_model=constants.DEFAULT_VIF_MODEL,
|
|
vsphere_location=None):
|
|
self.flags(use_linked_clone=global_lc_setting, group='vmware')
|
|
raw_disk_size_in_gb = 93
|
|
raw_disk_size_in_btyes = raw_disk_size_in_gb * units.Gi
|
|
|
|
image_id = nova.tests.unit.image.fake.get_valid_image_id()
|
|
mdata = {'size': raw_disk_size_in_btyes,
|
|
'disk_format': disk_format,
|
|
'properties': {
|
|
"vmware_ostype": os_type,
|
|
"vmware_adaptertype": adapter_type,
|
|
"vmware_disktype": disk_type,
|
|
"hw_vif_model": vif_model}}
|
|
|
|
if image_lc_setting is not None:
|
|
mdata['properties']["vmware_linked_clone"] = image_lc_setting
|
|
|
|
context = mock.Mock()
|
|
mdata = objects.ImageMeta.from_dict(mdata)
|
|
with mock.patch.object(
|
|
images, 'get_vsphere_location', return_value=vsphere_location):
|
|
return images.VMwareImage.from_image(context, image_id, mdata)
|
|
|
|
def test_use_linked_clone_override_nf(self):
|
|
image_props = self._image_build(None, False)
|
|
self.assertFalse(image_props.linked_clone,
|
|
"No overrides present but still overridden!")
|
|
|
|
def test_use_linked_clone_override_nt(self):
|
|
image_props = self._image_build(None, True)
|
|
self.assertTrue(image_props.linked_clone,
|
|
"No overrides present but still overridden!")
|
|
|
|
def test_use_linked_clone_override_ny(self):
|
|
image_props = self._image_build(None, "yes")
|
|
self.assertTrue(image_props.linked_clone,
|
|
"No overrides present but still overridden!")
|
|
|
|
def test_use_linked_clone_override_ft(self):
|
|
image_props = self._image_build(False, True)
|
|
self.assertFalse(image_props.linked_clone,
|
|
"image level metadata failed to override global")
|
|
|
|
def test_use_linked_clone_override_string_nt(self):
|
|
image_props = self._image_build("no", True)
|
|
self.assertFalse(image_props.linked_clone,
|
|
"image level metadata failed to override global")
|
|
|
|
def test_use_linked_clone_override_string_yf(self):
|
|
image_props = self._image_build("yes", False)
|
|
self.assertTrue(image_props.linked_clone,
|
|
"image level metadata failed to override global")
|
|
|
|
def test_use_disk_format_iso(self):
|
|
image = self._image_build(None, True, disk_format='iso')
|
|
self.assertEqual('iso', image.file_type)
|
|
self.assertTrue(image.is_iso)
|
|
|
|
def test_use_bad_disk_format(self):
|
|
self.assertRaises(exception.InvalidDiskFormat,
|
|
self._image_build,
|
|
None,
|
|
True,
|
|
disk_format='bad_disk_format')
|
|
|
|
def test_image_no_defaults(self):
|
|
image = self._image_build(False, False,
|
|
disk_format='iso',
|
|
os_type='otherGuest',
|
|
adapter_type='lsiLogic',
|
|
disk_type='preallocated',
|
|
vif_model='e1000e')
|
|
self.assertEqual('iso', image.file_type)
|
|
self.assertEqual('otherGuest', image.os_type)
|
|
self.assertEqual('lsiLogic', image.adapter_type)
|
|
self.assertEqual('preallocated', image.disk_type)
|
|
self.assertEqual('e1000e', image.vif_model)
|
|
self.assertFalse(image.linked_clone)
|
|
|
|
def test_image_defaults(self):
|
|
image = images.VMwareImage(image_id='fake-image-id')
|
|
|
|
# N.B. We intentially don't use the defined constants here. Amongst
|
|
# other potential failures, we're interested in changes to their
|
|
# values, which would not otherwise be picked up.
|
|
self.assertEqual('otherGuest', image.os_type)
|
|
self.assertEqual('lsiLogic', image.adapter_type)
|
|
self.assertEqual('preallocated', image.disk_type)
|
|
self.assertEqual('e1000', image.vif_model)
|
|
|
|
def test_use_vsphere_location(self):
|
|
image = self._image_build(None, True, vsphere_location='vsphere://ok')
|
|
self.assertEqual('vsphere://ok', image.vsphere_location)
|
|
|
|
def test_get_vsphere_location(self):
|
|
expected = 'vsphere://ok'
|
|
metadata = {'locations': [{}, {'url': 'http://ko'}, {'url': expected}]}
|
|
with mock.patch.object(images.IMAGE_API, 'get', return_value=metadata):
|
|
context = mock.Mock()
|
|
observed = images.get_vsphere_location(context, 'image_id')
|
|
self.assertEqual(expected, observed)
|
|
|
|
def test_get_no_vsphere_location(self):
|
|
metadata = {'locations': [{}, {'url': 'http://ko'}]}
|
|
with mock.patch.object(images.IMAGE_API, 'get', return_value=metadata):
|
|
context = mock.Mock()
|
|
observed = images.get_vsphere_location(context, 'image_id')
|
|
self.assertIsNone(observed)
|
|
|
|
def test_get_vsphere_location_no_image(self):
|
|
context = mock.Mock()
|
|
observed = images.get_vsphere_location(context, None)
|
|
self.assertIsNone(observed)
|