openstack/deb-nova
Code Issues Proposed changes

further cleanup

Browse Source
This commit is contained in:
Brian Waldon
2011-09-06 15:16:40 -04:00
parent c87009f009
commit 35e0ae794f
8 changed files with 463 additions and 584 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
nova/api/openstack/image_metadata.py
View File

@@ -17,6 +17,7 @@
from webob import exc
from nova import exception
from nova import flags
from nova import image
from nova import utils
@@ -33,21 +34,22 @@ class Controller(object):
def __init__(self):
self.image_service = image.get_default_image_service()
def _get_metadata(self, context, image_id, image=None):
if not image:
image = self.image_service.show(context, image_id)
metadata = image.get('properties', {})
return metadata
def _get_image(self, context, image_id):
try:
return self.image_service.show(context, image_id)
except exception.NotFound:
msg = _("Image not found.")
raise exc.HTTPNotFound(explanation=msg)
def index(self, req, image_id):
"""Returns the list of metadata for a given instance"""
context = req.environ['nova.context']
metadata = self._get_metadata(context, image_id)
metadata = self._get_image(context, image_id)['properties']
return dict(metadata=metadata)
def show(self, req, image_id, id):
context = req.environ['nova.context']
metadata = self._get_metadata(context, image_id)
metadata = self._get_image(context, image_id)['properties']
if id in metadata:
return {'meta': {id: metadata[id]}}
else:
@@ -55,15 +57,13 @@ class Controller(object):
def create(self, req, image_id, body):
context = req.environ['nova.context']
img = self.image_service.show(context, image_id)
metadata = self._get_metadata(context, image_id, img)
image = self._get_image(context, image_id)
if 'metadata' in body:
for key, value in body['metadata'].iteritems():
metadata[key] = value
common.check_img_metadata_quota_limit(context, metadata)
img['properties'] = metadata
self.image_service.update(context, image_id, img, None)
return dict(metadata=metadata)
image['properties'][key] = value
common.check_img_metadata_quota_limit(context, image['properties'])
self.image_service.update(context, image_id, image, None)
return dict(metadata=image['properties'])
def update(self, req, image_id, id, body):
context = req.environ['nova.context']
@@ -80,32 +80,30 @@ class Controller(object):
if len(meta) > 1:
expl = _('Request body contains too many items')
raise exc.HTTPBadRequest(explanation=expl)
img = self.image_service.show(context, image_id)
metadata = self._get_metadata(context, image_id, img)
metadata[id] = meta[id]
common.check_img_metadata_quota_limit(context, metadata)
img['properties'] = metadata
self.image_service.update(context, image_id, img, None)
image = self._get_image(context, image_id)
image['properties'][id] = meta[id]
common.check_img_metadata_quota_limit(context, image['properties'])
self.image_service.update(context, image_id, image, None)
return dict(meta=meta)
def update_all(self, req, image_id, body):
context = req.environ['nova.context']
img = self.image_service.show(context, image_id)
image = self._get_image(context, image_id)
metadata = body.get('metadata', {})
common.check_img_metadata_quota_limit(context, metadata)
img['properties'] = metadata
self.image_service.update(context, image_id, img, None)
image['properties'] = metadata
self.image_service.update(context, image_id, image, None)
return dict(metadata=metadata)
def delete(self, req, image_id, id):
context = req.environ['nova.context']
img = self.image_service.show(context, image_id)
metadata = self._get_metadata(context, image_id)
if not id in metadata:
raise exc.HTTPNotFound()
metadata.pop(id)
img['properties'] = metadata
self.image_service.update(context, image_id, img, None)
image = self._get_image(context, image_id)
if not id in image['properties']:
msg = _("Invalid metadata key")
raise exc.HTTPNotFound(explanation=msg)
image['properties'].pop(id)
self.image_service.update(context, image_id, image, None)
def create_resource():

44
nova/image/glance.py
View File

