1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2011 OpenStack LLC.
6
# Licensed under the Apache License, Version 2.0 (the "License"); you may
7
# not use this file except in compliance with the License. You may obtain
8
# a copy of the License at
10
# http://www.apache.org/licenses/LICENSE-2.0
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15
# License for the specific language governing permissions and limitations
23
import glanceclient.exc
25
from cinder import context
26
from cinder import exception
27
from cinder.image import glance
28
from cinder import test
29
from cinder.tests.api.openstack import fakes
30
from cinder.tests.glance import stubs as glance_stubs
33
class NullWriter(object):
34
"""Used to test ImageService.get which takes a writer object"""
36
def write(self, *arg, **kwargs):
40
class TestGlanceSerializer(test.TestCase):
41
def test_serialize(self):
42
metadata = {'name': 'image1',
46
'prop1': 'propvalue1',
52
'block_device_mapping': [
53
{'virtual_device': 'fake',
54
'device_name': '/dev/fake'},
55
{'virtual_device': 'ephemeral0',
56
'device_name': '/dev/fake0'}]}}
58
converted_expected = {
63
'prop1': 'propvalue1',
65
'[{"device": "bbb", "virtual": "aaa"}, '
66
'{"device": "yyy", "virtual": "xxx"}]',
67
'block_device_mapping':
68
'[{"virtual_device": "fake", "device_name": "/dev/fake"}, '
69
'{"virtual_device": "ephemeral0", '
70
'"device_name": "/dev/fake0"}]'}}
71
converted = glance._convert_to_string(metadata)
72
self.assertEqual(converted, converted_expected)
73
self.assertEqual(glance._convert_from_string(converted), metadata)
76
class TestGlanceImageService(test.TestCase):
78
Tests the Glance image service.
80
At a high level, the translations involved are:
82
1. Glance -> ImageService - This is needed so we can support
83
multple ImageServices (Glance, Local, etc)
85
2. ImageService -> API - This is needed so we can support multple
89
NOW_GLANCE_OLD_FORMAT = "2010-10-11T10:30:22"
90
NOW_GLANCE_FORMAT = "2010-10-11T10:30:22.000000"
92
class tzinfo(datetime.tzinfo):
94
def utcoffset(*args, **kwargs):
95
return datetime.timedelta()
97
NOW_DATETIME = datetime.datetime(2010, 10, 11, 10, 30, 22, tzinfo=tzinfo())
100
super(TestGlanceImageService, self).setUp()
101
#fakes.stub_out_compute_api_snapshot(self.stubs)
103
client = glance_stubs.StubGlanceClient()
104
self.service = self._create_image_service(client)
105
self.context = context.RequestContext('fake', 'fake', auth_token=True)
107
def _create_image_service(self, client):
108
def _fake_create_glance_client(context, host, port, version):
111
self.stubs.Set(glance, '_create_glance_client',
112
_fake_create_glance_client)
114
client_wrapper = glance.GlanceClientWrapper(
115
'fake', 'fake_host', 9292)
116
return glance.GlanceImageService(client=client_wrapper)
119
def _make_fixture(**kwargs):
120
fixture = {'name': None,
124
fixture.update(kwargs)
127
def _make_datetime_fixture(self):
128
return self._make_fixture(created_at=self.NOW_GLANCE_FORMAT,
129
updated_at=self.NOW_GLANCE_FORMAT,
130
deleted_at=self.NOW_GLANCE_FORMAT)
132
def test_create_with_instance_id(self):
133
"""Ensure instance_id is persisted as an image-property"""
134
fixture = {'name': 'test image',
136
'properties': {'instance_id': '42', 'user_id': 'fake'}}
138
image_id = self.service.create(self.context, fixture)['id']
139
image_meta = self.service.show(self.context, image_id)
142
'name': 'test image',
148
'container_format': None,
150
'created_at': self.NOW_DATETIME,
151
'updated_at': self.NOW_DATETIME,
155
'properties': {'instance_id': '42', 'user_id': 'fake'},
158
self.assertDictMatch(image_meta, expected)
160
image_metas = self.service.detail(self.context)
161
self.assertDictMatch(image_metas[0], expected)
163
def test_create_without_instance_id(self):
165
Ensure we can create an image without having to specify an
166
instance_id. Public images are an example of an image not tied to an
169
fixture = {'name': 'test image', 'is_public': False}
170
image_id = self.service.create(self.context, fixture)['id']
174
'name': 'test image',
180
'container_format': None,
182
'created_at': self.NOW_DATETIME,
183
'updated_at': self.NOW_DATETIME,
190
actual = self.service.show(self.context, image_id)
191
self.assertDictMatch(actual, expected)
193
def test_create(self):
194
fixture = self._make_fixture(name='test image')
195
num_images = len(self.service.detail(self.context))
196
image_id = self.service.create(self.context, fixture)['id']
198
self.assertNotEquals(None, image_id)
199
self.assertEquals(num_images + 1,
200
len(self.service.detail(self.context)))
202
def test_create_and_show_non_existing_image(self):
203
fixture = self._make_fixture(name='test image')
204
image_id = self.service.create(self.context, fixture)['id']
206
self.assertNotEquals(None, image_id)
207
self.assertRaises(exception.ImageNotFound,
212
def test_detail_private_image(self):
213
fixture = self._make_fixture(name='test image')
214
fixture['is_public'] = False
215
properties = {'owner_id': 'proj1'}
216
fixture['properties'] = properties
218
self.service.create(self.context, fixture)['id']
220
proj = self.context.project_id
221
self.context.project_id = 'proj1'
223
image_metas = self.service.detail(self.context)
225
self.context.project_id = proj
227
self.assertEqual(1, len(image_metas))
228
self.assertEqual(image_metas[0]['name'], 'test image')
229
self.assertEqual(image_metas[0]['is_public'], False)
231
def test_detail_marker(self):
235
fixture = self._make_fixture(name='TestImage %d' % (i))
236
fixtures.append(fixture)
237
ids.append(self.service.create(self.context, fixture)['id'])
239
image_metas = self.service.detail(self.context, marker=ids[1])
240
self.assertEquals(len(image_metas), 8)
242
for meta in image_metas:
247
'name': 'TestImage %d' % (i),
253
'container_format': None,
255
'created_at': self.NOW_DATETIME,
256
'updated_at': self.NOW_DATETIME,
262
self.assertDictMatch(meta, expected)
265
def test_detail_limit(self):
269
fixture = self._make_fixture(name='TestImage %d' % (i))
270
fixtures.append(fixture)
271
ids.append(self.service.create(self.context, fixture)['id'])
273
image_metas = self.service.detail(self.context, limit=5)
274
self.assertEquals(len(image_metas), 5)
276
def test_detail_default_limit(self):
280
fixture = self._make_fixture(name='TestImage %d' % (i))
281
fixtures.append(fixture)
282
ids.append(self.service.create(self.context, fixture)['id'])
284
image_metas = self.service.detail(self.context)
285
for i, meta in enumerate(image_metas):
286
self.assertEqual(meta['name'], 'TestImage %d' % (i))
288
def test_detail_marker_and_limit(self):
292
fixture = self._make_fixture(name='TestImage %d' % (i))
293
fixtures.append(fixture)
294
ids.append(self.service.create(self.context, fixture)['id'])
296
image_metas = self.service.detail(self.context, marker=ids[3], limit=5)
297
self.assertEquals(len(image_metas), 5)
299
for meta in image_metas:
304
'name': 'TestImage %d' % (i),
310
'container_format': None,
312
'created_at': self.NOW_DATETIME,
313
'updated_at': self.NOW_DATETIME,
318
self.assertDictMatch(meta, expected)
321
def test_detail_invalid_marker(self):
325
fixture = self._make_fixture(name='TestImage %d' % (i))
326
fixtures.append(fixture)
327
ids.append(self.service.create(self.context, fixture)['id'])
329
self.assertRaises(exception.Invalid, self.service.detail,
330
self.context, marker='invalidmarker')
332
def test_update(self):
333
fixture = self._make_fixture(name='test image')
334
image = self.service.create(self.context, fixture)
336
image_id = image['id']
337
fixture['name'] = 'new image name'
338
self.service.update(self.context, image_id, fixture)
340
new_image_data = self.service.show(self.context, image_id)
341
self.assertEquals('new image name', new_image_data['name'])
343
def test_delete(self):
344
fixture1 = self._make_fixture(name='test image 1')
345
fixture2 = self._make_fixture(name='test image 2')
346
fixtures = [fixture1, fixture2]
348
num_images = len(self.service.detail(self.context))
349
self.assertEquals(0, num_images)
352
for fixture in fixtures:
353
new_id = self.service.create(self.context, fixture)['id']
356
num_images = len(self.service.detail(self.context))
357
self.assertEquals(2, num_images)
359
self.service.delete(self.context, ids[0])
361
num_images = len(self.service.detail(self.context))
362
self.assertEquals(1, num_images)
364
def test_show_passes_through_to_client(self):
365
fixture = self._make_fixture(name='image1', is_public=True)
366
image_id = self.service.create(self.context, fixture)['id']
368
image_meta = self.service.show(self.context, image_id)
377
'container_format': None,
379
'created_at': self.NOW_DATETIME,
380
'updated_at': self.NOW_DATETIME,
387
self.assertEqual(image_meta, expected)
389
def test_show_raises_when_no_authtoken_in_the_context(self):
390
fixture = self._make_fixture(name='image1',
392
properties={'one': 'two'})
393
image_id = self.service.create(self.context, fixture)['id']
394
self.context.auth_token = False
395
self.assertRaises(exception.ImageNotFound,
400
def test_detail_passes_through_to_client(self):
401
fixture = self._make_fixture(name='image10', is_public=True)
402
image_id = self.service.create(self.context, fixture)['id']
403
image_metas = self.service.detail(self.context)
413
'container_format': None,
415
'created_at': self.NOW_DATETIME,
416
'updated_at': self.NOW_DATETIME,
424
self.assertEqual(image_metas, expected)
426
def test_show_makes_datetimes(self):
427
fixture = self._make_datetime_fixture()
428
image_id = self.service.create(self.context, fixture)['id']
429
image_meta = self.service.show(self.context, image_id)
430
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
431
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
433
def test_detail_makes_datetimes(self):
434
fixture = self._make_datetime_fixture()
435
self.service.create(self.context, fixture)
436
image_meta = self.service.detail(self.context)[0]
437
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
438
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
440
def test_download_with_retries(self):
443
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
444
"""A client that fails the first time, then succeeds."""
445
def get(self, image_id):
448
raise glanceclient.exc.ServiceUnavailable('')
452
client = MyGlanceStubClient()
453
service = self._create_image_service(client)
454
image_id = 1 # doesn't matter
455
writer = NullWriter()
457
# When retries are disabled, we should get an exception
458
self.flags(glance_num_retries=0)
459
self.assertRaises(exception.GlanceConnectionFailed,
460
service.download, self.context, image_id, writer)
462
# Now lets enable retries. No exception should happen now.
464
self.flags(glance_num_retries=1)
465
service.download(self.context, image_id, writer)
467
def test_client_forbidden_converts_to_imagenotauthed(self):
468
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
469
"""A client that raises a Forbidden exception."""
470
def get(self, image_id):
471
raise glanceclient.exc.Forbidden(image_id)
473
client = MyGlanceStubClient()
474
service = self._create_image_service(client)
475
image_id = 1 # doesn't matter
476
writer = NullWriter()
477
self.assertRaises(exception.ImageNotAuthorized, service.download,
478
self.context, image_id, writer)
480
def test_client_httpforbidden_converts_to_imagenotauthed(self):
481
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
482
"""A client that raises a HTTPForbidden exception."""
483
def get(self, image_id):
484
raise glanceclient.exc.HTTPForbidden(image_id)
486
client = MyGlanceStubClient()
487
service = self._create_image_service(client)
488
image_id = 1 # doesn't matter
489
writer = NullWriter()
490
self.assertRaises(exception.ImageNotAuthorized, service.download,
491
self.context, image_id, writer)
493
def test_client_notfound_converts_to_imagenotfound(self):
494
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
495
"""A client that raises a NotFound exception."""
496
def get(self, image_id):
497
raise glanceclient.exc.NotFound(image_id)
499
client = MyGlanceStubClient()
500
service = self._create_image_service(client)
501
image_id = 1 # doesn't matter
502
writer = NullWriter()
503
self.assertRaises(exception.ImageNotFound, service.download,
504
self.context, image_id, writer)
506
def test_client_httpnotfound_converts_to_imagenotfound(self):
507
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
508
"""A client that raises a HTTPNotFound exception."""
509
def get(self, image_id):
510
raise glanceclient.exc.HTTPNotFound(image_id)
512
client = MyGlanceStubClient()
513
service = self._create_image_service(client)
514
image_id = 1 # doesn't matter
515
writer = NullWriter()
516
self.assertRaises(exception.ImageNotFound, service.download,
517
self.context, image_id, writer)
519
def test_glance_client_image_id(self):
520
fixture = self._make_fixture(name='test image')
521
image_id = self.service.create(self.context, fixture)['id']
522
(service, same_id) = glance.get_remote_image_service(
523
self.context, image_id)
524
self.assertEquals(same_id, image_id)
526
def test_glance_client_image_ref(self):
527
fixture = self._make_fixture(name='test image')
528
image_id = self.service.create(self.context, fixture)['id']
529
image_url = 'http://something-less-likely/%s' % image_id
530
(service, same_id) = glance.get_remote_image_service(
531
self.context, image_url)
532
self.assertEquals(same_id, image_id)
533
self.assertEquals(service._client.host,
534
'something-less-likely')
537
def _create_failing_glance_client(info):
538
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
539
"""A client that fails the first time, then succeeds."""
540
def get(self, image_id):
541
info['num_calls'] += 1
542
if info['num_calls'] == 1:
543
raise glanceclient.exc.ServiceUnavailable('')
546
return MyGlanceStubClient()
added
removed
cinder/tests/image/test_glance.py
