¡@

OpenStack Study: test_glance.py

OpenStack Index

**** CubicPower OpenStack Study ****

# 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 glanceclient.exc

import glanceclient.v2.client

from glanceclient.v2.client import Client as glanceclient_v2

from oslo.config import cfg

from cinder import context

from cinder import exception

from cinder.image import glance

from cinder import test

from cinder.tests.glance import stubs as glance_stubs

CONF = cfg.CONF

**** CubicPower OpenStack Study ****

class NullWriter(object):

"""Used to test ImageService.get which takes a writer object."""

**** CubicPower OpenStack Study ****

def write(self, *arg, **kwargs):

pass

**** CubicPower OpenStack Study ****

class TestGlanceSerializer(test.TestCase):

**** CubicPower OpenStack Study ****

def test_serialize(self):

metadata = {'name': 'image1',

'is_public': True,

'foo': 'bar',

'properties': {

'prop1': 'propvalue1',

'block_device_mapping': [

{'virtual_device': 'fake',

'device_name': '/dev/fake'},

{'virtual_device': 'ephemeral0',

'device_name': '/dev/fake0'}]}}

converted_expected = {

'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)

**** CubicPower OpenStack Study ****

class TestGlanceImageService(test.TestCase):

"""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)

"""

NOW_GLANCE_OLD_FORMAT = "2010-10-11T10:30:22"

NOW_GLANCE_FORMAT = "2010-10-11T10:30:22.000000"

class tzinfo(datetime.tzinfo):

@staticmethod

**** CubicPower OpenStack Study ****

def setUp(self):

super(TestGlanceImageService, self).setUp()

#fakes.stub_out_compute_api_snapshot(self.stubs)

client = glance_stubs.StubGlanceClient()

self.service = self._create_image_service(client)

self.context = context.RequestContext('fake', 'fake', auth_token=True)

self.stubs.Set(glance.time, 'sleep', lambda s: None)

**** CubicPower OpenStack Study ****

def _create_image_service(self, client):

def _fake_create_glance_client(context, netloc, 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)

@staticmethod

**** CubicPower OpenStack Study ****

def _fake_create_glance_client(context, netloc, 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)

@staticmethod

**** CubicPower OpenStack Study ****

def _make_fixture(**kwargs):

fixture = {'name': None,

'properties': {},

'status': None,

'is_public': None}

fixture.update(kwargs)

return fixture

**** CubicPower OpenStack Study ****

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)

**** CubicPower OpenStack Study ****

def test_create_with_instance_id(self):

"""Ensure instance_id is persisted as an image-property."""

fixture = {'name': 'test image',

'is_public': False,

'properties': {'instance_id': '42', 'user_id': 'fake'}}

image_id = self.service.create(self.context, fixture)['id']

image_meta = self.service.show(self.context, image_id)

expected = {

'id': image_id,

'name': 'test image',

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

'deleted_at': None,

'deleted': None,

'status': None,

'properties': {'instance_id': '42', 'user_id': 'fake'},

'owner': None,

}

self.assertDictMatch(image_meta, expected)

image_metas = self.service.detail(self.context)

self.assertDictMatch(image_metas[0], expected)

**** CubicPower OpenStack Study ****

def test_create_without_instance_id(self):

"""Test Creating images without instance_id.

Ensure we can create an image without having to specify an

instance_id. Public images are an example of an image not tied to an

instance.

"""

fixture = {'name': 'test image', 'is_public': False}

image_id = self.service.create(self.context, fixture)['id']