@@ -104,7 +104,7 @@ class GlanceImageService(object):
images = []
for image_meta in image_metas:
if self._is_image_available(context, image_meta):
base_image_meta = self._translate_to_base(image_meta)
base_image_meta = self._translate_from_glance(image_meta)
images.append(base_image_meta)
return images
@@ -169,7 +169,7 @@ class GlanceImageService(object):
if not self._is_image_available(context, image_meta):
raise exception.ImageNotFound(image_id=image_id)
base_image_meta = self._translate_to_base(image_meta)
base_image_meta = self._translate_from_glance(image_meta)
return base_image_meta
def show_by_name(self, context, name):
@@ -193,7 +193,7 @@ class GlanceImageService(object):
for chunk in image_chunks:
data.write(chunk)
base_image_meta = self._translate_to_base(image_meta)
base_image_meta = self._translate_from_glance(image_meta)
return base_image_meta
def create(self, context, image_meta, data=None):
@@ -206,7 +206,7 @@ class GlanceImageService(object):
# Translate Base -> Service
LOG.debug(_('Creating image in Glance. Metadata passed in %s'),
image_meta)
sent_service_image_meta = self._translate_to_service(image_meta)
sent_service_image_meta = self._translate_to_glance(image_meta)
LOG.debug(_('Metadata after formatting for Glance %s'),
sent_service_image_meta)
@@ -214,7 +214,7 @@ class GlanceImageService(object):
sent_service_image_meta, data)
# Translate Service -> Base
base_image_meta = self._translate_to_base(recv_service_image_meta)
base_image_meta = self._translate_from_glance(recv_service_image_meta)
LOG.debug(_('Metadata returned from Glance formatted for Base %s'),
base_image_meta)
return base_image_meta
@@ -228,13 +228,13 @@ class GlanceImageService(object):
self._set_client_context(context)
# NOTE(vish): show is to check if image is available
self.show(context, image_id)
image_meta = _convert_to_string(image_meta)
image_meta = self._translate_to_glance(image_meta)
try:
image_meta = self.client.update_image(image_id, image_meta, data)
except glance_exception.NotFound:
raise exception.ImageNotFound(image_id=image_id)
base_image_meta = self._translate_to_base(image_meta)
base_image_meta = self._translate_from_glance(image_meta)
return base_image_meta
def delete(self, context, image_id):
@@ -257,13 +257,13 @@ class GlanceImageService(object):
pass
@classmethod
def _translate_to_service(cls, image_meta):
def _translate_to_glance(cls, image_meta):
image_meta = _convert_to_string(image_meta)
image_meta = _remove_read_only(image_meta)
return image_meta
@classmethod
def _translate_to_base(cls, image_meta):
"""Override translation to handle conversion to datetime objects."""
def _translate_from_glance(cls, image_meta):
image_meta = _limit_attributes(image_meta)
image_meta = _convert_timestamps_to_datetimes(image_meta)
image_meta = _convert_from_string(image_meta)
@@ -277,20 +277,7 @@ class GlanceImageService(object):
an auth_token.
"""
if hasattr(context, 'auth_token') and context.auth_token:
return True
properties = image_meta.get('properties', {})
if context.project_id and ('project_id' in properties):
return str(properties['project_id']) == str(context.project_id)
try:
user_id = properties['user_id']
except KeyError:
return False
return str(user_id) == str(context.user_id)
return hasattr(context, 'auth_token') and context.auth_token
# utility functions
@@ -365,3 +352,12 @@ def _limit_attributes(image_meta):
output['properties'] = image_meta.get('properties', {})
return output
def _remove_read_only(image_meta):
IMAGE_ATTRIBUTES = ['updated_at', 'created_at', 'deleted_at']
output = copy.deepcopy(image_meta)
for attr in IMAGE_ATTRIBUTES:
if attr in output:
del output[attr]
return output

125
nova/tests/api/openstack/fakes.py
View File

@@ -42,6 +42,7 @@ import nova.image.fake
from nova.image import glance
from nova.image import service
from nova.tests import fake_flags
from nova.tests.glance import stubs as glance_stubs
class Context(object):
@@ -83,7 +84,7 @@ def wsgi_app(inner_app10=None, inner_app11=None, fake_auth=True,
if fake_auth_context is not None:
ctxt = fake_auth_context
else:
ctxt = context.RequestContext('fake', 'fake')
ctxt = context.RequestContext('fake', 'fake', auth_token=True)
api10 = openstack.FaultWrapper(api_auth.InjectContext(ctxt,
limits.RateLimitingMiddleware(inner_app10)))
api11 = openstack.FaultWrapper(api_auth.InjectContext(ctxt,
@@ -177,6 +178,38 @@ def stub_out_compute_api_backup(stubs):
stubs.Set(nova.compute.API, 'backup', backup)
def _make_image_fixtures():
NOW_GLANCE_FORMAT = "2010-10-11T10:30:22"
image_id = 123
base_attrs = {'deleted': False}
fixtures = []
def add_fixture(**kwargs):
kwargs.update(base_attrs)
fixtures.append(kwargs)
# Public image
add_fixture(id=image_id, name='public image', is_public=True,
status='active', properties={'key1': 'value1'})
image_id += 1
# Snapshot for User 1
server_ref = 'http://localhost/v1.1/servers/42'
snapshot_properties = {'instance_ref': server_ref, 'user_id': 'fake'}
for status in ('queued', 'saving', 'active', 'killed'):
add_fixture(id=image_id, name='%s snapshot' % status,
is_public=False, status=status,
properties=snapshot_properties)
image_id += 1
# Image without a name
add_fixture(id=image_id, is_public=True, status='active', properties={})
return fixtures
def stub_out_glance_add_image(stubs, sent_to_glance):
"""
We return the metadata sent to glance by modifying the sent_to_glance dict
@@ -192,91 +225,11 @@ def stub_out_glance_add_image(stubs, sent_to_glance):
stubs.Set(glance_client.Client, 'add_image', fake_add_image)
def stub_out_glance(stubs, initial_fixtures=None):
class FakeGlanceClient:
def __init__(self, initial_fixtures):
self.fixtures = initial_fixtures or []
def _filter_images(self, filters=None, marker=None, limit=None):
found = True
if marker:
found = False
if limit == 0:
limit = None
fixtures = []
count = 0
for f in self.fixtures:
if limit and count >= limit:
break
if found:
fixtures.append(f)
count = count + 1
if f['id'] == marker:
found = True
return fixtures
def fake_get_images(self, filters=None, marker=None, limit=None):
fixtures = self._filter_images(filters, marker, limit)
return [dict(id=f['id'], name=f['name'])
for f in fixtures]
def fake_get_images_detailed(self, filters=None,
marker=None, limit=None):
return self._filter_images(filters, marker, limit)
def fake_get_image_meta(self, image_id):
image = self._find_image(image_id)
if image:
return copy.deepcopy(image)
raise glance_exc.NotFound
def fake_add_image(self, image_meta, data=None):
image_meta = copy.deepcopy(image_meta)
image_id = ''.join(random.choice(string.letters)
for _ in range(20))
image_meta['id'] = image_id
self.fixtures.append(image_meta)
return copy.deepcopy(image_meta)
def fake_update_image(self, image_id, image_meta, data=None):
for attr in ('created_at', 'updated_at', 'deleted_at', 'deleted'):
if attr in image_meta:
del image_meta[attr]
f = self._find_image(image_id)
if not f:
raise glance_exc.NotFound
f.update(image_meta)
return copy.deepcopy(f)
def fake_delete_image(self, image_id):
f = self._find_image(image_id)
if not f:
raise glance_exc.NotFound
self.fixtures.remove(f)
def _find_image(self, image_id):
for f in self.fixtures:
if str(f['id']) == str(image_id):
return f
return None
GlanceClient = glance_client.Client
fake = FakeGlanceClient(initial_fixtures)
stubs.Set(GlanceClient, 'get_images', fake.fake_get_images)
stubs.Set(GlanceClient, 'get_images_detailed',
fake.fake_get_images_detailed)
stubs.Set(GlanceClient, 'get_image_meta', fake.fake_get_image_meta)
stubs.Set(GlanceClient, 'add_image', fake.fake_add_image)
stubs.Set(GlanceClient, 'update_image', fake.fake_update_image)
stubs.Set(GlanceClient, 'delete_image', fake.fake_delete_image)
def stub_out_glance(stubs):
def fake_get_image_service():
client = glance_stubs.StubGlanceClient(_make_image_fixtures())
return nova.image.glance.GlanceImageService(client)
stubs.Set(nova.image, 'get_default_image_service', fake_get_image_service)
class FakeToken(object):

