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
21
import glance.common.exception as glance_exception
23
from nova.tests.api.openstack import fakes
24
from nova import context
25
from nova import exception
26
from nova.image import glance
28
from nova.tests.glance import stubs as glance_stubs
31
class NullWriter(object):
32
"""Used to test ImageService.get which takes a writer object"""
34
def write(self, *arg, **kwargs):
38
class TestGlanceSerializer(test.TestCase):
39
def test_serialize(self):
40
metadata = {'name': 'image1',
44
'prop1': 'propvalue1',
50
'block_device_mapping': [
51
{'virtual_device': 'fake',
52
'device_name': '/dev/fake'},
53
{'virtual_device': 'ephemeral0',
54
'device_name': '/dev/fake0'}]}}
56
converted_expected = {
61
'prop1': 'propvalue1',
63
'[{"device": "bbb", "virtual": "aaa"}, '
64
'{"device": "yyy", "virtual": "xxx"}]',
65
'block_device_mapping':
66
'[{"virtual_device": "fake", "device_name": "/dev/fake"}, '
67
'{"virtual_device": "ephemeral0", '
68
'"device_name": "/dev/fake0"}]'}}
69
converted = glance._convert_to_string(metadata)
70
self.assertEqual(converted, converted_expected)
71
self.assertEqual(glance._convert_from_string(converted), metadata)
74
class TestGlanceImageService(test.TestCase):
76
Tests the Glance image service.
78
At a high level, the translations involved are:
80
1. Glance -> ImageService - This is needed so we can support
81
multple ImageServices (Glance, Local, etc)
83
2. ImageService -> API - This is needed so we can support multple
87
NOW_GLANCE_OLD_FORMAT = "2010-10-11T10:30:22"
88
NOW_GLANCE_FORMAT = "2010-10-11T10:30:22.000000"
89
NOW_DATETIME = datetime.datetime(2010, 10, 11, 10, 30, 22)
92
super(TestGlanceImageService, self).setUp()
93
fakes.stub_out_compute_api_snapshot(self.stubs)
94
client = glance_stubs.StubGlanceClient()
95
self.service = glance.GlanceImageService(client=client)
96
self.context = context.RequestContext('fake', 'fake', auth_token=True)
97
self.service.delete_all()
100
def _make_fixture(**kwargs):
101
fixture = {'name': None,
105
fixture.update(kwargs)
108
def _make_datetime_fixture(self):
109
return self._make_fixture(created_at=self.NOW_GLANCE_FORMAT,
110
updated_at=self.NOW_GLANCE_FORMAT,
111
deleted_at=self.NOW_GLANCE_FORMAT)
113
def test_create_with_instance_id(self):
114
"""Ensure instance_id is persisted as an image-property"""
115
fixture = {'name': 'test image',
117
'properties': {'instance_id': '42', 'user_id': 'fake'}}
119
image_id = self.service.create(self.context, fixture)['id']
120
image_meta = self.service.show(self.context, image_id)
123
'name': 'test image',
129
'container_format': None,
131
'created_at': self.NOW_DATETIME,
132
'updated_at': self.NOW_DATETIME,
136
'properties': {'instance_id': '42', 'user_id': 'fake'},
138
self.assertDictMatch(image_meta, expected)
140
image_metas = self.service.detail(self.context)
141
self.assertDictMatch(image_metas[0], expected)
143
def test_create_without_instance_id(self):
145
Ensure we can create an image without having to specify an
146
instance_id. Public images are an example of an image not tied to an
149
fixture = {'name': 'test image', 'is_public': False}
150
image_id = self.service.create(self.context, fixture)['id']
154
'name': 'test image',
160
'container_format': None,
162
'created_at': self.NOW_DATETIME,
163
'updated_at': self.NOW_DATETIME,
169
actual = self.service.show(self.context, image_id)
170
self.assertDictMatch(actual, expected)
172
def test_create(self):
173
fixture = self._make_fixture(name='test image')
174
num_images = len(self.service.index(self.context))
175
image_id = self.service.create(self.context, fixture)['id']
177
self.assertNotEquals(None, image_id)
178
self.assertEquals(num_images + 1,
179
len(self.service.index(self.context)))
181
def test_create_and_show_non_existing_image(self):
182
fixture = self._make_fixture(name='test image')
183
image_id = self.service.create(self.context, fixture)['id']
185
self.assertNotEquals(None, image_id)
186
self.assertRaises(exception.ImageNotFound,
191
def test_create_and_show_non_existing_image_by_name(self):
192
self.assertRaises(exception.ImageNotFound,
193
self.service.show_by_name,
197
def test_index(self):
198
fixture = self._make_fixture(name='test image')
199
image_id = self.service.create(self.context, fixture)['id']
200
image_metas = self.service.index(self.context)
201
expected = [{'id': image_id, 'name': 'test image'}]
202
self.assertDictListMatch(image_metas, expected)
204
def test_index_default_limit(self):
208
fixture = self._make_fixture(name='TestImage %d' % (i))
209
fixtures.append(fixture)
210
ids.append(self.service.create(self.context, fixture)['id'])
212
image_metas = self.service.index(self.context)
214
for meta in image_metas:
215
expected = {'id': 'DONTCARE',
216
'name': 'TestImage %d' % (i)}
217
self.assertDictMatch(meta, expected)
220
def test_index_marker(self):
224
fixture = self._make_fixture(name='TestImage %d' % (i))
225
fixtures.append(fixture)
226
ids.append(self.service.create(self.context, fixture)['id'])
228
image_metas = self.service.index(self.context, marker=ids[1])
229
self.assertEquals(len(image_metas), 8)
231
for meta in image_metas:
232
expected = {'id': 'DONTCARE',
233
'name': 'TestImage %d' % (i)}
234
self.assertDictMatch(meta, expected)
237
def test_index_limit(self):
241
fixture = self._make_fixture(name='TestImage %d' % (i))
242
fixtures.append(fixture)
243
ids.append(self.service.create(self.context, fixture)['id'])
245
image_metas = self.service.index(self.context, limit=5)
246
self.assertEquals(len(image_metas), 5)
248
def test_index_marker_and_limit(self):
252
fixture = self._make_fixture(name='TestImage %d' % (i))
253
fixtures.append(fixture)
254
ids.append(self.service.create(self.context, fixture)['id'])
256
image_metas = self.service.index(self.context, marker=ids[3], limit=1)
257
self.assertEquals(len(image_metas), 1)
259
for meta in image_metas:
260
expected = {'id': ids[i],
261
'name': 'TestImage %d' % (i)}
262
self.assertDictMatch(meta, expected)
265
def test_index_invalid_marker(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
self.assertRaises(exception.Invalid, self.service.index,
274
self.context, marker='invalidmarker')
276
def test_index_private_image(self):
277
fixture = self._make_fixture(name='test image')
278
fixture['is_public'] = False
279
properties = {'owner_id': 'proj1'}
280
fixture['properties'] = properties
282
image_id = self.service.create(self.context, fixture)['id']
284
proj = self.context.project_id
285
self.context.project_id = 'proj1'
287
image_metas = self.service.index(self.context)
289
self.context.project_id = proj
291
expected = [{'id': 'DONTCARE', 'name': 'test image'}]
292
self.assertDictListMatch(image_metas, expected)
294
def test_detail_marker(self):
298
fixture = self._make_fixture(name='TestImage %d' % (i))
299
fixtures.append(fixture)
300
ids.append(self.service.create(self.context, fixture)['id'])
302
image_metas = self.service.detail(self.context, marker=ids[1])
303
self.assertEquals(len(image_metas), 8)
305
for meta in image_metas:
310
'name': 'TestImage %d' % (i),
316
'container_format': None,
318
'created_at': self.NOW_DATETIME,
319
'updated_at': self.NOW_DATETIME,
324
self.assertDictMatch(meta, expected)
327
def test_detail_limit(self):
331
fixture = self._make_fixture(name='TestImage %d' % (i))
332
fixtures.append(fixture)
333
ids.append(self.service.create(self.context, fixture)['id'])
335
image_metas = self.service.detail(self.context, limit=5)
336
self.assertEquals(len(image_metas), 5)
338
def test_detail_marker_and_limit(self):
342
fixture = self._make_fixture(name='TestImage %d' % (i))
343
fixtures.append(fixture)
344
ids.append(self.service.create(self.context, fixture)['id'])
346
image_metas = self.service.detail(self.context, marker=ids[3], limit=5)
347
self.assertEquals(len(image_metas), 5)
349
for meta in image_metas:
354
'name': 'TestImage %d' % (i),
360
'container_format': None,
362
'created_at': self.NOW_DATETIME,
363
'updated_at': self.NOW_DATETIME,
367
self.assertDictMatch(meta, expected)
370
def test_detail_invalid_marker(self):
374
fixture = self._make_fixture(name='TestImage %d' % (i))
375
fixtures.append(fixture)
376
ids.append(self.service.create(self.context, fixture)['id'])
378
self.assertRaises(exception.Invalid, self.service.detail,
379
self.context, marker='invalidmarker')
381
def test_update(self):
382
fixture = self._make_fixture(name='test image')
383
image_id = self.service.create(self.context, fixture)['id']
384
fixture['name'] = 'new image name'
385
self.service.update(self.context, image_id, fixture)
387
new_image_data = self.service.show(self.context, image_id)
388
self.assertEquals('new image name', new_image_data['name'])
390
def test_delete(self):
391
fixture1 = self._make_fixture(name='test image 1')
392
fixture2 = self._make_fixture(name='test image 2')
393
fixtures = [fixture1, fixture2]
395
num_images = len(self.service.index(self.context))
396
self.assertEquals(0, num_images, str(self.service.index(self.context)))
399
for fixture in fixtures:
400
new_id = self.service.create(self.context, fixture)['id']
403
num_images = len(self.service.index(self.context))
404
self.assertEquals(2, num_images, str(self.service.index(self.context)))
406
self.service.delete(self.context, ids[0])
408
num_images = len(self.service.index(self.context))
409
self.assertEquals(1, num_images)
411
def test_delete_not_by_owner(self):
412
# this test is only relevant for deprecated auth mode
413
self.flags(auth_strategy='deprecated')
415
fixture = self._make_fixture(name='test image')
416
properties = {'project_id': 'proj1'}
417
fixture['properties'] = properties
419
num_images = len(self.service.index(self.context))
420
self.assertEquals(0, num_images)
422
image_id = self.service.create(self.context, fixture)['id']
423
num_images = len(self.service.index(self.context))
424
self.assertEquals(1, num_images)
426
proj_id = self.context.project_id
427
self.context.project_id = 'proj2'
429
self.assertRaises(exception.NotAuthorized, self.service.delete,
430
self.context, image_id)
432
self.context.project_id = proj_id
434
num_images = len(self.service.index(self.context))
435
self.assertEquals(1, num_images)
437
def test_show_passes_through_to_client(self):
438
fixture = self._make_fixture(name='image1', is_public=True)
439
image_id = self.service.create(self.context, fixture)['id']
441
image_meta = self.service.show(self.context, image_id)
450
'container_format': None,
452
'created_at': self.NOW_DATETIME,
453
'updated_at': self.NOW_DATETIME,
459
self.assertEqual(image_meta, expected)
461
def test_show_raises_when_no_authtoken_in_the_context(self):
462
fixture = self._make_fixture(name='image1',
464
properties={'one': 'two'})
465
image_id = self.service.create(self.context, fixture)['id']
466
self.context.auth_token = False
467
self.assertRaises(exception.ImageNotFound,
472
def test_show_raises_on_missing_credential(self):
473
def raise_missing_credentials(*args, **kwargs):
474
raise glance_exception.MissingCredentialError()
476
self.stubs.Set(glance_stubs.StubGlanceClient, 'get_image_meta',
477
raise_missing_credentials)
478
self.assertRaises(exception.ImageNotAuthorized,
483
def test_detail_passes_through_to_client(self):
484
fixture = self._make_fixture(name='image10', is_public=True)
485
image_id = self.service.create(self.context, fixture)['id']
486
image_metas = self.service.detail(self.context)
496
'container_format': None,
498
'created_at': self.NOW_DATETIME,
499
'updated_at': self.NOW_DATETIME,
506
self.assertEqual(image_metas, expected)
508
def test_show_makes_datetimes(self):
509
fixture = self._make_datetime_fixture()
510
image_id = self.service.create(self.context, fixture)['id']
511
image_meta = self.service.show(self.context, image_id)
512
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
513
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
515
def test_detail_makes_datetimes(self):
516
fixture = self._make_datetime_fixture()
517
self.service.create(self.context, fixture)
518
image_meta = self.service.detail(self.context)[0]
519
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
520
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
522
def test_get_makes_datetimes(self):
523
fixture = self._make_datetime_fixture()
524
image_id = self.service.create(self.context, fixture)['id']
525
writer = NullWriter()
526
image_meta = self.service.get(self.context, image_id, writer)
527
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
528
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
530
def test_get_with_retries(self):
533
class GlanceBusyException(Exception):
536
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
537
"""A client that fails the first time, then succeeds."""
538
def get_image(self, image_id):
541
raise GlanceBusyException()
545
client = MyGlanceStubClient()
546
service = glance.GlanceImageService(client=client)
547
image_id = 1 # doesn't matter
548
writer = NullWriter()
550
# When retries are disabled, we should get an exception
551
self.flags(glance_num_retries=0)
552
self.assertRaises(GlanceBusyException, service.get, self.context,
555
# Now lets enable retries. No exception should happen now.
556
self.flags(glance_num_retries=1)
557
service.get(self.context, image_id, writer)
559
def test_client_raises_forbidden(self):
560
class MyGlanceStubClient(glance_stubs.StubGlanceClient):
561
"""A client that fails the first time, then succeeds."""
562
def get_image(self, image_id):
563
raise glance_exception.Forbidden()
565
client = MyGlanceStubClient()
566
service = glance.GlanceImageService(client=client)
567
image_id = 1 # doesn't matter
568
writer = NullWriter()
569
self.assertRaises(exception.ImageNotAuthorized, service.get,
570
self.context, image_id, writer)
572
def test_glance_client_image_id(self):
573
fixture = self._make_fixture(name='test image')
574
image_id = self.service.create(self.context, fixture)['id']
575
client, same_id = glance.get_glance_client(self.context, image_id)
576
self.assertEquals(same_id, image_id)
578
def test_glance_client_image_ref(self):
579
fixture = self._make_fixture(name='test image')
580
image_id = self.service.create(self.context, fixture)['id']
581
image_url = 'http://foo/%s' % image_id
582
client, same_id = glance.get_glance_client(self.context, image_url)
583
self.assertEquals(same_id, image_id)
added
removed
.pc/upstream/0001-fix-bug-where-nova-ignores-glance-host-in-imageref.patch/nova/tests/image/test_glance.py