expected = {

'id': image_id,

'name': 'test image',

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

actual = self.service.show(self.context, image_id)

self.assertDictMatch(actual, expected)

**** CubicPower OpenStack Study ****

def test_create(self):

fixture = self._make_fixture(name='test image')

num_images = len(self.service.detail(self.context))

image_id = self.service.create(self.context, fixture)['id']

self.assertIsNotNone(image_id)

self.assertEqual(num_images + 1,

len(self.service.detail(self.context)))

**** CubicPower OpenStack Study ****

def test_create_and_show_non_existing_image(self):

fixture = self._make_fixture(name='test image')

image_id = self.service.create(self.context, fixture)['id']

self.assertIsNotNone(image_id)

self.assertRaises(exception.ImageNotFound,

self.service.show,

self.context,

'bad image id')

**** CubicPower OpenStack Study ****

def test_detail_private_image(self):

fixture = self._make_fixture(name='test image')

fixture['is_public'] = False

properties = {'owner_id': 'proj1'}

fixture['properties'] = properties

self.service.create(self.context, fixture)['id']

proj = self.context.project_id

self.context.project_id = 'proj1'

image_metas = self.service.detail(self.context)

self.context.project_id = proj

self.assertEqual(1, len(image_metas))

self.assertEqual(image_metas[0]['name'], 'test image')

self.assertEqual(image_metas[0]['is_public'], False)

**** CubicPower OpenStack Study ****

def test_detail_marker(self):

fixtures = []

ids = []

for i in range(10):

fixture = self._make_fixture(name='TestImage %d' % (i))

fixtures.append(fixture)

ids.append(self.service.create(self.context, fixture)['id'])

image_metas = self.service.detail(self.context, marker=ids[1])

self.assertEqual(len(image_metas), 8)

i = 2

for meta in image_metas:

'name': 'TestImage %d' % (i),

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

self.assertDictMatch(meta, expected)

i = i + 1

**** CubicPower OpenStack Study ****

def test_detail_limit(self):

fixtures = []

ids = []

for i in range(10):

fixture = self._make_fixture(name='TestImage %d' % (i))

fixtures.append(fixture)

ids.append(self.service.create(self.context, fixture)['id'])

image_metas = self.service.detail(self.context, limit=5)

self.assertEqual(len(image_metas), 5)

**** CubicPower OpenStack Study ****

def test_detail_default_limit(self):

fixtures = []

ids = []

for i in range(10):

fixture = self._make_fixture(name='TestImage %d' % (i))

fixtures.append(fixture)

ids.append(self.service.create(self.context, fixture)['id'])

image_metas = self.service.detail(self.context)

for i, meta in enumerate(image_metas):

self.assertEqual(meta['name'], 'TestImage %d' % (i))

**** CubicPower OpenStack Study ****

def test_detail_marker_and_limit(self):

fixtures = []

ids = []

for i in range(10):

fixture = self._make_fixture(name='TestImage %d' % (i))

fixtures.append(fixture)

ids.append(self.service.create(self.context, fixture)['id'])

image_metas = self.service.detail(self.context, marker=ids[3], limit=5)

self.assertEqual(len(image_metas), 5)

i = 4

for meta in image_metas:

'name': 'TestImage %d' % (i),

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

self.assertDictMatch(meta, expected)

i = i + 1

**** CubicPower OpenStack Study ****

def test_detail_invalid_marker(self):

fixtures = []

ids = []

for i in range(10):

fixture = self._make_fixture(name='TestImage %d' % (i))

fixtures.append(fixture)

ids.append(self.service.create(self.context, fixture)['id'])

self.assertRaises(exception.Invalid, self.service.detail,

self.context, marker='invalidmarker')

**** CubicPower OpenStack Study ****

def test_update(self):

fixture = self._make_fixture(name='test image')

image = self.service.create(self.context, fixture)

image_id = image['id']

fixture['name'] = 'new image name'

self.service.update(self.context, image_id, fixture)

new_image_data = self.service.show(self.context, image_id)

self.assertEqual('new image name', new_image_data['name'])

**** CubicPower OpenStack Study ****

def test_delete(self):

fixture1 = self._make_fixture(name='test image 1')

fixture2 = self._make_fixture(name='test image 2')

fixtures = [fixture1, fixture2]

num_images = len(self.service.detail(self.context))

self.assertEqual(0, num_images)

ids = []

for fixture in fixtures:

new_id = self.service.create(self.context, fixture)['id']

ids.append(new_id)

num_images = len(self.service.detail(self.context))

self.assertEqual(2, num_images)

self.service.delete(self.context, ids[0])

num_images = len(self.service.detail(self.context))

self.assertEqual(1, num_images)

**** CubicPower OpenStack Study ****

def test_show_passes_through_to_client(self):

fixture = self._make_fixture(name='image1', is_public=True)

image_id = self.service.create(self.context, fixture)['id']

image_meta = self.service.show(self.context, image_id)

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

self.assertEqual(image_meta, expected)

**** CubicPower OpenStack Study ****

def test_show_raises_when_no_authtoken_in_the_context(self):

fixture = self._make_fixture(name='image1',

is_public=False,

properties={'one': 'two'})

image_id = self.service.create(self.context, fixture)['id']

self.context.auth_token = False

self.assertRaises(exception.ImageNotFound,

self.service.show,

self.context,

image_id)

**** CubicPower OpenStack Study ****

def test_detail_passes_through_to_client(self):

fixture = self._make_fixture(name='image10', is_public=True)

image_id = self.service.create(self.context, fixture)['id']

image_metas = self.service.detail(self.context)

'container_format': None,

'checksum': None,

'created_at': self.NOW_DATETIME,

'updated_at': self.NOW_DATETIME,

self.assertEqual(image_metas, expected)

**** CubicPower OpenStack Study ****

def test_show_makes_datetimes(self):

fixture = self._make_datetime_fixture()

image_id = self.service.create(self.context, fixture)['id']

image_meta = self.service.show(self.context, image_id)

self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)