167
nova/tests/api/openstack/test_image_metadata.py
View File

@@ -23,7 +23,6 @@ from nova import flags
from nova.api import openstack
from nova import test
from nova.tests.api.openstack import fakes
import nova.wsgi
FLAGS = flags.FLAGS
@@ -31,76 +30,20 @@ FLAGS = flags.FLAGS
class ImageMetaDataTest(test.TestCase):
IMAGE_FIXTURES = [
{'status': 'active',
'name': 'image1',
'deleted': False,
'container_format': None,
'checksum': None,
'created_at': '2011-03-22T17:40:15',
'disk_format': None,
'updated_at': '2011-03-22T17:40:15',
'id': '1',
'location': 'file:///var/lib/glance/images/1',
'is_public': True,
'deleted_at': None,
'properties': {
'key1': 'value1',
'key2': 'value2'},
'size': 5882349},
{'status': 'active',
'name': 'image2',
'deleted': False,
'container_format': None,
'checksum': None,
'created_at': '2011-03-22T17:40:15',
'disk_format': None,
'updated_at': '2011-03-22T17:40:15',
'id': '2',
'location': 'file:///var/lib/glance/images/2',
'is_public': True,
'deleted_at': None,
'properties': {
'key1': 'value1',
'key2': 'value2'},
'size': 5882349},
{'status': 'active',
'name': 'image3',
'deleted': False,
'container_format': None,
'checksum': None,
'created_at': '2011-03-22T17:40:15',
'disk_format': None,
'updated_at': '2011-03-22T17:40:15',
'id': '3',
'location': 'file:///var/lib/glance/images/2',
'is_public': True,
'deleted_at': None,
'properties': {},
'size': 5882349},
]
def setUp(self):
super(ImageMetaDataTest, self).setUp()
self.flags(image_service='nova.image.glance.GlanceImageService')
# NOTE(dprince) max out properties/metadata in image 3 for testing
img3 = self.IMAGE_FIXTURES[2]
for num in range(FLAGS.quota_metadata_items):
img3['properties']['key%i' % num] = "blah"
fakes.stub_out_glance(self.stubs, self.IMAGE_FIXTURES)
fakes.stub_out_glance(self.stubs)
def test_index(self):
req = webob.Request.blank('/v1.1/123/images/1/metadata')
req = webob.Request.blank('/v1.1/123/images/123/metadata')
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(200, res.status_int)
expected = self.IMAGE_FIXTURES[0]['properties']
self.assertEqual(len(expected), len(res_dict['metadata']))
for (key, value) in res_dict['metadata'].items():
self.assertEqual(value, res_dict['metadata'][key])
expected = {'metadata': {'key1': 'value1'}}
self.assertEqual(res_dict, expected)
def test_show(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
res = req.get_response(fakes.wsgi_app())
res_dict = json.loads(res.body)
self.assertEqual(200, res.status_int)
@@ -109,32 +52,38 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual('value1', res_dict['meta']['key1'])
def test_show_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key9')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key9')
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_show_image_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/100/metadata/key1')
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_create(self):
req = webob.Request.blank('/v1.1/fake/images/2/metadata')
req = webob.Request.blank('/v1.1/fake/images/123/metadata')
req.method = 'POST'
req.body = '{"metadata": {"key9": "value9"}}'
req.body = '{"metadata": {"key7": "value7"}}'
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(200, res.status_int)
actual_output = json.loads(res.body)
expected_output = {
'metadata': {
'key1': 'value1',
'key2': 'value2',
'key9': 'value9',
},
}
expected_output = {'metadata': {'key1': 'value1', 'key7': 'value7'}}
self.assertEqual(expected_output, actual_output)
def test_create_image_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/100/metadata')
req.method = 'POST'
req.body = '{"metadata": {"key7": "value7"}}'
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_update_all(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata')
req = webob.Request.blank('/v1.1/fake/images/123/metadata')
req.method = 'PUT'
req.body = '{"metadata": {"key9": "value9"}}'
req.headers["content-type"] = "application/json"
@@ -142,17 +91,20 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(200, res.status_int)
actual_output = json.loads(res.body)
expected_output = {
'metadata': {
'key9': 'value9',
},
}
expected_output = {'metadata': {'key9': 'value9'}}
self.assertEqual(expected_output, actual_output)
def test_update_all_image_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/100/metadata')
req.method = 'PUT'
req.body = '{"metadata": {"key9": "value9"}}'
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_update_item(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
req.method = 'PUT'
req.body = '{"meta": {"key1": "zz"}}'
req.headers["content-type"] = "application/json"
@@ -160,15 +112,20 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(200, res.status_int)
actual_output = json.loads(res.body)
expected_output = {
'meta': {
'key1': 'zz',
},
}
expected_output = {'meta': {'key1': 'zz'}}
self.assertEqual(actual_output, expected_output)
def test_update_item_image_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/100/metadata/key1')
req.method = 'PUT'
req.body = '{"meta": {"key1": "zz"}}'
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_update_item_bad_body(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
req.method = 'PUT'
req.body = '{"key1": "zz"}'
req.headers["content-type"] = "application/json"
@@ -176,15 +133,18 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(400, res.status_int)
def test_update_item_too_many_keys(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
req.method = 'PUT'
req.body = '{"meta": {"key1": "value1", "key2": "value2"}}'
overload = {}
for num in range(FLAGS.quota_metadata_items + 1):
overload['key%s' % num] = 'value%s' % num
req.body = json.dumps({'meta': overload})
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
self.assertEqual(400, res.status_int)
def test_update_item_body_uri_mismatch(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/bad')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/bad')
req.method = 'PUT'
req.body = '{"meta": {"key1": "value1"}}'
req.headers["content-type"] = "application/json"
@@ -192,7 +152,7 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(400, res.status_int)
def test_update_item_xml(self):
req = webob.Request.blank('/v1.1/fake/images/1/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
req.method = 'PUT'
req.body = '<meta key="key1">five</meta>'
req.headers["content-type"] = "application/xml"
@@ -200,22 +160,24 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(200, res.status_int)
actual_output = json.loads(res.body)
expected_output = {
'meta': {
'key1': 'five',
},
}
expected_output = {'meta': {'key1': 'five'}}
self.assertEqual(actual_output, expected_output)
def test_delete(self):
req = webob.Request.blank('/v1.1/fake/images/2/metadata/key1')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/key1')
req.method = 'DELETE'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(204, res.status_int)
self.assertEqual('', res.body)
def test_delete_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/2/metadata/blah')
req = webob.Request.blank('/v1.1/fake/images/123/metadata/blah')
req.method = 'DELETE'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
def test_delete_image_not_found(self):
req = webob.Request.blank('/v1.1/fake/images/100/metadata/key1')
req.method = 'DELETE'
res = req.get_response(fakes.wsgi_app())
self.assertEqual(404, res.status_int)
@@ -225,7 +187,7 @@ class ImageMetaDataTest(test.TestCase):
for num in range(FLAGS.quota_metadata_items + 1):
data['metadata']['key%i' % num] = "blah"
json_string = str(data).replace("\'", "\"")
req = webob.Request.blank('/v1.1/fake/images/2/metadata')
req = webob.Request.blank('/v1.1/fake/images/123/metadata')
req.method = 'POST'
req.body = json_string
req.headers["content-type"] = "application/json"
@@ -233,7 +195,8 @@ class ImageMetaDataTest(test.TestCase):
self.assertEqual(413, res.status_int)
def test_too_many_metadata_items_on_put(self):
req = webob.Request.blank('/v1.1/fake/images/3/metadata/blah')
FLAGS.quota_metadata_items = 1
req = webob.Request.blank('/v1.1/fake/images/123/metadata/blah')
req.method = 'PUT'
req.body = '{"meta": {"blah": "blah"}}'
req.headers["content-type"] = "application/json"

313
nova/tests/api/openstack/test_images.py
View File

@@ -29,31 +29,30 @@ import stubout
import webob
from nova import context
from nova import test
import nova.api.openstack
from nova.api.openstack import images
from nova import test
from nova.tests.api.openstack import fakes
NOW_API_FORMAT = "2010-10-11T10:30:22Z"
class ImagesTest(test.TestCase):
"""
Test of the OpenStack API /images application controller w/Glance.
"""
NOW_GLANCE_FORMAT = "2010-10-11T10:30:22"
NOW_API_FORMAT = "2010-10-11T10:30:22Z"
def setUp(self):
"""Run before each test."""
super(ImagesTest, self).setUp()
self.flags(image_service='nova.image.glance.GlanceImageService')
self.stubs = stubout.StubOutForTesting()
fakes.stub_out_networking(self.stubs)
fakes.stub_out_rate_limiting(self.stubs)
fakes.stub_out_key_pair_funcs(self.stubs)
self.fixtures = self._make_image_fixtures()
fakes.stub_out_glance(self.stubs, initial_fixtures=self.fixtures)
fakes.stub_out_compute_api_snapshot(self.stubs)
fakes.stub_out_compute_api_backup(self.stubs)
fakes.stub_out_glance(self.stubs)
def tearDown(self):
"""Run after each test."""
@@ -63,36 +62,30 @@ class ImagesTest(test.TestCase):
def _get_fake_context(self):
class Context(object):
project_id = 'fake'
auth_token = True
return Context()
def _applicable_fixture(self, fixture, user_id):
"""Determine if this fixture is applicable for given user id."""
is_public = fixture["is_public"]
try:
uid = fixture["properties"]["user_id"]
except KeyError:
uid = None
return uid == user_id or is_public
def test_get_image_index(self):
request = webob.Request.blank('/v1.0/images')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
response_dict = json.loads(response.body)
response_list = response_dict["images"]
expected = [{'id': 123, 'name': 'public image'},
{'id': 124, 'name': 'queued snapshot'},
{'id': 125, 'name': 'saving snapshot'},
{'id': 126, 'name': 'active snapshot'},
{'id': 127, 'name': 'killed snapshot'},
{'id': 129, 'name': None}]
expected = [{'id': '123', 'name': 'public image'},
{'id': '124', 'name': 'queued snapshot'},
{'id': '125', 'name': 'saving snapshot'},
{'id': '126', 'name': 'active snapshot'},
{'id': '127', 'name': 'killed snapshot'},
{'id': '128', 'name': None}]
self.assertDictListMatch(response_list, expected)
def test_get_image(self):
request = webob.Request.blank('/v1.0/images/123')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
self.assertEqual(200, response.status_int)
@@ -100,20 +93,21 @@ class ImagesTest(test.TestCase):
expected_image = {
"image": {
"id": 123,
"id": "123",
"name": "public image",
"updated": self.NOW_API_FORMAT,
"created": self.NOW_API_FORMAT,
"updated": NOW_API_FORMAT,
"created": NOW_API_FORMAT,
"status": "ACTIVE",
"progress": 100,
},
}
self.assertEqual(expected_image, actual_image)
self.assertDictMatch(expected_image, actual_image)
def test_get_image_v1_1(self):
request = webob.Request.blank('/v1.1/fake/images/124')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
actual_image = json.loads(response.body)
@@ -124,10 +118,10 @@ class ImagesTest(test.TestCase):
expected_image = {
"image": {
"id": 124,
"id": "124",
"name": "queued snapshot",
"updated": self.NOW_API_FORMAT,
"created": self.NOW_API_FORMAT,
"updated": NOW_API_FORMAT,
"created": NOW_API_FORMAT,
"status": "QUEUED",
"progress": 0,
'server': {
@@ -161,11 +155,12 @@ class ImagesTest(test.TestCase):
def test_get_image_xml(self):
request = webob.Request.blank('/v1.0/images/123')
request.accept = "application/xml"
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
actual_image = minidom.parseString(response.body.replace(" ", ""))
expected_now = self.NOW_API_FORMAT
expected_now = NOW_API_FORMAT
expected_image = minidom.parseString("""
<image id="123"
name="public image"
@@ -179,15 +174,16 @@ class ImagesTest(test.TestCase):
self.assertEqual(expected_image.toxml(), actual_image.toxml())
def test_get_image_xml_no_name(self):
request = webob.Request.blank('/v1.0/images/129')
request = webob.Request.blank('/v1.0/images/128')
request.accept = "application/xml"
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
actual_image = minidom.parseString(response.body.replace(" ", ""))
expected_now = self.NOW_API_FORMAT
expected_now = NOW_API_FORMAT
expected_image = minidom.parseString("""
<image id="129"
<image id="128"
name="None"
updated="%(expected_now)s"
created="%(expected_now)s"
@@ -272,90 +268,154 @@ class ImagesTest(test.TestCase):
def test_get_image_index_v1_1(self):
request = webob.Request.blank('/v1.1/fake/images')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
response_dict = json.loads(response.body)
response_list = response_dict["images"]
fixtures = copy.copy(self.fixtures)
for image in fixtures:
if not self._applicable_fixture(image, "fake"):
fixtures.remove(image)
continue
href = "http://localhost/v1.1/fake/images/%s" % image["id"]
bookmark = "http://localhost/fake/images/%s" % image["id"]
test_image = {
"id": image["id"],
"name": image["name"],
expected = [
{
"id": "123",
"name": "public image",
"links": [
{
"rel": "self",
"href": href,
"href": "http://localhost/v1.1/fake/images/123",
},
{
"rel": "bookmark",
"href": bookmark,
"href": "http://localhost/fake/images/123",
},
],
}
self.assertTrue(test_image in response_list)
},
{
"id": "124",
"name": "queued snapshot",
"links": [
{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/124",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/124",
},
],
},
{
"id": "125",
"name": "saving snapshot",
"links": [
{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/125",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/125",
},
],
},
{
"id": "126",
"name": "active snapshot",
"links": [
{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/126",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/126",
},
],
},
{
"id": "127",
"name": "killed snapshot",
"links": [
{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/127",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/127",
},
],
},
{
"id": "128",
"name": None,
"links": [
{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/128",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/128",
},
],
},
]
self.assertEqual(len(response_list), len(fixtures))
self.assertDictListMatch(response_list, expected)
def test_get_image_details(self):
request = webob.Request.blank('/v1.0/images/detail')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
response_dict = json.loads(response.body)
response_list = response_dict["images"]
expected = [{
'id': 123,
'id': '123',
'name': 'public image',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
},
{
'id': 124,
'id': '124',
'name': 'queued snapshot',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'QUEUED',
'progress': 0,
},
{
'id': 125,
'id': '125',
'name': 'saving snapshot',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'SAVING',
'progress': 0,
},
{
'id': 126,
'id': '126',
'name': 'active snapshot',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
},
{
'id': 127,
'id': '127',
'name': 'killed snapshot',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'FAILED',
'progress': 0,
},
{
'id': 129,
'id': '128',
'name': None,
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
}]
@@ -364,7 +424,8 @@ class ImagesTest(test.TestCase):
def test_get_image_details_v1_1(self):
request = webob.Request.blank('/v1.1/fake/images/detail')
response = request.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
response = request.get_response(app)
response_dict = json.loads(response.body)
response_list = response_dict["images"]
@@ -372,11 +433,11 @@ class ImagesTest(test.TestCase):
server_bookmark = "http://localhost/servers/42"
expected = [{
'id': 123,
'id': '123',
'name': 'public image',
'metadata': {},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'metadata': {'key1': 'value1'},
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
"links": [{
@@ -389,14 +450,14 @@ class ImagesTest(test.TestCase):
}],
},
{
'id': 124,
'id': '124',
'name': 'queued snapshot',
'metadata': {
u'instance_ref': u'http://localhost/v1.1/servers/42',
u'user_id': u'fake',
},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'QUEUED',
'progress': 0,
'server': {
@@ -420,14 +481,14 @@ class ImagesTest(test.TestCase):
}],
},
{
'id': 125,
'id': '125',
'name': 'saving snapshot',
'metadata': {
u'instance_ref': u'http://localhost/v1.1/servers/42',
u'user_id': u'fake',
},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'SAVING',
'progress': 0,
'server': {
@@ -451,14 +512,14 @@ class ImagesTest(test.TestCase):
}],
},
{
'id': 126,
'id': '126',
'name': 'active snapshot',
'metadata': {
u'instance_ref': u'http://localhost/v1.1/servers/42',
u'user_id': u'fake',
},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
'server': {
@@ -482,14 +543,14 @@ class ImagesTest(test.TestCase):
}],
},
{
'id': 127,
'id': '127',
'name': 'killed snapshot',
'metadata': {
u'instance_ref': u'http://localhost/v1.1/servers/42',
u'user_id': u'fake',
},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'FAILED',
'progress': 0,
'server': {
@@ -513,20 +574,20 @@ class ImagesTest(test.TestCase):
}],
},
{
'id': 129,
'id': '128',
'name': None,
'metadata': {},
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT,
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT,
'status': 'ACTIVE',
'progress': 100,
"links": [{
"rel": "self",
"href": "http://localhost/v1.1/fake/images/129",
"href": "http://localhost/v1.1/fake/images/128",
},
{
"rel": "bookmark",
"href": "http://localhost/fake/images/129",
"href": "http://localhost/fake/images/128",
}],
},
]
@@ -738,11 +799,12 @@ class ImagesTest(test.TestCase):
def test_get_image_found(self):
req = webob.Request.blank('/v1.0/images/123')
res = req.get_response(fakes.wsgi_app())
app = fakes.wsgi_app(fake_auth_context=self._get_fake_context())
res = req.get_response(app)
image_meta = json.loads(res.body)['image']
expected = {'id': 123, 'name': 'public image',
'updated': self.NOW_API_FORMAT,
'created': self.NOW_API_FORMAT, 'status': 'ACTIVE',
expected = {'id': '123', 'name': 'public image',
'updated': NOW_API_FORMAT,
'created': NOW_API_FORMAT, 'status': 'ACTIVE',
'progress': 100}
self.assertDictMatch(image_meta, expected)
@@ -751,14 +813,6 @@ class ImagesTest(test.TestCase):
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 404)
def test_get_image_not_owned(self):
"""We should return a 404 if we request an image that doesn't belong
to us
"""
req = webob.Request.blank('/v1.0/images/128')
res = req.get_response(fakes.wsgi_app())
self.assertEqual(res.status_int, 404)
def test_create_image(self):
body = dict(image=dict(serverId='123', name='Snapshot 1'))
req = webob.Request.blank('/v1.0/images')
@@ -801,49 +855,6 @@ class ImagesTest(test.TestCase):
response = req.get_response(fakes.wsgi_app())
self.assertEqual(400, response.status_int)
@classmethod
def _make_image_fixtures(cls):
image_id = 123
base_attrs = {'created_at': cls.NOW_GLANCE_FORMAT,
'updated_at': cls.NOW_GLANCE_FORMAT,
'deleted_at': None,
'deleted': False}
fixtures = []
def add_fixture(**kwargs):
kwargs.update(base_attrs)
fixtures.append(kwargs)
# Public image
add_fixture(id=image_id, name='public image', is_public=True,
status='active', properties={})
image_id += 1
# Snapshot for User 1
server_ref = 'http://localhost/v1.1/servers/42'
snapshot_properties = {'instance_ref': server_ref, 'user_id': 'fake'}
for status in ('queued', 'saving', 'active', 'killed'):
add_fixture(id=image_id, name='%s snapshot' % status,
is_public=False, status=status,
properties=snapshot_properties)
image_id += 1
# Snapshot for User 2
other_snapshot_properties = {'instance_id': '43', 'user_id': 'other'}
add_fixture(id=image_id, name='someone elses snapshot',
is_public=False, status='active',
properties=other_snapshot_properties)
image_id += 1
# Image without a name
add_fixture(id=image_id, is_public=True, status='active',
properties={})
image_id += 1
return fixtures
class ImageXMLSerializationTest(test.TestCase):

69
nova/tests/glance/stubs.py
View File

@@ -16,7 +16,9 @@
import StringIO
from nova import exception
import nova.image
import nova.image.glance
def stubout_glance_client(stubs):
@@ -78,3 +80,70 @@ class FakeGlance(object):
def get_image(self, image_id):
image = self.IMAGE_FIXTURES[int(image_id)]
return image['image_meta'], image['image_data']
NOW_GLANCE_FORMAT = "2010-10-11T10:30:22"
class StubGlanceClient(object):
def __init__(self, images=None):
self.images = []
_images = images or []
map(lambda image: self.add_image(image, None), _images)
def set_auth_token(self, auth_tok):
pass
def get_image_meta(self, image_id):
for image in self.images:
if image['id'] == str(image_id):
return image
raise exception.ImageNotFound(image_id=image_id)
#TODO(bcwaldon): implement filters
def get_images_detailed(self, filters=None, marker=None, limit=3):
if marker is None:
index = 0
else:
for index, image in enumerate(self.images):
if image['id'] == str(marker):
index += 1
break
return self.images[index:index + limit]
def get_image(self, image_id):
return self.get_image_meta(image_id), []
def add_image(self, metadata, data):
metadata['created_at'] = NOW_GLANCE_FORMAT
metadata['updated_at'] = NOW_GLANCE_FORMAT
self.images.append(metadata)
try:
image_id = str(metadata['id'])
except KeyError:
# auto-generate an id if one wasn't provided
image_id = str(len(self.images))
self.images[-1]['id'] = image_id
return self.images[-1]
def update_image(self, image_id, metadata, data):
for i, image in enumerate(self.images):
if image['id'] == str(image_id):
if 'id' in metadata:
metadata['id'] = str(metadata['id'])
self.images[i].update(metadata)
return self.images[i]
raise exception.ImageNotFound(image_id=image_id)
def delete_image(self, image_id):
for i, image in enumerate(self.images):
if image['id'] == image_id:
del self.images[i]
return
raise exception.ImageNotFound(image_id=image_id)

269
nova/tests/image/test_glance.py
View File

@@ -24,64 +24,7 @@ from nova import context
from nova import exception
from nova.image import glance
from nova import test
class StubGlanceClient(object):
def __init__(self, images=None, add_response=None, update_response=None):
self.images = images or []
self.add_response = add_response
self.update_response = update_response
def set_auth_token(self, auth_tok):
pass
def get_image_meta(self, image_id):
for image in self.images:
if image['id'] == image_id:
return image
raise exception.ImageNotFound(image_id=image_id)
#TODO(bcwaldon): implement filters
def get_images_detailed(self, filters=None, marker=None, limit=3):
if marker is None:
index = 0
else:
for index, image in enumerate(self.images):
if image['id'] == marker:
index += 1
break
return self.images[index:index + limit]
def get_image(self, image_id):
return self.get_image_meta(image_id), []
def add_image(self, metadata, data):
self.images.append(metadata)
try:
image_id = int(metadata['id'])
except KeyError:
# auto-generate an id if one wasn't provided
image_id = len(self.images)
metadata['id'] = image_id
return metadata
def update_image(self, image_id, metadata, data):
for i, image in enumerate(self.images):
if image['id'] == image_id:
self.images[i].update(metadata)
return self.images[i]
raise exception.ImageNotFound(image_id=image_id)
def delete_image(self, image_id):
for i, image in enumerate(self.images):
if image['id'] == image_id:
del self.images[i]
return
raise exception.ImageNotFound(image_id=image_id)
from nova.tests.glance import stubs as glance_stubs
class NullWriter(object):
@@ -147,10 +90,9 @@ class TestGlanceImageService(test.TestCase):
def setUp(self):
super(TestGlanceImageService, self).setUp()
self.stubs = stubout.StubOutForTesting()
fakes.stub_out_glance(self.stubs)
fakes.stub_out_compute_api_snapshot(self.stubs)
self.client = StubGlanceClient()
self.service = glance.GlanceImageService(client=self.client)
client = glance_stubs.StubGlanceClient()
self.service = glance.GlanceImageService(client=client)
self.context = context.RequestContext('fake', 'fake', auth_token=True)
self.service.delete_all()
@@ -158,7 +100,6 @@ class TestGlanceImageService(test.TestCase):
self.stubs.UnsetAll()
super(TestGlanceImageService, self).tearDown()
@staticmethod
def _make_fixture(**kwargs):
fixture = {'name': None,
@@ -168,47 +109,11 @@ class TestGlanceImageService(test.TestCase):
fixture.update(kwargs)
return fixture
def _make_datetime_fixtures(self):
return [
{
'id': '1',
'name': 'image1',
'is_public': True,
'created_at': self.NOW_GLANCE_FORMAT,
'updated_at': self.NOW_GLANCE_FORMAT,
'deleted_at': self.NOW_GLANCE_FORMAT,
},
{
'id': '2',
'name': 'image2',
'is_public': True,
'created_at': self.NOW_GLANCE_OLD_FORMAT,
'updated_at': self.NOW_GLANCE_OLD_FORMAT,
'deleted_at': self.NOW_GLANCE_OLD_FORMAT,
},
]
def _make_datetime_fixture(self):
return self._make_fixture(created_at=self.NOW_GLANCE_FORMAT,
updated_at=self.NOW_GLANCE_FORMAT,
deleted_at=self.NOW_GLANCE_FORMAT)
def _make_none_datetime_fixture(self):
return self._make_fixture(updated_at=None, deleted_at=None)
def assertDateTimesFilled(self, image_meta):
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
self.assertEqual(image_meta['deleted_at'], self.NOW_DATETIME)
def assertDateTimesEmpty(self, image_meta):
self.assertEqual(image_meta['updated_at'], None)
self.assertEqual(image_meta['deleted_at'], None)
def assertDateTimesBlank(self, image_meta):
self.assertEqual(image_meta['updated_at'], '')
self.assertEqual(image_meta['deleted_at'], '')
def test_create_with_instance_id(self):
"""Ensure instance_id is persisted as an image-property"""
fixture = {'name': 'test image',
@@ -216,14 +121,23 @@ class TestGlanceImageService(test.TestCase):
'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', 'is_public': False,
'size': None, 'location': None, 'disk_format': None,
'container_format': None, 'checksum': None,
'created_at': None, 'updated_at': None,
'deleted_at': None, 'deleted': None, 'status': None,
'properties': {'instance_id': '42', 'user_id': 'fake'}}
expected = {
'id': image_id,
'name': 'test image',
'is_public': False,
'size': None,
'location': None,
'disk_format': None,
'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'},
}
self.assertDictMatch(image_meta, expected)
image_metas = self.service.detail(self.context)
@@ -238,12 +152,22 @@ class TestGlanceImageService(test.TestCase):
fixture = {'name': 'test image', 'is_public': False}
image_id = self.service.create(self.context, fixture)['id']
expected = {'id': image_id, 'name': 'test image', 'is_public': False,
'size': None, 'location': None, 'disk_format': None,
'container_format': None, 'checksum': None,
'created_at': None, 'updated_at': None,
'deleted_at': None, 'deleted': None, 'status': None,
'properties': {}}
expected = {
'id': image_id,
'name': 'test image',
'is_public': False,
'size': None,
'location': None,
'disk_format': None,
'container_format': None,
'checksum': None,
'created_at': self.NOW_DATETIME,
'updated_at': self.NOW_DATETIME,
'deleted_at': None,
'deleted': None,
'status': None,
'properties': {},
}
actual = self.service.show(self.context, image_id)
self.assertDictMatch(actual, expected)
@@ -367,8 +291,8 @@ class TestGlanceImageService(test.TestCase):
'disk_format': None,
'container_format': None,
'checksum': None,
'created_at': None,
'updated_at': None,
'created_at': self.NOW_DATETIME,
'updated_at': self.NOW_DATETIME,
'deleted_at': None,
'deleted': None
}
@@ -410,8 +334,8 @@ class TestGlanceImageService(test.TestCase):
'disk_format': None,
'container_format': None,
'checksum': None,
'created_at': None,
'updated_at': None,
'created_at': self.NOW_DATETIME,
'updated_at': self.NOW_DATETIME,
'deleted_at': None,
'deleted': None
}
@@ -453,12 +377,22 @@ class TestGlanceImageService(test.TestCase):
image_id = self.service.create(self.context, fixture)['id']
image_meta = self.service.show(self.context, image_id)
expected = {'id': image_id, 'name': 'image1', 'is_public': True,
'size': None, 'location': None, 'disk_format': None,
'container_format': None, 'checksum': None,
'created_at': None, 'updated_at': None,
'deleted_at': None, 'deleted': None, 'status': None,
'properties': {}}
expected = {
'id': image_id,
'name': 'image1',
'is_public': True,
'size': None,
'location': None,
'disk_format': None,
'container_format': None,
'checksum': None,
'created_at': self.NOW_DATETIME,
'updated_at': self.NOW_DATETIME,
'deleted_at': None,
'deleted': None,
'status': None,
'properties': {},
}
self.assertEqual(image_meta, expected)
def test_show_raises_when_no_authtoken_in_the_context(self):
@@ -476,89 +410,44 @@ class TestGlanceImageService(test.TestCase):
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)
expected = [{'id': image_id, 'name': 'image10', 'is_public': True,
'size': None, 'location': None, 'disk_format': None,
'container_format': None, 'checksum': None,
'created_at': None, 'updated_at': None,
'deleted_at': None, 'deleted': None, 'status': None,
'properties': {}}]
expected = [
{
'id': image_id,
'name': 'image10',
'is_public': True,
'size': None,
'location': None,
'disk_format': None,
'container_format': None,
'checksum': None,
'created_at': self.NOW_DATETIME,
'updated_at': self.NOW_DATETIME,
'deleted_at': None,
'deleted': None,
'status': None,
'properties': {},
},
]
self.assertEqual(image_metas, expected)
def test_show_handles_none_datetimes(self):
fixture = self._make_fixture(updated_at=None, deleted_at=None)
image_id = self.service.create(self.context, fixture)['id']
image_meta = self.service.show(self.context, image_id)
self.assertDateTimesEmpty(image_meta)
def test_show_handles_blank_datetimes(self):
fixture = self._make_fixture(updated_at='', deleted_at='')
image_id = self.service.create(self.context, fixture)['id']
image_meta = self.service.show(self.context, image_id)
self.assertDateTimesBlank(image_meta)
def test_detail_handles_none_datetimes(self):
fixture = self._make_fixture(updated_at=None, deleted_at=None)
self.service.create(self.context, fixture)
image_meta = self.service.detail(self.context)[0]
self.assertDateTimesEmpty(image_meta)
def test_detail_handles_blank_datetimes(self):
fixture = self._make_fixture(updated_at='', deleted_at='')
self.service.create(self.context, fixture)
image_meta = self.service.detail(self.context)[0]
self.assertDateTimesBlank(image_meta)
def test_get_handles_none_datetimes(self):
fixture = self._make_fixture(updated_at=None, deleted_at=None)
image_id = self.service.create(self.context, fixture)['id']
writer = NullWriter()
image_meta = self.service.get(self.context, image_id, writer)
self.assertDateTimesEmpty(image_meta)
def test_get_handles_blank_datetimes(self):
fixture = self._make_fixture(updated_at='', deleted_at='')
image_id = self.service.create(self.context, fixture)['id']
writer = NullWriter()
image_meta = self.service.get(self.context, image_id, writer)
self.assertDateTimesBlank(image_meta)
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.assertDateTimesFilled(image_meta)
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
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.assertDateTimesFilled(image_meta)
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)
def test_get_makes_datetimes(self):
fixture = self._make_datetime_fixture()
image_id = self.service.create(self.context, fixture)['id']
writer = NullWriter()
image_meta = self.service.get(self.context, image_id, writer)
self.assertDateTimesFilled(image_meta)
def test_create_handles_datetimes(self):
fixture = self._make_datetime_fixture()
image_meta = self.service.create(self.context, fixture)
self.assertDateTimesFilled(image_meta)
def test_create_handles_none_datetimes(self):
fixture = self._make_none_datetime_fixture()
image_meta = self.service.create(self.context, fixture)
self.assertDateTimesEmpty(image_meta)
def test_update_handles_datetimes(self):
fixture = self._make_datetime_fixture()
image_id = self.service.create(self.context, fixture)['id']
image_meta = self.service.update(self.context, image_id, {})
self.assertDateTimesFilled(image_meta)
def test_update_handles_none_datetimes(self):
fixture = self._make_none_datetime_fixture()
image_id = self.service.create(self.context, fixture)['id']
image_meta = self.service.update(self.context, image_id, {})
self.assertDateTimesEmpty(image_meta)
self.assertEqual(image_meta['created_at'], self.NOW_DATETIME)
self.assertEqual(image_meta['updated_at'], self.NOW_DATETIME)

2
nova/tests/test_xenapi.py
View File

@@ -745,7 +745,7 @@ class XenAPIMigrateInstance(test.TestCase):
fake_utils.stub_out_utils_execute(self.stubs)
stubs.stub_out_migration_methods(self.stubs)
stubs.stubout_get_this_vm_uuid(self.stubs)
glance_stubs.stubout_glance_client(self.stubs)
#glance_stubs.stubout_glance_client(self.stubs)
def test_migrate_disk_and_power_off(self):
instance = db.instance_create(self.context, self.values)