text: # Copyright 2011 OpenStack Foundation
  # 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 filecmp
  import os
  import random
  import sys
  import tempfile
  import time
  
  import glanceclient.exc
  import mock
  from oslo.config import cfg
  import testtools
  
  from nova import context
  from nova import exception
  from nova.image import glance
  from nova import test
  from nova.tests.api.openstack import fakes
  from nova.tests.glance import stubs as glance_stubs
  from nova import utils
  import nova.virt.libvirt.utils as lv_utils
  
  CONF = cfg.CONF
  NOW_GLANCE_FORMAT = "2010-10-11T10:30:22.000000"
  
  
  class tzinfo(datetime.tzinfo):
      @staticmethod
      def utcoffset(*args, **kwargs):
          return datetime.timedelta()
  
  NOW_DATETIME = datetime.datetime(2010, 10, 11, 10, 30, 22, tzinfo=tzinfo())
  
  
  class TestConversions(test.NoDBTestCase):
      def test_convert_timestamps_to_datetimes(self):
          fixture = {'name': None,
                     'properties': {},
                     'status': None,
                     'is_public': None,
                     'created_at': NOW_GLANCE_FORMAT,
                     'updated_at': NOW_GLANCE_FORMAT,
                     'deleted_at': NOW_GLANCE_FORMAT}
          result = glance._convert_timestamps_to_datetimes(fixture)
          self.assertEqual(result['created_at'], NOW_DATETIME)
          self.assertEqual(result['updated_at'], NOW_DATETIME)
          self.assertEqual(result['deleted_at'], NOW_DATETIME)
  
      def _test_extracting_missing_attributes(self, include_locations):
          # Verify behavior from glance objects that are missing attributes
          class MyFakeGlanceImage(glance_stubs.FakeImage):
              def __init__(self, metadata):
                  IMAGE_ATTRIBUTES = ['size', 'owner', 'id', 'created_at',
                                      'updated_at', 'status', 'min_disk',
                                      'min_ram', 'is_public']
                  raw = dict.fromkeys(IMAGE_ATTRIBUTES)
                  raw.update(metadata)
                  self.__dict__['raw'] = raw
  
          metadata = {
              'id': 1,
              'created_at': NOW_DATETIME,
              'updated_at': NOW_DATETIME,
          }
          image = MyFakeGlanceImage(metadata)
          observed = glance._extract_attributes(
              image, include_locations=include_locations)
          expected = {
              'id': 1,
              'name': None,
              'is_public': None,
              'size': None,
              'min_disk': None,
              'min_ram': None,
              'disk_format': None,
              'container_format': None,
              'checksum': None,
              'created_at': NOW_DATETIME,
              'updated_at': NOW_DATETIME,
              'deleted_at': None,
              'deleted': None,
              'status': None,
              'properties': {},
              'owner': None
          }
          if include_locations:
              expected['locations'] = None
              expected['direct_url'] = None
          self.assertEqual(expected, observed)
  
      def test_extracting_missing_attributes_include_locations(self):
          self._test_extracting_missing_attributes(include_locations=True)
  
      def test_extracting_missing_attributes_exclude_locations(self):
          self._test_extracting_missing_attributes(include_locations=False)
  
  
  class TestExceptionTranslations(test.NoDBTestCase):
  
      def test_client_forbidden_to_imagenotauthed(self):
          in_exc = glanceclient.exc.Forbidden('123')
          out_exc = glance._translate_image_exception('123', in_exc)
          self.assertIsInstance(out_exc, exception.ImageNotAuthorized)
  
      def test_client_httpforbidden_converts_to_imagenotauthed(self):
          in_exc = glanceclient.exc.HTTPForbidden('123')
          out_exc = glance._translate_image_exception('123', in_exc)
          self.assertIsInstance(out_exc, exception.ImageNotAuthorized)
  
      def test_client_notfound_converts_to_imagenotfound(self):
          in_exc = glanceclient.exc.NotFound('123')
          out_exc = glance._translate_image_exception('123', in_exc)
          self.assertIsInstance(out_exc, exception.ImageNotFound)
  
      def test_client_httpnotfound_converts_to_imagenotfound(self):
          in_exc = glanceclient.exc.HTTPNotFound('123')
          out_exc = glance._translate_image_exception('123', in_exc)
          self.assertIsInstance(out_exc, exception.ImageNotFound)
  
  
  class TestGlanceSerializer(test.NoDBTestCase):
      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'}]}}
  
          converted_expected = {
              'name': 'image1',
              'is_public': True,
              'foo': 'bar',
              'properties': {
                  'prop1': 'propvalue1',
                  'mappings':
                  '[{"device": "bbb", "virtual": "aaa"}, '
                  '{"device": "yyy", "virtual": "xxx"}]',
                  'block_device_mapping':
                  '[{"virtual_device": "fake", "device_name": "/dev/fake"}, '
                  '{"virtual_device": "ephemeral0", '
                  '"device_name": "/dev/fake0"}]'}}
          converted = glance._convert_to_string(metadata)
          self.assertEqual(converted, converted_expected)
          self.assertEqual(glance._convert_from_string(converted), metadata)
  
  
  class TestGetImageService(test.NoDBTestCase):
      @mock.patch.object(glance.GlanceClientWrapper, '__init__',
                         return_value=None)
      def test_get_remote_service_from_id(self, gcwi_mocked):
          id_or_uri = '123'
          _ignored, image_id = glance.get_remote_image_service(
                  mock.sentinel.ctx, id_or_uri)
          self.assertEqual(id_or_uri, image_id)
          gcwi_mocked.assert_called_once_with()
  
      @mock.patch.object(glance.GlanceClientWrapper, '__init__',
                         return_value=None)
      def test_get_remote_service_from_href(self, gcwi_mocked):
          id_or_uri = 'http://127.0.0.1/123'
          _ignored, image_id = glance.get_remote_image_service(
                  mock.sentinel.ctx, id_or_uri)
          self.assertEqual('123', image_id)
          gcwi_mocked.assert_called_once_with(context=mock.sentinel.ctx,
                                              host='127.0.0.1',
                                              port=80,
                                              use_ssl=False)
  
  
  class NullWriter(object):
      """Used to test ImageService.get which takes a writer object."""
  
      def write(self, *arg, **kwargs):
          pass
  
  
  class TestGlanceImageService(test.NoDBTestCase):
      """Tests the Glance image service.
  
      At a high level, the translations involved are:
  
          1. Glance -> ImageService - This is needed so we can support
             multple ImageServices (Glance, Local, etc)
  
          2. ImageService -> API - This is needed so we can support multple
             APIs (OpenStack, EC2)
  
      """
  
      def setUp(self):
          super(TestGlanceImageService, self).setUp()
          fakes.stub_out_compute_api_snapshot(self.stubs)
  
          self.client = glance_stubs.StubGlanceClient()
          self.service = self._create_image_service(self.client)
          self.context = context.RequestContext('fake', 'fake', auth_token=True)
          self.files_to_clean = []
  
      def tearDown(self):
          super(TestGlanceImageService, self).tearDown()
          for f in self.files_to_clean:
              try:
                  os.unlink(f)
              except os.error:
                  pass
  
      def _get_tempfile(self):
          (outfd, config_filename) = tempfile.mkstemp(prefix='nova_glance_tests')
          self.files_to_clean.append(config_filename)
          return (outfd, config_filename)
  
      def _create_image_service(self, client):
          def _fake_create_glance_client(context, host, port, use_ssl, version):
              return client
  
          self.stubs.Set(glance, '_create_glance_client',
                  _fake_create_glance_client)
  
          client_wrapper = glance.GlanceClientWrapper(
                  'fake', 'fake_host', 9292)
          return glance.GlanceImageService(client=client_wrapper)
  
      def test_download_with_retries(self):
          tries = [0]
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              """A client that fails the first time, then succeeds."""
              def get(self, image_id):
                  if tries[0] == 0:
                      tries[0] = 1
                      raise glanceclient.exc.ServiceUnavailable('')
                  else:
                      return {}
  
          client = MyGlanceStubClient()
          service = self._create_image_service(client)
          image_id = 1  # doesn't matter
          writer = NullWriter()
  
          # When retries are disabled, we should get an exception
          self.flags(num_retries=0, group='glance')
          self.assertRaises(exception.GlanceConnectionFailed,
                  service.download, self.context, image_id, data=writer)
  
          # Now lets enable retries. No exception should happen now.
          tries = [0]
          self.flags(num_retries=1, group='glance')
          service.download(self.context, image_id, data=writer)
  
      def test_download_file_url(self):
          self.flags(allowed_direct_url_schemes=['file'], group='glance')
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              """A client that returns a file url."""
  
              (outfd, s_tmpfname) = tempfile.mkstemp(prefix='directURLsrc')
              outf = os.fdopen(outfd, 'w')
              inf = open('/dev/urandom', 'r')
              for i in range(10):
                  _data = inf.read(1024)
                  outf.write(_data)
              outf.close()
  
              def get(self, image_id):
                  return type('GlanceTestDirectUrlMeta', (object,),
                              {'status': 'active',
                               'direct_url': 'file://%s' + self.s_tmpfname})
  
          client = MyGlanceStubClient()
          (outfd, tmpfname) = tempfile.mkstemp(prefix='directURLdst')
          os.close(outfd)
  
          service = self._create_image_service(client)
          image_id = 1  # doesn't matter
  
          service.download(self.context, image_id, dst_path=tmpfname)
  
          # compare the two files
          rc = filecmp.cmp(tmpfname, client.s_tmpfname)
          self.assertTrue(rc, "The file %s and %s should be the same" %
                          (tmpfname, client.s_tmpfname))
          os.remove(client.s_tmpfname)
          os.remove(tmpfname)
  
      @mock.patch('nova.virt.libvirt.utils.copy_image')
      def test_download_module_filesystem_match(self, mock_copy_image):
  
          mountpoint = '/'
          fs_id = 'someid'
          desc = {'id': fs_id, 'mountpoint': mountpoint}
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              outer_test = self
  
              def get(self, image_id):
                  return type('GlanceLocations', (object,),
                              {'status': 'active',
                               'locations': [
                                  {'url': 'file:///' + os.devnull,
                                   'metadata': desc}]})
  
              def data(self, image_id):
                  self.outer_test.fail('This should not be called because the '
                                       'transfer module should have intercepted '
                                       'it.')
  
          image_id = 1  # doesn't matter
          client = MyGlanceStubClient()
          self.flags(allowed_direct_url_schemes=['file'], group='glance')
          self.flags(group='image_file_url', filesystems=['gluster'])
          service = self._create_image_service(client)
          # NOTE(Jbresnah) The following options must be added after the module
          # has added the specific groups.
          self.flags(group='image_file_url:gluster', id=fs_id)
          self.flags(group='image_file_url:gluster', mountpoint=mountpoint)
  
          dest_file = os.devnull
          service.download(self.context, image_id, dst_path=dest_file)
          mock_copy_image.assert_called_once_with('/' + os.devnull, os.devnull)
  
      def test_download_module_no_filesystem_match(self):
          mountpoint = '/'
          fs_id = 'someid'
          desc = {'id': fs_id, 'mountpoint': mountpoint}
          some_data = "sfxvdwjer"
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              outer_test = self
  
              def get(self, image_id):
                  return type('GlanceLocations', (object,),
                              {'status': 'active',
                               'locations': [
                                  {'url': 'file:///' + os.devnull,
                                   'metadata': desc}]})
  
              def data(self, image_id):
                  return some_data
  
          def _fake_copyfile(source, dest):
              self.fail('This should not be called because a match should not '
                        'have been found.')
          self.stubs.Set(lv_utils, 'copy_image', _fake_copyfile)
  
          image_id = 1  # doesn't matter
          client = MyGlanceStubClient()
          self.flags(allowed_direct_url_schemes=['file'], group='glance')
          self.flags(group='image_file_url', filesystems=['gluster'])
          service = self._create_image_service(client)
          # NOTE(Jbresnah) The following options must be added after the module
          # has added the specific groups.
          self.flags(group='image_file_url:gluster', id='someotherid')
          self.flags(group='image_file_url:gluster', mountpoint=mountpoint)
  
          service.download(self.context, image_id,
                           dst_path=os.devnull,
                           data=None)
  
      def test_download_module_mountpoints(self):
          glance_mount = '/glance/mount/point'
          _, data_filename = self._get_tempfile()
          nova_mount = os.path.dirname(data_filename)
          source_path = os.path.basename(data_filename)
          file_url = 'file://%s' % os.path.join(glance_mount, source_path)
          file_system_id = 'test_FS_ID'
          file_system_desc = {'id': file_system_id, 'mountpoint': glance_mount}
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              outer_test = self
  
              def get(self, image_id):
                  return type('GlanceLocations', (object,),
                              {'status': 'active',
                               'locations': [{'url': file_url,
                                              'metadata': file_system_desc}]})
  
              def data(self, image_id):
                  self.outer_test.fail('This should not be called because the '
                                       'transfer module should have intercepted '
                                       'it.')
  
          self.copy_called = False
  
          def _fake_copyfile(source, dest):
              self.assertEqual(source, data_filename)
              self.copy_called = True
          self.stubs.Set(lv_utils, 'copy_image', _fake_copyfile)
  
          self.flags(allowed_direct_url_schemes=['file'], group='glance')
          self.flags(group='image_file_url', filesystems=['gluster'])
          image_id = 1  # doesn't matter
          client = MyGlanceStubClient()
          service = self._create_image_service(client)
          self.flags(group='image_file_url:gluster', id=file_system_id)
          self.flags(group='image_file_url:gluster', mountpoint=nova_mount)
  
          service.download(self.context, image_id, dst_path=os.devnull)
          self.assertTrue(self.copy_called)
  
      @mock.patch('nova.virt.libvirt.utils.copy_image')
      def test_download_module_file_bad_module(self, mock_copy_image):
          _, data_filename = self._get_tempfile()
          file_url = 'applesauce://%s' % data_filename
  
          class MyGlanceStubClient(glance_stubs.StubGlanceClient):
              data_called = False
  
              def get(self, image_id):
                  return type('GlanceLocations', (object,),
                              {'status': 'active',
                               'locations': [{'url': file_url,
                                              'metadata': {}}]})
  
              def data(self, image_id):
                  self.data_called = True
                  return "someData"
  
          self.flags(allowed_direct_url_schemes=['applesauce'], group='glance')
          self.flags(allowed_direct_url_schemes=['file'], group='glance')
          image_id = 1  # doesn't matter
          client = MyGlanceStubClient()
          service = self._create_image_service(client)
  
          # by not calling copyfileobj in the file download module we verify
          # that the requirements were not met for its use
          service.download(self.context, image_id, dst_path=os.devnull)
          self.assertTrue(client.data_called)
          self.assertFalse(mock_copy_image.called)
  
  
  def _create_failing_glance_client(info):
      class MyGlanceStubClient(glance_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 glanceclient.exc.ServiceUnavailable('')
              return {}
  
      return MyGlanceStubClient()
  
  
  class TestIsImageAvailable(test.NoDBTestCase):
      """Tests the internal _is_image_available function."""
  
      class ImageSpecV2(object):
          visibility = None
          properties = None
  
      class ImageSpecV1(object):
          is_public = None
          properties = None
  
      def test_auth_token_override(self):
          ctx = mock.MagicMock(auth_token=True)
          img = mock.MagicMock()
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
          self.assertFalse(img.called)
  
      def test_admin_override(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=True)
          img = mock.MagicMock()
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
          self.assertFalse(img.called)
  
      def test_v2_visibility(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False)
          # We emulate warlock validation that throws an AttributeError
          # if you try to call is_public on an image model returned by
          # a call to V2 image.get(). Here, the ImageSpecV2 does not have
          # an is_public attribute and MagicMock will throw an AttributeError.
          img = mock.MagicMock(visibility='PUBLIC',
                               spec=TestIsImageAvailable.ImageSpecV2)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
      def test_v1_is_public(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False)
          img = mock.MagicMock(is_public=True,
                               spec=TestIsImageAvailable.ImageSpecV1)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
      def test_project_is_owner(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False,
                               project_id='123')
          props = {
              'owner_id': '123'
          }
          img = mock.MagicMock(visibility='private', properties=props,
                               spec=TestIsImageAvailable.ImageSpecV2)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
          ctx.reset_mock()
          img = mock.MagicMock(is_public=False, properties=props,
                               spec=TestIsImageAvailable.ImageSpecV1)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
      def test_project_context_matches_project_prop(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False,
                               project_id='123')
          props = {
              'project_id': '123'
          }
          img = mock.MagicMock(visibility='private', properties=props,
                               spec=TestIsImageAvailable.ImageSpecV2)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
          ctx.reset_mock()
          img = mock.MagicMock(is_public=False, properties=props,
                               spec=TestIsImageAvailable.ImageSpecV1)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
      def test_no_user_in_props(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False,
                               project_id='123')
          props = {
          }
          img = mock.MagicMock(visibility='private', properties=props,
                               spec=TestIsImageAvailable.ImageSpecV2)
  
          res = glance._is_image_available(ctx, img)
          self.assertFalse(res)
  
          ctx.reset_mock()
          img = mock.MagicMock(is_public=False, properties=props,
                               spec=TestIsImageAvailable.ImageSpecV1)
  
          res = glance._is_image_available(ctx, img)
          self.assertFalse(res)
  
      def test_user_matches_context(self):
          ctx = mock.MagicMock(auth_token=False, is_admin=False,
                               user_id='123')
          props = {
              'user_id': '123'
          }
          img = mock.MagicMock(visibility='private', properties=props,
                               spec=TestIsImageAvailable.ImageSpecV2)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
          ctx.reset_mock()
          img = mock.MagicMock(is_public=False, properties=props,
                               spec=TestIsImageAvailable.ImageSpecV1)
  
          res = glance._is_image_available(ctx, img)
          self.assertTrue(res)
  
  
  class TestShow(test.NoDBTestCase):
  
      """Tests the show method of the GlanceImageService."""
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_show_success(self, is_avail_mock, trans_from_mock):
          is_avail_mock.return_value = True
          trans_from_mock.return_value = {'mock': mock.sentinel.trans_from}
          client = mock.MagicMock()
          client.call.return_value = {}
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          info = service.show(ctx, mock.sentinel.image_id)
  
          client.call.assert_called_once_with(ctx, 1, 'get',
                                              mock.sentinel.image_id)
          is_avail_mock.assert_called_once_with(ctx, {})
          trans_from_mock.assert_called_once_with({}, include_locations=False)
          self.assertIn('mock', info)
          self.assertEqual(mock.sentinel.trans_from, info['mock'])
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_show_not_available(self, is_avail_mock, trans_from_mock):
          is_avail_mock.return_value = False
          client = mock.MagicMock()
          client.call.return_value = mock.sentinel.images_0
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
  
          with testtools.ExpectedException(exception.ImageNotFound):
              service.show(ctx, mock.sentinel.image_id)
  
          client.call.assert_called_once_with(ctx, 1, 'get',
                                              mock.sentinel.image_id)
          is_avail_mock.assert_called_once_with(ctx, mock.sentinel.images_0)
          self.assertFalse(trans_from_mock.called)
  
      @mock.patch('nova.image.glance._reraise_translated_image_exception')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_show_client_failure(self, is_avail_mock, trans_from_mock,
                                   reraise_mock):
          raised = exception.ImageNotAuthorized(image_id=123)
          client = mock.MagicMock()
          client.call.side_effect = glanceclient.exc.Forbidden
          ctx = mock.sentinel.ctx
          reraise_mock.side_effect = raised
          service = glance.GlanceImageService(client)
  
          with testtools.ExpectedException(exception.ImageNotAuthorized):
              service.show(ctx, mock.sentinel.image_id)
              client.call.assert_called_once_with(ctx, 1, 'get',
                                                  mock.sentinel.image_id)
              self.assertFalse(is_avail_mock.called)
              self.assertFalse(trans_from_mock.called)
              reraise_mock.assert_called_once_with(mock.sentinel.image_id)
  
      @mock.patch('nova.image.glance._is_image_available')
      def test_show_queued_image_without_some_attrs(self, is_avail_mock):
          is_avail_mock.return_value = True
          client = mock.MagicMock()
  
          # fake image cls without disk_format, container_format, name attributes
          class fake_image_cls(dict):
              id = 'b31aa5dd-f07a-4748-8f15-398346887584'
              deleted = False
              protected = False
              min_disk = 0
              created_at = '2014-05-20T08:16:48'
              size = 0
              status = 'queued'
              is_public = False
              min_ram = 0
              owner = '980ec4870033453ead65c0470a78b8a8'
              updated_at = '2014-05-20T08:16:48'
          glance_image = fake_image_cls()
          client.call.return_value = glance_image
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          image_info = service.show(ctx, glance_image.id)
          client.call.assert_called_once_with(ctx, 1, 'get',
                                              glance_image.id)
          NOVA_IMAGE_ATTRIBUTES = set(['size', 'disk_format', 'owner',
                                       'container_format', 'status', 'id',
                                       'name', 'created_at', 'updated_at',
                                       'deleted', 'deleted_at', 'checksum',
                                       'min_disk', 'min_ram', 'is_public',
                                       'properties'])
  
          self.assertEqual(NOVA_IMAGE_ATTRIBUTES, set(image_info.keys()))
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_include_locations_success(self, avail_mock, trans_from_mock):
          locations = [mock.sentinel.loc1]
          avail_mock.return_value = True
          trans_from_mock.return_value = {'locations': locations}
  
          client = mock.Mock()
          client.call.return_value = mock.sentinel.image
          service = glance.GlanceImageService(client)
          ctx = mock.sentinel.ctx
          image_id = mock.sentinel.image_id
          info = service.show(ctx, image_id, include_locations=True)
  
          client.call.assert_called_once_with(ctx, 2, 'get', image_id)
          avail_mock.assert_called_once_with(ctx, mock.sentinel.image)
          trans_from_mock.assert_called_once_with(mock.sentinel.image,
                                                  include_locations=True)
          self.assertIn('locations', info)
          self.assertEqual(locations, info['locations'])
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_include_direct_uri_success(self, avail_mock, trans_from_mock):
          locations = [mock.sentinel.loc1]
          avail_mock.return_value = True
          trans_from_mock.return_value = {'locations': locations,
                                          'direct_uri': mock.sentinel.duri}
  
          client = mock.Mock()
          client.call.return_value = mock.sentinel.image
          service = glance.GlanceImageService(client)
          ctx = mock.sentinel.ctx
          image_id = mock.sentinel.image_id
          info = service.show(ctx, image_id, include_locations=True)
  
          client.call.assert_called_once_with(ctx, 2, 'get', image_id)
          expected = locations
          expected.append({'url': mock.sentinel.duri, 'metadata': {}})
          self.assertIn('locations', info)
          self.assertEqual(expected, info['locations'])
  
  
  class TestDetail(test.NoDBTestCase):
  
      """Tests the detail method of the GlanceImageService."""
  
      @mock.patch('nova.image.glance._extract_query_params')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_detail_success_available(self, is_avail_mock, trans_from_mock,
                                        ext_query_mock):
          params = {}
          is_avail_mock.return_value = True
          ext_query_mock.return_value = params
          trans_from_mock.return_value = mock.sentinel.trans_from
          client = mock.MagicMock()
          client.call.return_value = [mock.sentinel.images_0]
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          images = service.detail(ctx, **params)
  
          client.call.assert_called_once_with(ctx, 1, 'list')
          is_avail_mock.assert_called_once_with(ctx, mock.sentinel.images_0)
          trans_from_mock.assert_called_once_with(mock.sentinel.images_0)
          self.assertEqual([mock.sentinel.trans_from], images)
  
      @mock.patch('nova.image.glance._extract_query_params')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_detail_success_unavailable(self, is_avail_mock, trans_from_mock,
                                          ext_query_mock):
          params = {}
          is_avail_mock.return_value = False
          ext_query_mock.return_value = params
          trans_from_mock.return_value = mock.sentinel.trans_from
          client = mock.MagicMock()
          client.call.return_value = [mock.sentinel.images_0]
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          images = service.detail(ctx, **params)
  
          client.call.assert_called_once_with(ctx, 1, 'list')
          is_avail_mock.assert_called_once_with(ctx, mock.sentinel.images_0)
          self.assertFalse(trans_from_mock.called)
          self.assertEqual([], images)
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_detail_params_passed(self, is_avail_mock, _trans_from_mock):
          client = mock.MagicMock()
          client.call.return_value = [mock.sentinel.images_0]
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          service.detail(ctx, page_size=5, limit=10)
  
          expected_filters = {
              'is_public': 'none'
          }
          client.call.assert_called_once_with(ctx, 1, 'list',
                                              filters=expected_filters,
                                              page_size=5,
                                              limit=10)
  
      @mock.patch('nova.image.glance._reraise_translated_exception')
      @mock.patch('nova.image.glance._extract_query_params')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._is_image_available')
      def test_detail_client_failure(self, is_avail_mock, trans_from_mock,
                                     ext_query_mock, reraise_mock):
          params = {}
          ext_query_mock.return_value = params
          raised = exception.Forbidden()
          client = mock.MagicMock()
          client.call.side_effect = glanceclient.exc.Forbidden
          ctx = mock.sentinel.ctx
          reraise_mock.side_effect = raised
          service = glance.GlanceImageService(client)
  
          with testtools.ExpectedException(exception.Forbidden):
              service.detail(ctx, **params)
  
          client.call.assert_called_once_with(ctx, 1, 'list')
          self.assertFalse(is_avail_mock.called)
          self.assertFalse(trans_from_mock.called)
          reraise_mock.assert_called_once_with()
  
  
  class TestCreate(test.NoDBTestCase):
  
      """Tests the create method of the GlanceImageService."""
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._translate_to_glance')
      def test_create_success(self, trans_to_mock, trans_from_mock):
          translated = {
              'image_id': mock.sentinel.image_id
          }
          trans_to_mock.return_value = translated
          trans_from_mock.return_value = mock.sentinel.trans_from
          image_mock = mock.MagicMock(spec=dict)
          client = mock.MagicMock()
          client.call.return_value = mock.sentinel.image_meta
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          image_meta = service.create(ctx, image_mock)
  
          trans_to_mock.assert_called_once_with(image_mock)
          client.call.assert_called_once_with(ctx, 1, 'create',
                                              image_id=mock.sentinel.image_id)
          trans_from_mock.assert_called_once_with(mock.sentinel.image_meta)
  
          self.assertEqual(mock.sentinel.trans_from, image_meta)
  
          # Now verify that if we supply image data to the call,
          # that the client is also called with the data kwarg
          client.reset_mock()
          image_meta = service.create(ctx, image_mock, data=mock.sentinel.data)
  
          client.call.assert_called_once_with(ctx, 1, 'create',
                                              image_id=mock.sentinel.image_id,
                                              data=mock.sentinel.data)
  
      @mock.patch('nova.image.glance._reraise_translated_exception')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._translate_to_glance')
      def test_create_client_failure(self, trans_to_mock, trans_from_mock,
                                     reraise_mock):
          translated = {}
          trans_to_mock.return_value = translated
          image_mock = mock.MagicMock(spec=dict)
          raised = exception.Invalid()
          client = mock.MagicMock()
          client.call.side_effect = glanceclient.exc.BadRequest
          ctx = mock.sentinel.ctx
          reraise_mock.side_effect = raised
          service = glance.GlanceImageService(client)
  
          self.assertRaises(exception.Invalid, service.create, ctx, image_mock)
          trans_to_mock.assert_called_once_with(image_mock)
          self.assertFalse(trans_from_mock.called)
  
  
  class TestUpdate(test.NoDBTestCase):
  
      """Tests the update method of the GlanceImageService."""
  
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._translate_to_glance')
      def test_update_success(self, trans_to_mock, trans_from_mock):
          translated = {
              'id': mock.sentinel.image_id,
              'name': mock.sentinel.name
          }
          trans_to_mock.return_value = translated
          trans_from_mock.return_value = mock.sentinel.trans_from
          image_mock = mock.MagicMock(spec=dict)
          client = mock.MagicMock()
          client.call.return_value = mock.sentinel.image_meta
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          image_meta = service.update(ctx, mock.sentinel.image_id, image_mock)
  
          trans_to_mock.assert_called_once_with(image_mock)
          # Verify that the 'id' element has been removed as a kwarg to
          # the call to glanceclient's update (since the image ID is
          # supplied as a positional arg), and that the
          # purge_props default is True.
          client.call.assert_called_once_with(ctx, 1, 'update',
                                              mock.sentinel.image_id,
                                              name=mock.sentinel.name,
                                              purge_props=True)
          trans_from_mock.assert_called_once_with(mock.sentinel.image_meta)
          self.assertEqual(mock.sentinel.trans_from, image_meta)
  
          # Now verify that if we supply image data to the call,
          # that the client is also called with the data kwarg
          client.reset_mock()
          image_meta = service.update(ctx, mock.sentinel.image_id,
                                      image_mock, data=mock.sentinel.data)
  
          client.call.assert_called_once_with(ctx, 1, 'update',
                                              mock.sentinel.image_id,
                                              name=mock.sentinel.name,
                                              purge_props=True,
                                              data=mock.sentinel.data)
  
      @mock.patch('nova.image.glance._reraise_translated_image_exception')
      @mock.patch('nova.image.glance._translate_from_glance')
      @mock.patch('nova.image.glance._translate_to_glance')
      def test_update_client_failure(self, trans_to_mock, trans_from_mock,
                                     reraise_mock):
          translated = {
              'name': mock.sentinel.name
          }
          trans_to_mock.return_value = translated
          trans_from_mock.return_value = mock.sentinel.trans_from
          image_mock = mock.MagicMock(spec=dict)
          raised = exception.ImageNotAuthorized(image_id=123)
          client = mock.MagicMock()
          client.call.side_effect = glanceclient.exc.Forbidden
          ctx = mock.sentinel.ctx
          reraise_mock.side_effect = raised
          service = glance.GlanceImageService(client)
  
          self.assertRaises(exception.ImageNotAuthorized,
                            service.update, ctx, mock.sentinel.image_id,
                            image_mock)
          client.call.assert_called_once_with(ctx, 1, 'update',
                                              mock.sentinel.image_id,
                                              purge_props=True,
                                              name=mock.sentinel.name)
          self.assertFalse(trans_from_mock.called)
          reraise_mock.assert_called_once_with(mock.sentinel.image_id)
  
  
  class TestDelete(test.NoDBTestCase):
  
      """Tests the delete method of the GlanceImageService."""
  
      def test_delete_success(self):
          client = mock.MagicMock()
          client.call.return_value = True
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          service.delete(ctx, mock.sentinel.image_id)
          client.call.assert_called_once_with(ctx, 1, 'delete',
                                              mock.sentinel.image_id)
  
      def test_delete_client_failure(self):
          client = mock.MagicMock()
          client.call.side_effect = glanceclient.exc.NotFound
          ctx = mock.sentinel.ctx
          service = glance.GlanceImageService(client)
          self.assertRaises(exception.ImageNotFound, service.delete, ctx,
                            mock.sentinel.image_id)
  
  
  class TestGlanceClientWrapper(test.NoDBTestCase):
  
      def setUp(self):
          super(TestGlanceClientWrapper, self).setUp()
          # host1 has no scheme, which is http by default
          self.flags(api_servers=['host1:9292', 'https://host2:9293',
              'http://host3:9294'], group='glance')
  
          # Make the test run fast
          def _fake_sleep(secs):
              pass
          self.stubs.Set(time, 'sleep', _fake_sleep)
  
      def test_headers_passed_glanceclient(self):
          auth_token = 'auth_token'
          ctxt = context.RequestContext('fake', 'fake', auth_token=auth_token)
          fake_host = 'host4'
          fake_port = 9295
          fake_use_ssl = False
  
          def _get_fake_glanceclient(version, endpoint, **params):
              fake_client = glance_stubs.StubGlanceClient(version,
                                         endpoint, **params)
              self.assertIsNotNone(fake_client.auth_token)
              self.assertIsNotNone(fake_client.identity_headers)
              self.assertEqual(fake_client.identity_header['X-Auth_Token'],
                               auth_token)
              self.assertEqual(fake_client.identity_header['X-User-Id'], 'fake')
              self.assertIsNone(fake_client.identity_header['X-Roles'])
              self.assertIsNone(fake_client.identity_header['X-Tenant-Id'])
              self.assertIsNone(fake_client.identity_header['X-Service-Catalog'])
              self.assertEqual(fake_client.
                               identity_header['X-Identity-Status'],
                               'Confirmed')
  
          self.stubs.Set(glanceclient.Client, '__init__',
                         _get_fake_glanceclient)
  
          glance._create_glance_client(ctxt, fake_host, fake_port, fake_use_ssl)
  
      def test_static_client_without_retries(self):
          self.flags(num_retries=0, group='glance')
  
          ctxt = context.RequestContext('fake', 'fake')
          fake_host = 'host4'
          fake_port = 9295
          fake_use_ssl = False
  
          info = {'num_calls': 0}
  
          def _fake_create_glance_client(context, host, port, use_ssl, version):
              self.assertEqual(host, fake_host)
              self.assertEqual(port, fake_port)
              self.assertEqual(use_ssl, fake_use_ssl)
              return _create_failing_glance_client(info)
  
          self.stubs.Set(glance, '_create_glance_client',
                  _fake_create_glance_client)
  
          client = glance.GlanceClientWrapper(context=ctxt,
                  host=fake_host, port=fake_port, use_ssl=fake_use_ssl)
          self.assertRaises(exception.GlanceConnectionFailed,
                  client.call, ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 1)
  
      def test_default_client_without_retries(self):
          self.flags(num_retries=0, group='glance')
  
          ctxt = context.RequestContext('fake', 'fake')
  
          info = {'num_calls': 0,
                  'host': 'host1',
                  'port': 9292,
                  'use_ssl': False}
  
          # Leave the list in a known-order
          def _fake_shuffle(servers):
              pass
  
          def _fake_create_glance_client(context, host, port, use_ssl, version):
              self.assertEqual(host, info['host'])
              self.assertEqual(port, info['port'])
              self.assertEqual(use_ssl, info['use_ssl'])
              return _create_failing_glance_client(info)
  
          self.stubs.Set(random, 'shuffle', _fake_shuffle)
          self.stubs.Set(glance, '_create_glance_client',
                  _fake_create_glance_client)
  
          client = glance.GlanceClientWrapper()
          client2 = glance.GlanceClientWrapper()
          self.assertRaises(exception.GlanceConnectionFailed,
                  client.call, ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 1)
  
          info = {'num_calls': 0,
                  'host': 'host2',
                  'port': 9293,
                  'use_ssl': True}
  
          def _fake_shuffle2(servers):
              # fake shuffle in a known manner
              servers.append(servers.pop(0))
  
          self.stubs.Set(random, 'shuffle', _fake_shuffle2)
  
          self.assertRaises(exception.GlanceConnectionFailed,
                  client2.call, ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 1)
  
      def test_static_client_with_retries(self):
          self.flags(num_retries=1, group='glance')
  
          ctxt = context.RequestContext('fake', 'fake')
          fake_host = 'host4'
          fake_port = 9295
          fake_use_ssl = False
  
          info = {'num_calls': 0}
  
          def _fake_create_glance_client(context, host, port, use_ssl, version):
              self.assertEqual(host, fake_host)
              self.assertEqual(port, fake_port)
              self.assertEqual(use_ssl, fake_use_ssl)
              return _create_failing_glance_client(info)
  
          self.stubs.Set(glance, '_create_glance_client',
                  _fake_create_glance_client)
  
          client = glance.GlanceClientWrapper(context=ctxt,
                  host=fake_host, port=fake_port, use_ssl=fake_use_ssl)
          client.call(ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 2)
  
      def test_default_client_with_retries(self):
          self.flags(num_retries=1, group='glance')
  
          ctxt = context.RequestContext('fake', 'fake')
  
          info = {'num_calls': 0,
                  'host0': 'host1',
                  'port0': 9292,
                  'use_ssl0': False,
                  'host1': 'host2',
                  'port1': 9293,
                  'use_ssl1': True}
  
          # Leave the list in a known-order
          def _fake_shuffle(servers):
              pass
  
          def _fake_create_glance_client(context, host, port, use_ssl, version):
              attempt = info['num_calls']
              self.assertEqual(host, info['host%s' % attempt])
              self.assertEqual(port, info['port%s' % attempt])
              self.assertEqual(use_ssl, info['use_ssl%s' % attempt])
              return _create_failing_glance_client(info)
  
          self.stubs.Set(random, 'shuffle', _fake_shuffle)
          self.stubs.Set(glance, '_create_glance_client',
                  _fake_create_glance_client)
  
          client = glance.GlanceClientWrapper()
          client2 = glance.GlanceClientWrapper()
          client.call(ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 2)
  
          def _fake_shuffle2(servers):
              # fake shuffle in a known manner
              servers.append(servers.pop(0))
  
          self.stubs.Set(random, 'shuffle', _fake_shuffle2)
  
          info = {'num_calls': 0,
                  'host0': 'host2',
                  'port0': 9293,
                  'use_ssl0': True,
                  'host1': 'host3',
                  'port1': 9294,
                  'use_ssl1': False}
  
          client2.call(ctxt, 1, 'get', 'meow')
          self.assertEqual(info['num_calls'], 2)
  
  
  class TestGlanceUrl(test.NoDBTestCase):
  
      def test_generate_glance_http_url(self):
          generated_url = glance.generate_glance_url()
          glance_host = CONF.glance.host
          # ipv6 address, need to wrap it with '[]'
          if utils.is_valid_ipv6(glance_host):
              glance_host = '[%s]' % glance_host
          http_url = "http://%s:%d" % (glance_host, CONF.glance.port)
          self.assertEqual(generated_url, http_url)
  
      def test_generate_glance_https_url(self):
          self.flags(protocol="https", group='glance')
          generated_url = glance.generate_glance_url()
          glance_host = CONF.glance.host
          # ipv6 address, need to wrap it with '[]'
          if utils.is_valid_ipv6(glance_host):
              glance_host = '[%s]' % glance_host
          https_url = "https://%s:%d" % (glance_host, CONF.glance.port)
          self.assertEqual(generated_url, https_url)
  
  
  class TestGlanceApiServers(test.NoDBTestCase):
  
      def test_get_ipv4_api_servers(self):
          self.flags(api_servers=['10.0.1.1:9292',
                                  'https://10.0.0.1:9293',
                                  'http://10.0.2.2:9294'], group='glance')
          glance_host = ['10.0.1.1', '10.0.0.1',
                          '10.0.2.2']
          api_servers = glance.get_api_servers()
          i = 0
          for server in api_servers:
              i += 1
              self.assertIn(server[0], glance_host)
              if i > 2:
                  break
  
      # Python 2.6 can not parse ipv6 address correctly
      @testtools.skipIf(sys.version_info < (2, 7), "py27 or greater only")
      def test_get_ipv6_api_servers(self):
          self.flags(api_servers=['[2001:2012:1:f101::1]:9292',
                                  'https://[2010:2013:1:f122::1]:9293',
                                  'http://[2001:2011:1:f111::1]:9294'],
                     group='glance')
          glance_host = ['2001:2012:1:f101::1', '2010:2013:1:f122::1',
                          '2001:2011:1:f111::1']
          api_servers = glance.get_api_servers()
          i = 0
          for server in api_servers:
              i += 1
              self.assertIn(server[0], glance_host)
              if i > 2:
                  break
  
  
  class TestUpdateGlanceImage(test.NoDBTestCase):
      @mock.patch('nova.image.glance.GlanceImageService')
      def test_start(self, mock_glance_image_service):
          consumer = glance.UpdateGlanceImage(
              'context', 'id', 'metadata', 'stream')
  
          with mock.patch.object(glance, 'get_remote_image_service') as a_mock:
              a_mock.return_value = (mock_glance_image_service, 'image_id')
  
              consumer.start()
              mock_glance_image_service.update.assert_called_with(
                  'context', 'image_id', 'metadata', 'stream', purge_props=False)
  