self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)

**** CubicPower OpenStack Study ****

def test_detail_makes_datetimes(self):

fixture = self._make_datetime_fixture()

self.service.create(self.context, fixture)

image_meta = self.service.detail(self.context)[0]

self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)

self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)

**** CubicPower OpenStack Study ****

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(glance_num_retries=0)

self.assertRaises(exception.GlanceConnectionFailed,

# Now lets enable retries. No exception should happen now.

tries = [0]

self.flags(glance_num_retries=1)

service.download(self.context, image_id, writer)

**** CubicPower OpenStack Study ****

def test_client_forbidden_converts_to_imagenotauthed(self):

class MyGlanceStubClient(glance_stubs.StubGlanceClient):

"""A client that raises a Forbidden exception."""

def get(self, image_id):

raise glanceclient.exc.Forbidden(image_id)

client = MyGlanceStubClient()

service = self._create_image_service(client)

image_id = 1 # doesn't matter

writer = NullWriter()

self.assertRaises(exception.ImageNotAuthorized, service.download,

self.context, image_id, writer)

**** CubicPower OpenStack Study ****

def test_client_httpforbidden_converts_to_imagenotauthed(self):

class MyGlanceStubClient(glance_stubs.StubGlanceClient):

"""A client that raises a HTTPForbidden exception."""

def get(self, image_id):

raise glanceclient.exc.HTTPForbidden(image_id)

client = MyGlanceStubClient()

service = self._create_image_service(client)

image_id = 1 # doesn't matter

writer = NullWriter()

self.assertRaises(exception.ImageNotAuthorized, service.download,

self.context, image_id, writer)

**** CubicPower OpenStack Study ****

def test_client_notfound_converts_to_imagenotfound(self):

class MyGlanceStubClient(glance_stubs.StubGlanceClient):

"""A client that raises a NotFound exception."""

def get(self, image_id):

raise glanceclient.exc.NotFound(image_id)

client = MyGlanceStubClient()

service = self._create_image_service(client)

image_id = 1 # doesn't matter

writer = NullWriter()

self.assertRaises(exception.ImageNotFound, service.download,

self.context, image_id, writer)

**** CubicPower OpenStack Study ****

def test_client_httpnotfound_converts_to_imagenotfound(self):

class MyGlanceStubClient(glance_stubs.StubGlanceClient):

"""A client that raises a HTTPNotFound exception."""

def get(self, image_id):

raise glanceclient.exc.HTTPNotFound(image_id)

client = MyGlanceStubClient()

service = self._create_image_service(client)

image_id = 1 # doesn't matter

writer = NullWriter()

self.assertRaises(exception.ImageNotFound, service.download,

self.context, image_id, writer)

**** CubicPower OpenStack Study ****

def test_glance_client_image_id(self):

fixture = self._make_fixture(name='test image')

image_id = self.service.create(self.context, fixture)['id']

(service, same_id) = glance.get_remote_image_service(self.context,

image_id)

self.assertEqual(same_id, image_id)

**** CubicPower OpenStack Study ****

def test_glance_client_image_ref(self):

fixture = self._make_fixture(name='test image')

image_id = self.service.create(self.context, fixture)['id']

image_url = 'http://something-less-likely/%s' % image_id

(service, same_id) = glance.get_remote_image_service(self.context,

image_url)

self.assertEqual(same_id, image_id)

self.assertEqual(service._client.netloc, 'something-less-likely')

for ipv6_url in ('[::1]', '::1', '[::1]:444'):

image_url = 'http://%s/%s' % (ipv6_url, image_id)

(service, same_id) = glance.get_remote_image_service(self.context,

image_url)

self.assertEqual(same_id, image_id)

self.assertEqual(service._client.netloc, ipv6_url)

**** CubicPower OpenStack Study ****

class TestGlanceClientVersion(test.TestCase):

"""Tests the version of the glance client generated."""

**** CubicPower OpenStack Study ****

def setUp(self):

super(TestGlanceClientVersion, self).setUp()

def fake_get_model(self):

return

self.stubs.Set(glanceclient_v2, '_get_image_model',

fake_get_model)

try:

self.stubs.Set(glanceclient_v2, '_get_member_model',

fake_get_model)

except AttributeError:

# method requires stubbing only with newer glanceclients.

pass

**** CubicPower OpenStack Study ****

def fake_get_model(self):

return

self.stubs.Set(glanceclient_v2, '_get_image_model',

fake_get_model)

try:

self.stubs.Set(glanceclient_v2, '_get_member_model',

fake_get_model)

except AttributeError:

# method requires stubbing only with newer glanceclients.

pass

**** CubicPower OpenStack Study ****

def test_glance_version_by_flag(self):

"""Test glance version set by flag is honoured."""

client_wrapper_v1 = glance.GlanceClientWrapper('fake', 'fake_host',

9292)

self.assertEqual(client_wrapper_v1.client.__module__,

'glanceclient.v1.client')

self.flags(glance_api_version=2)

client_wrapper_v2 = glance.GlanceClientWrapper('fake', 'fake_host',

9292)

self.assertEqual(client_wrapper_v2.client.__module__,

'glanceclient.v2.client')

CONF.reset()

**** CubicPower OpenStack Study ****

def test_glance_version_by_arg(self):

"""Test glance version set by arg to GlanceClientWrapper"""

client_wrapper_v1 = glance.GlanceClientWrapper('fake', 'fake_host',

9292, version=1)

self.assertEqual(client_wrapper_v1.client.__module__,

'glanceclient.v1.client')

client_wrapper_v2 = glance.GlanceClientWrapper('fake', 'fake_host',

9292, version=2)

self.assertEqual(client_wrapper_v2.client.__module__,

'glanceclient.v2.client')

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()

**** CubicPower OpenStack Study ****

def get(self, image_id):

info['num_calls'] += 1

if info['num_calls'] == 1:

raise glanceclient.exc.ServiceUnavailable('')

return {}

return MyGlanceStubClient()

**** CubicPower OpenStack Study ****

class TestGlanceImageServiceClient(test.TestCase):

**** CubicPower OpenStack Study ****

def setUp(self):

super(TestGlanceImageServiceClient, self).setUp()

self.context = context.RequestContext('fake', 'fake', auth_token=True)

self.stubs.Set(glance.time, 'sleep', lambda s: None)

**** CubicPower OpenStack Study ****

def test_create_glance_client(self):

self.flags(auth_strategy='keystone')

self.flags(glance_request_timeout=60)

class MyGlanceStubClient(object):

def __init__(inst, version, *args, **kwargs):

self.assertEqual('1', version)

self.assertEqual("http://fake_host:9292", args[0])

self.assertEqual(True, kwargs['token'])

self.assertEqual(60, kwargs['timeout'])

self.stubs.Set(glance.glanceclient, 'Client', MyGlanceStubClient)

client = glance._create_glance_client(self.context, 'fake_host:9292',

False)

self.assertIsInstance(client, MyGlanceStubClient)

**** CubicPower OpenStack Study ****

def test_create_glance_client_auth_strategy_is_not_keystone(self):

self.flags(auth_strategy='noauth')

self.flags(glance_request_timeout=60)

class MyGlanceStubClient(object):

def __init__(inst, version, *args, **kwargs):

self.assertEqual('1', version)

self.assertEqual('http://fake_host:9292', args[0])

self.assertNotIn('token', kwargs)

self.assertEqual(60, kwargs['timeout'])

self.stubs.Set(glance.glanceclient, 'Client', MyGlanceStubClient)

client = glance._create_glance_client(self.context, 'fake_host:9292',

False)

self.assertIsInstance(client, MyGlanceStubClient)

**** CubicPower OpenStack Study ****

def test_create_glance_client_glance_request_default_timeout(self):

self.flags(auth_strategy='keystone')

self.flags(glance_request_timeout=None)

class MyGlanceStubClient(object):

def __init__(inst, version, *args, **kwargs):

self.assertEqual("1", version)

self.assertEqual("http://fake_host:9292", args[0])

self.assertEqual(True, kwargs['token'])

self.assertNotIn('timeout', kwargs)

self.stubs.Set(glance.glanceclient, 'Client', MyGlanceStubClient)

client = glance._create_glance_client(self.context, 'fake_host:9292',

False)

self.assertIsInstance(client, MyGlanceStubClient)

**** CubicPower OpenStack Study ****

def tearDown(self):

self.stubs.UnsetAll()

super(TestGlanceImageServiceClient, self).tearDown()