# Copyright 2013 Hewlett-Packard Development Company, L.P. # 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. import datetime import time from glanceclient import client as glance_client from glanceclient import exc as glance_exc from keystoneauth1 import loading as kaloading import mock from oslo_config import cfg from oslo_utils import uuidutils from six.moves.urllib import parse as urlparse import testtools from ironic.common import context from ironic.common import exception from ironic.common.glance_service import base_image_service from ironic.common.glance_service import service_utils from ironic.common.glance_service.v2 import image_service as glance_v2 from ironic.common import image_service as service from ironic.tests import base from ironic.tests.unit import stubs CONF = cfg.CONF class NullWriter(object): """Used to test ImageService.get which takes a writer object.""" def write(self, *arg, **kwargs): pass class TestGlanceSerializer(testtools.TestCase): def test_serialize(self): metadata = {'name': 'image1', 'is_public': True, 'foo': 'bar', 'properties': { 'prop1': 'propvalue1', 'mappings': '[' '{"virtual":"aaa","device":"bbb"},' '{"virtual":"xxx","device":"yyy"}]', 'block_device_mapping': '[' '{"virtual_device":"fake","device_name":"/dev/fake"},' '{"virtual_device":"ephemeral0",' '"device_name":"/dev/fake0"}]'}} expected = { 'name': 'image1', 'is_public': True, 'foo': 'bar', 'properties': {'prop1': 'propvalue1', 'mappings': [ {'virtual': 'aaa', 'device': 'bbb'}, {'virtual': 'xxx', 'device': 'yyy'}, ], 'block_device_mapping': [ {'virtual_device': 'fake', 'device_name': '/dev/fake'}, {'virtual_device': 'ephemeral0', 'device_name': '/dev/fake0'} ] } } converted = service_utils._convert(metadata) self.assertEqual(expected, converted) class TestGlanceImageService(base.TestCase): NOW_GLANCE_OLD_FORMAT = "2010-10-11T10:30:22" NOW_GLANCE_FORMAT = "2010-10-11T10:30:22.000000" NOW_DATETIME = datetime.datetime(2010, 10, 11, 10, 30, 22) def setUp(self): super(TestGlanceImageService, self).setUp() self.client = stubs.StubGlanceClient() self.context = context.RequestContext(auth_token=True) self.context.user_id = 'fake' self.context.project_id = 'fake' self.service = service.GlanceImageService(self.client, 2, self.context) self.config(glance_api_servers=['http://localhost'], group='glance') self.config(auth_strategy='keystone', group='glance') @staticmethod def _make_fixture(**kwargs): fixture = {'name': None, 'properties': {}, 'status': None, 'is_public': None} fixture.update(kwargs) return stubs.FakeImage(fixture) @property def endpoint(self): # For glanceclient versions >= 0.13, the endpoint is located # under http_client (blueprint common-client-library-2) # I5addc38eb2e2dd0be91b566fda7c0d81787ffa75 # Test both options to keep backward compatibility if getattr(self.service.client, 'endpoint', None): endpoint = self.service.client.endpoint else: endpoint = self.service.client.http_client.endpoint return endpoint def _make_datetime_fixture(self): return self._make_fixture(created_at=self.NOW_GLANCE_FORMAT, updated_at=self.NOW_GLANCE_FORMAT, deleted_at=self.NOW_GLANCE_FORMAT) def test_show_passes_through_to_client(self): image_id = uuidutils.generate_uuid() image = self._make_fixture(name='image1', is_public=True, id=image_id) expected = { 'id': image_id, 'name': 'image1', 'is_public': True, 'size': None, 'min_disk': None, 'min_ram': None, 'disk_format': None, 'container_format': None, 'checksum': None, 'created_at': None, 'updated_at': None, 'deleted_at': None, 'deleted': None, 'status': None, 'properties': {}, 'owner': None, } with mock.patch.object(self.service, 'call', return_value=image, autospec=True): image_meta = self.service.show(image_id) self.service.call.assert_called_once_with('get', image_id) self.assertEqual(expected, image_meta) def test_show_makes_datetimes(self): image_id = uuidutils.generate_uuid() image = self._make_datetime_fixture() with mock.patch.object(self.service, 'call', return_value=image, autospec=True): image_meta = self.service.show(image_id) self.service.call.assert_called_once_with('get', image_id) self.assertEqual(self.NOW_DATETIME, image_meta['created_at']) self.assertEqual(self.NOW_DATETIME, image_meta['updated_at']) def test_show_raises_when_no_authtoken_in_the_context(self): self.context.auth_token = False self.assertRaises(exception.ImageNotFound, self.service.show, uuidutils.generate_uuid()) @mock.patch.object(time, 'sleep', autospec=True) def test_download_with_retries(self, mock_sleep): tries = [0] class MyGlanceStubClient(stubs.StubGlanceClient): """A client that fails the first time, then succeeds.""" def get(self, image_id): if tries[0] == 0: tries[0] = 1 raise glance_exc.ServiceUnavailable('') else: return {} stub_client = MyGlanceStubClient() stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_service = service.GlanceImageService(stub_client, 1, stub_context) image_id = uuidutils.generate_uuid() writer = NullWriter() # When retries are disabled, we should get an exception self.config(glance_num_retries=0, group='glance') self.assertRaises(exception.GlanceConnectionFailed, stub_service.download, image_id, writer) # Now lets enable retries. No exception should happen now. tries = [0] self.config(glance_num_retries=1, group='glance') stub_service.download(image_id, writer) self.assertTrue(mock_sleep.called) def test_download_no_data(self): self.client.fake_wrapped = None image_id = uuidutils.generate_uuid() image = self._make_datetime_fixture() with mock.patch.object(self.client, 'get', return_value=image, autospec=True): self.assertRaisesRegex(exception.ImageDownloadFailed, 'image contains no data', self.service.download, image_id) @mock.patch('sendfile.sendfile', autospec=True) @mock.patch('os.path.getsize', autospec=True) @mock.patch('%s.open' % __name__, new=mock.mock_open(), create=True) def test_download_file_url(self, mock_getsize, mock_sendfile): # NOTE: only in v2 API class MyGlanceStubClient(stubs.StubGlanceClient): """A client that returns a file url.""" s_tmpfname = '/whatever/source' def get(self, image_id): return type('GlanceTestDirectUrlMeta', (object,), {'direct_url': 'file://%s' + self.s_tmpfname}) stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_client = MyGlanceStubClient() stub_service = service.GlanceImageService(stub_client, context=stub_context, version=2) image_id = uuidutils.generate_uuid() self.config(allowed_direct_url_schemes=['file'], group='glance') # patching open in base_image_service module namespace # to make call-spec assertions with mock.patch('ironic.common.glance_service.base_image_service.open', new=mock.mock_open(), create=True) as mock_ironic_open: with open('/whatever/target', 'w') as mock_target_fd: stub_service.download(image_id, mock_target_fd) # assert the image data was neither read nor written # but rather sendfiled mock_ironic_open.assert_called_once_with(MyGlanceStubClient.s_tmpfname, 'r') mock_source_fd = mock_ironic_open() self.assertFalse(mock_source_fd.read.called) self.assertFalse(mock_target_fd.write.called) mock_sendfile.assert_called_once_with( mock_target_fd.fileno(), mock_source_fd.fileno(), 0, mock_getsize(MyGlanceStubClient.s_tmpfname)) def test_client_forbidden_converts_to_imagenotauthed(self): class MyGlanceStubClient(stubs.StubGlanceClient): """A client that raises a Forbidden exception.""" def get(self, image_id): raise glance_exc.Forbidden(image_id) stub_client = MyGlanceStubClient() stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_service = service.GlanceImageService(stub_client, 1, stub_context) image_id = uuidutils.generate_uuid() writer = NullWriter() self.assertRaises(exception.ImageNotAuthorized, stub_service.download, image_id, writer) def test_client_httpforbidden_converts_to_imagenotauthed(self): class MyGlanceStubClient(stubs.StubGlanceClient): """A client that raises a HTTPForbidden exception.""" def get(self, image_id): raise glance_exc.HTTPForbidden(image_id) stub_client = MyGlanceStubClient() stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_service = service.GlanceImageService(stub_client, 1, stub_context) image_id = uuidutils.generate_uuid() writer = NullWriter() self.assertRaises(exception.ImageNotAuthorized, stub_service.download, image_id, writer) def test_client_notfound_converts_to_imagenotfound(self): class MyGlanceStubClient(stubs.StubGlanceClient): """A client that raises a NotFound exception.""" def get(self, image_id): raise glance_exc.NotFound(image_id) stub_client = MyGlanceStubClient() stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_service = service.GlanceImageService(stub_client, 1, stub_context) image_id = uuidutils.generate_uuid() writer = NullWriter() self.assertRaises(exception.ImageNotFound, stub_service.download, image_id, writer) def test_client_httpnotfound_converts_to_imagenotfound(self): class MyGlanceStubClient(stubs.StubGlanceClient): """A client that raises a HTTPNotFound exception.""" def get(self, image_id): raise glance_exc.HTTPNotFound(image_id) stub_client = MyGlanceStubClient() stub_context = context.RequestContext(auth_token=True) stub_context.user_id = 'fake' stub_context.project_id = 'fake' stub_service = service.GlanceImageService(stub_client, 1, stub_context) image_id = uuidutils.generate_uuid() writer = NullWriter() self.assertRaises(exception.ImageNotFound, stub_service.download, image_id, writer) @mock.patch('ironic.common.keystone.get_auth', autospec=True, return_value=mock.sentinel.auth) @mock.patch('ironic.common.keystone.get_service_auth', autospec=True, return_value=mock.sentinel.sauth) @mock.patch('ironic.common.keystone.get_adapter', autospec=True) @mock.patch('ironic.common.keystone.get_session', autospec=True, return_value=mock.sentinel.session) @mock.patch.object(glance_client, 'Client', autospec=True) class CheckImageServiceTestCase(base.TestCase): def setUp(self): super(CheckImageServiceTestCase, self).setUp() self.context = context.RequestContext(global_request_id='global') self.service = service.GlanceImageService(None, 1, self.context) # NOTE(pas-ha) register keystoneauth dynamic options manually plugin = kaloading.get_plugin_loader('password') opts = kaloading.get_auth_plugin_conf_options(plugin) self.cfg_fixture.register_opts(opts, group='glance') self.config(auth_type='password', auth_url='viking', username='spam', password='ham', project_name='parrot', service_type='image', region_name='SomeRegion', interface='internal', auth_strategy='keystone', group='glance') base_image_service._GLANCE_SESSION = None def test_check_image_service_client_already_set(self, mock_gclient, mock_sess, mock_adapter, mock_sauth, mock_auth): def func(self): return True self.service.client = True wrapped_func = base_image_service.check_image_service(func) self.assertTrue(wrapped_func(self.service)) self.assertEqual(0, mock_gclient.call_count) self.assertEqual(0, mock_sess.call_count) self.assertEqual(0, mock_adapter.call_count) self.assertEqual(0, mock_auth.call_count) self.assertEqual(0, mock_sauth.call_count) def _assert_client_call(self, mock_gclient, url, user=False): mock_gclient.assert_called_once_with( 1, session=mock.sentinel.session, global_request_id='global', auth=mock.sentinel.sauth if user else mock.sentinel.auth, endpoint_override=url) def test_check_image_service__config_auth(self, mock_gclient, mock_sess, mock_adapter, mock_sauth, mock_auth): def func(service, *args, **kwargs): return args, kwargs mock_adapter.return_value = adapter = mock.Mock() adapter.get_endpoint.return_value = 'glance_url' uuid = uuidutils.generate_uuid() params = {'image_href': uuid} wrapped_func = base_image_service.check_image_service(func) self.assertEqual(((), params), wrapped_func(self.service, **params)) self._assert_client_call(mock_gclient, 'glance_url') mock_auth.assert_called_once_with('glance') mock_sess.assert_called_once_with('glance') mock_adapter.assert_called_once_with('glance', session=mock.sentinel.session, auth=mock.sentinel.auth) adapter.get_endpoint.assert_called_once_with() self.assertEqual(0, mock_sauth.call_count) def test_check_image_service__token_auth(self, mock_gclient, mock_sess, mock_adapter, mock_sauth, mock_auth): def func(service, *args, **kwargs): return args, kwargs self.service.context = context.RequestContext( auth_token='token', global_request_id='global') mock_adapter.return_value = adapter = mock.Mock() adapter.get_endpoint.return_value = 'glance_url' uuid = uuidutils.generate_uuid() params = {'image_href': uuid} wrapped_func = base_image_service.check_image_service(func) self.assertEqual(((), params), wrapped_func(self.service, **params)) self._assert_client_call(mock_gclient, 'glance_url', user=True) mock_sess.assert_called_once_with('glance') mock_adapter.assert_called_once_with('glance', session=mock.sentinel.session, auth=mock.sentinel.auth) mock_sauth.assert_called_once_with(self.service.context, 'glance_url', mock.sentinel.auth) mock_auth.assert_called_once_with('glance') def test_check_image_service__deprecated_opts(self, mock_gclient, mock_sess, mock_adapter, mock_sauth, mock_auth): def func(service, *args, **kwargs): return args, kwargs mock_adapter.return_value = adapter = mock.Mock() adapter.get_endpoint.return_value = 'glance_url' uuid = uuidutils.generate_uuid() params = {'image_href': uuid} self.config(glance_api_servers='https://localhost:1234', glance_api_insecure=True, glance_cafile='cafile', region_name=None, group='glance') self.config(region_name='OtherRegion', group='keystone') wrapped_func = base_image_service.check_image_service(func) self.assertEqual(((), params), wrapped_func(self.service, **params)) self.assertEqual('https://localhost:1234', base_image_service.CONF.glance.endpoint_override) self._assert_client_call(mock_gclient, 'glance_url') mock_sess.assert_called_once_with('glance', insecure=True, cacert='cafile') mock_adapter.assert_called_once_with( 'glance', session=mock.sentinel.session, auth=mock.sentinel.auth, region_name='OtherRegion') self.assertEqual(0, mock_sauth.call_count) mock_auth.assert_called_once_with('glance') def test_check_image_service__no_auth(self, mock_gclient, mock_sess, mock_adapter, mock_sauth, mock_auth): def func(service, *args, **kwargs): return args, kwargs self.config(endpoint_override='foo', auth_strategy='noauth', group='glance') mock_adapter.return_value = adapter = mock.Mock() adapter.get_endpoint.return_value = 'foo' uuid = uuidutils.generate_uuid() params = {'image_href': uuid} wrapped_func = base_image_service.check_image_service(func) self.assertEqual(((), params), wrapped_func(self.service, **params)) self.assertEqual('none', base_image_service.CONF.glance.auth_type) self._assert_client_call(mock_gclient, 'foo') mock_sess.assert_called_once_with('glance') mock_adapter.assert_called_once_with('glance', session=mock.sentinel.session, auth=mock.sentinel.auth) self.assertEqual(0, mock_sauth.call_count) def _create_failing_glance_client(info): class MyGlanceStubClient(stubs.StubGlanceClient): """A client that fails the first time, then succeeds.""" def get(self, image_id): info['num_calls'] += 1 if info['num_calls'] == 1: raise glance_exc.ServiceUnavailable('') return {} return MyGlanceStubClient() class TestGlanceSwiftTempURL(base.TestCase): def setUp(self): super(TestGlanceSwiftTempURL, self).setUp() client = stubs.StubGlanceClient() self.context = context.RequestContext() self.context.auth_token = 'fake' self.service = service.GlanceImageService(client, 2, self.context) self.config(swift_temp_url_key='correcthorsebatterystaple', group='glance') self.config(swift_endpoint_url='https://swift.example.com', group='glance') self.config(swift_account='AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30', group='glance') self.config(swift_api_version='v1', group='glance') self.config(swift_container='glance', group='glance') self.config(swift_temp_url_duration=1200, group='glance') self.config(swift_store_multiple_containers_seed=0, group='glance') self.fake_image = { 'id': '757274c4-2856-4bd2-bb20-9a4a231e187b' } @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url(self, tempurl_mock): path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') @mock.patch('ironic.common.keystone.get_adapter', autospec=True) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_endpoint_detected(self, tempurl_mock, adapter_mock): self.config(swift_endpoint_url=None, group='glance') path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') endpoint = 'http://another.example.com:8080' adapter_mock.return_value.get_endpoint.return_value = endpoint self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(endpoint + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') @mock.patch('ironic.common.keystone.get_adapter', autospec=True) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_endpoint_with_suffix(self, tempurl_mock, adapter_mock): self.config(swift_endpoint_url=None, group='glance') path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') endpoint = 'http://another.example.com:8080' adapter_mock.return_value.get_endpoint.return_value = ( endpoint + '/v1/AUTH_foobar') self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(endpoint + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') @mock.patch('ironic.common.swift.get_swift_session', autospec=True) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_account_detected(self, tempurl_mock, swift_mock): self.config(swift_account=None, group='glance') path = ('/v1/AUTH_42/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') auth_ref = swift_mock.return_value.auth.get_auth_ref.return_value auth_ref.project_id = '42' self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') swift_mock.assert_called_once_with() @mock.patch('ironic.common.swift.SwiftAPI', autospec=True) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_key_detected(self, tempurl_mock, swift_mock): self.config(swift_temp_url_key=None, group='glance') path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') conn = swift_mock.return_value.connection conn.head_account.return_value = { 'x-account-meta-temp-url-key': 'secret' } self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key='secret', method='GET') conn.head_account.assert_called_once_with() @mock.patch('ironic.common.swift.SwiftAPI', autospec=True) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_no_key_detected(self, tempurl_mock, swift_mock): self.config(swift_temp_url_key=None, group='glance') path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') conn = swift_mock.return_value.connection conn.head_account.return_value = {} self.service._validate_temp_url_config = mock.Mock() self.assertRaises(exception.InvalidParameterValue, self.service.swift_temp_url, image_info=self.fake_image) conn.head_account.assert_called_once_with() @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_invalid_image_info(self, tempurl_mock): self.service._validate_temp_url_config = mock.Mock() image_info = {} self.assertRaises(exception.ImageUnacceptable, self.service.swift_temp_url, image_info) image_info = {'id': 'not an id'} self.assertRaises(exception.ImageUnacceptable, self.service.swift_temp_url, image_info) self.assertFalse(tempurl_mock.called) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_radosgw(self, tempurl_mock): self.config(object_store_endpoint_type='radosgw', group='deploy') path = ('/v1' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual( (urlparse.urljoin(CONF.glance.swift_endpoint_url, 'swift') + tempurl_mock.return_value), temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_radosgw_endpoint_with_swift(self, tempurl_mock): self.config(swift_endpoint_url='https://swift.radosgw.com/swift', group='glance') self.config(object_store_endpoint_type='radosgw', group='deploy') path = ('/v1' '/glance' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual( CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_radosgw_endpoint_invalid(self, tempurl_mock): self.config(swift_endpoint_url='https://swift.radosgw.com/eggs/', group='glance') self.config(object_store_endpoint_type='radosgw', group='deploy') self.service._validate_temp_url_config = mock.Mock() self.assertRaises(exception.InvalidParameterValue, self.service.swift_temp_url, self.fake_image) self.assertFalse(tempurl_mock.called) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_swift_temp_url_multiple_containers(self, tempurl_mock): self.config(swift_store_multiple_containers_seed=8, group='glance') path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance_757274c4' '/757274c4-2856-4bd2-bb20-9a4a231e187b') tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') self.service._validate_temp_url_config = mock.Mock() temp_url = self.service.swift_temp_url(image_info=self.fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') def test_swift_temp_url_url_bad_no_info(self): self.assertRaises(exception.ImageUnacceptable, self.service.swift_temp_url, image_info={}) def test__validate_temp_url_config(self): self.service._validate_temp_url_config() def test__validate_temp_url_key_no_exception(self): self.config(swift_temp_url_key=None, group='glance') self.config(object_store_endpoint_type='swift', group='deploy') self.service._validate_temp_url_config() def test__validate_temp_url_key_exception(self): self.config(swift_temp_url_key=None, group='glance') self.config(object_store_endpoint_type='radosgw', group='deploy') self.assertRaises(exception.MissingParameterValue, self.service._validate_temp_url_config) def test__validate_temp_url_no_account_exception_radosgw(self): self.config(swift_account=None, group='glance') self.config(object_store_endpoint_type='radosgw', group='deploy') self.service._validate_temp_url_config() def test__validate_temp_url_endpoint_less_than_download_delay(self): self.config(swift_temp_url_expected_download_start_delay=1000, group='glance') self.config(swift_temp_url_duration=15, group='glance') self.assertRaises(exception.InvalidParameterValue, self.service._validate_temp_url_config) def test__validate_temp_url_multiple_containers(self): self.config(swift_store_multiple_containers_seed=-1, group='glance') self.assertRaises(exception.InvalidParameterValue, self.service._validate_temp_url_config) self.config(swift_store_multiple_containers_seed=None, group='glance') self.assertRaises(exception.InvalidParameterValue, self.service._validate_temp_url_config) self.config(swift_store_multiple_containers_seed=33, group='glance') self.assertRaises(exception.InvalidParameterValue, self.service._validate_temp_url_config) class TestSwiftTempUrlCache(base.TestCase): def setUp(self): super(TestSwiftTempUrlCache, self).setUp() client = stubs.StubGlanceClient() self.context = context.RequestContext() self.context.auth_token = 'fake' self.config(swift_temp_url_expected_download_start_delay=100, group='glance') self.config(swift_temp_url_key='correcthorsebatterystaple', group='glance') self.config(swift_endpoint_url='https://swift.example.com', group='glance') self.config(swift_account='AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30', group='glance') self.config(swift_api_version='v1', group='glance') self.config(swift_container='glance', group='glance') self.config(swift_temp_url_duration=1200, group='glance') self.config(swift_temp_url_cache_enabled=True, group='glance') self.config(swift_store_multiple_containers_seed=0, group='glance') self.glance_service = service.GlanceImageService(client, version=2, context=self.context) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_add_items_to_cache(self, tempurl_mock): fake_image = { 'id': uuidutils.generate_uuid() } path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/%s' % fake_image['id']) exp_time = int(time.time()) + 1200 tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=%s' % exp_time) cleanup_mock = mock.Mock() self.glance_service._remove_expired_items_from_cache = cleanup_mock self.glance_service._validate_temp_url_config = mock.Mock() temp_url = self.glance_service.swift_temp_url( image_info=fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) cleanup_mock.assert_called_once_with() tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') self.assertEqual((temp_url, exp_time), self.glance_service._cache[fake_image['id']]) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_return_cached_tempurl(self, tempurl_mock): fake_image = { 'id': uuidutils.generate_uuid() } exp_time = int(time.time()) + 1200 temp_url = CONF.glance.swift_endpoint_url + ( '/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/%(uuid)s' '?temp_url_sig=hmacsig&temp_url_expires=%(exp_time)s' % {'uuid': fake_image['id'], 'exp_time': exp_time} ) self.glance_service._cache[fake_image['id']] = ( glance_v2.TempUrlCacheElement(url=temp_url, url_expires_at=exp_time) ) cleanup_mock = mock.Mock() self.glance_service._remove_expired_items_from_cache = cleanup_mock self.glance_service._validate_temp_url_config = mock.Mock() self.assertEqual( temp_url, self.glance_service.swift_temp_url(image_info=fake_image) ) cleanup_mock.assert_called_once_with() self.assertFalse(tempurl_mock.called) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def test_do_not_return_expired_tempurls(self, tempurl_mock): fake_image = { 'id': uuidutils.generate_uuid() } old_exp_time = int(time.time()) + 99 path = ( '/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/%s' % fake_image['id'] ) query = '?temp_url_sig=hmacsig&temp_url_expires=%s' self.glance_service._cache[fake_image['id']] = ( glance_v2.TempUrlCacheElement( url=(CONF.glance.swift_endpoint_url + path + query % old_exp_time), url_expires_at=old_exp_time) ) new_exp_time = int(time.time()) + 1200 tempurl_mock.return_value = ( path + query % new_exp_time) self.glance_service._validate_temp_url_config = mock.Mock() fresh_temp_url = self.glance_service.swift_temp_url( image_info=fake_image) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, fresh_temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') self.assertEqual( (fresh_temp_url, new_exp_time), self.glance_service._cache[fake_image['id']]) def test_remove_expired_items_from_cache(self): expired_items = { uuidutils.generate_uuid(): glance_v2.TempUrlCacheElement( 'fake-url-1', int(time.time()) - 10 ), uuidutils.generate_uuid(): glance_v2.TempUrlCacheElement( 'fake-url-2', int(time.time()) + 90 # Agent won't be able to start in time ) } valid_items = { uuidutils.generate_uuid(): glance_v2.TempUrlCacheElement( 'fake-url-3', int(time.time()) + 1000 ), uuidutils.generate_uuid(): glance_v2.TempUrlCacheElement( 'fake-url-4', int(time.time()) + 2000 ) } self.glance_service._cache.update(expired_items) self.glance_service._cache.update(valid_items) self.glance_service._remove_expired_items_from_cache() for uuid in valid_items: self.assertEqual(valid_items[uuid], self.glance_service._cache[uuid]) for uuid in expired_items: self.assertNotIn(uuid, self.glance_service._cache) @mock.patch('swiftclient.utils.generate_temp_url', autospec=True) def _test__generate_temp_url(self, fake_image, tempurl_mock): path = ('/v1/AUTH_a422b2-91f3-2f46-74b7-d7c9e8958f5d30' '/glance' '/%s' % fake_image['id']) tempurl_mock.return_value = ( path + '?temp_url_sig=hmacsig&temp_url_expires=1400001200') self.glance_service._validate_temp_url_config = mock.Mock() temp_url = self.glance_service._generate_temp_url( path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET', endpoint=CONF.glance.swift_endpoint_url, image_id=fake_image['id'] ) self.assertEqual(CONF.glance.swift_endpoint_url + tempurl_mock.return_value, temp_url) tempurl_mock.assert_called_with( path=path, seconds=CONF.glance.swift_temp_url_duration, key=CONF.glance.swift_temp_url_key, method='GET') def test_swift_temp_url_cache_enabled(self): fake_image = { 'id': uuidutils.generate_uuid() } rm_expired = mock.Mock() self.glance_service._remove_expired_items_from_cache = rm_expired self._test__generate_temp_url(fake_image) rm_expired.assert_called_once_with() self.assertIn(fake_image['id'], self.glance_service._cache) def test_swift_temp_url_cache_disabled(self): self.config(swift_temp_url_cache_enabled=False, group='glance') fake_image = { 'id': uuidutils.generate_uuid() } rm_expired = mock.Mock() self.glance_service._remove_expired_items_from_cache = rm_expired self._test__generate_temp_url(fake_image) self.assertFalse(rm_expired.called) self.assertNotIn(fake_image['id'], self.glance_service._cache) class TestServiceUtils(base.TestCase): def setUp(self): super(TestServiceUtils, self).setUp() service_utils._GLANCE_API_SERVER = None def test_parse_image_id_from_uuid(self): image_href = uuidutils.generate_uuid() parsed_id = service_utils.parse_image_id(image_href) self.assertEqual(image_href, parsed_id) def test_parse_image_id_from_glance(self): uuid = uuidutils.generate_uuid() image_href = u'glance://some-stuff/%s' % uuid parsed_id = service_utils.parse_image_id(image_href) self.assertEqual(uuid, parsed_id) def test_parse_image_id_from_glance_fail(self): self.assertRaises(exception.InvalidImageRef, service_utils.parse_image_id, u'glance://not-a-uuid') def test_parse_image_id_fail(self): self.assertRaises(exception.InvalidImageRef, service_utils.parse_image_id, u'http://spam.ham/eggs') def test_get_glance_api_server_fail(self): self.assertRaises(exception.InvalidImageRef, service_utils.get_glance_api_server, u'http://spam.ham/eggs') # TODO(pas-ha) remove in Rocky def test_get_glance_api_server(self): self.config(glance_api_servers='http://spam:1234, https://ham', group='glance') api_servers = {service_utils.get_glance_api_server( uuidutils.generate_uuid()) for i in range(2)} self.assertEqual({'http://spam:1234', 'https://ham'}, api_servers) def test_is_glance_image(self): image_href = u'uui\u0111' self.assertFalse(service_utils.is_glance_image(image_href)) image_href = u'733d1c44-a2ea-414b-aca7-69decf20d810' self.assertTrue(service_utils.is_glance_image(image_href)) image_href = u'glance://uui\u0111' self.assertTrue(service_utils.is_glance_image(image_href)) image_href = 'http://aaa/bbb' self.assertFalse(service_utils.is_glance_image(image_href)) image_href = None self.assertFalse(service_utils.is_glance_image(image_href)) def test_is_image_href_ordinary_file_name_true(self): image = u"\u0111eploy.iso" result = service_utils.is_image_href_ordinary_file_name(image) self.assertTrue(result) def test_is_image_href_ordinary_file_name_false(self): for image in ('733d1c44-a2ea-414b-aca7-69decf20d810', u'glance://\u0111eploy_iso', u'http://\u0111eploy_iso', u'https://\u0111eploy_iso', u'file://\u0111eploy_iso',): result = service_utils.is_image_href_ordinary_file_name(image) self.assertFalse(result)