openstack/deb-cinder
Code Issues Proposed changes

Fix volume test uuid warnings in api.contrib

Browse Source 
This commit fixes invalid uuid usage in the various
volume unit tests under cinder/tests/unit/api/contrib,
cleaning up FutureWarnings from oslo_versioned_objects [1].

[1] http://docs.openstack.org/developer/oslo.versionedobjects/api/fields.html#oslo_versionedobjects.fields.UUIDField

Change-Id: I526f9cf1c383f75490f1ac86c0fc5182376f5b3e
This commit is contained in:
Tom Barron 2016-04-11 19:25:44 -04:00
parent 358f78bccd
commit 453e4a563a
8 changed files with 398 additions and 262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

285
cinder/tests/unit/api/contrib/test_volume_actions.py
View File

@ -45,7 +45,8 @@ class VolumeActionsTest(test.TestCase):
def setUp(self):
super(VolumeActionsTest, self).setUp()
self.context = context.RequestContext('fake', 'fake', is_admin=False)
self.context = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
is_admin=False)
self.UUID = uuid.uuid4()
self.controller = volume_actions.VolumeActionsController()
self.api_patchers = {}
@ -56,9 +57,10 @@ class VolumeActionsTest(test.TestCase):
self.addCleanup(self.api_patchers[_meth].stop)
self.api_patchers[_meth].return_value = True
db_vol = {'id': 'fake', 'host': 'fake', 'status': 'available',
db_vol = {'id': fake.VOLUME_ID, 'host': 'fake', 'status': 'available',
'size': 1, 'migration_status': None,
'volume_type_id': 'fake', 'project_id': 'project_id'}
'volume_type_id': fake.VOLUME_TYPE_ID,
'project_id': fake.PROJECT_ID}
vol = fake_volume.fake_volume_obj(self.context, **db_vol)
self.get_patcher = mock.patch('cinder.volume.api.API.get')
self.mock_volume_get = self.get_patcher.start()
@ -76,10 +78,10 @@ class VolumeActionsTest(test.TestCase):
self.flags(rpc_backend='cinder.openstack.common.rpc.impl_fake')
def test_simple_api_actions(self):
app = fakes.wsgi_app()
app = fakes.wsgi_app(fake_auth_context=self.context)
for _action in self._actions:
req = webob.Request.blank('/v2/fake/volumes/%s/action' %
self.UUID)
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, self.UUID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes({_action: None})
req.content_type = 'application/json'
@ -91,12 +93,14 @@ class VolumeActionsTest(test.TestCase):
'initialize_connection') as init_conn:
init_conn.return_value = {}
body = {'os-initialize_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(200, res.status_int)
def test_initialize_connection_without_connector(self):
@ -104,12 +108,14 @@ class VolumeActionsTest(test.TestCase):
'initialize_connection') as init_conn:
init_conn.return_value = {}
body = {'os-initialize_connection': {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_initialize_connection_exception(self):
@ -118,12 +124,14 @@ class VolumeActionsTest(test.TestCase):
init_conn.side_effect = \
exception.VolumeBackendAPIException(data=None)
body = {'os-initialize_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(500, res.status_int)
def test_terminate_connection(self):
@ -131,12 +139,14 @@ class VolumeActionsTest(test.TestCase):
'terminate_connection') as terminate_conn:
terminate_conn.return_value = {}
body = {'os-terminate_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_terminate_connection_without_connector(self):
@ -144,12 +154,14 @@ class VolumeActionsTest(test.TestCase):
'terminate_connection') as terminate_conn:
terminate_conn.return_value = {}
body = {'os-terminate_connection': {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_terminate_connection_with_exception(self):
@ -158,46 +170,54 @@ class VolumeActionsTest(test.TestCase):
terminate_conn.side_effect = \
exception.VolumeBackendAPIException(data=None)
body = {'os-terminate_connection': {'connector': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(500, res.status_int)
def test_attach_to_instance(self):
body = {'os-attach': {'instance_uuid': 'fake',
body = {'os-attach': {'instance_uuid': fake.INSTANCE_ID,
'mountpoint': '/dev/vdc',
'mode': 'rw'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
body = {'os-attach': {'instance_uuid': 'fake',
body = {'os-attach': {'instance_uuid': fake.INSTANCE_ID,
'host_name': 'fake_host',
'mountpoint': '/dev/vdc'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.headers["content-type"] = "application/json"
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_attach_to_host(self):
# using 'read-write' mode attach volume by default
body = {'os-attach': {'host_name': 'fake_host',
'mountpoint': '/dev/vdc'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_volume_attach_to_instance_raises_remote_error(self):
@ -205,12 +225,13 @@ class VolumeActionsTest(test.TestCase):
messaging.RemoteError(exc_type='InvalidUUID')
with mock.patch.object(volume_api.API, 'attach',
side_effect=volume_remote_error):
id = 1
id = fake.VOLUME_ID
vol = {"instance_uuid": self.UUID,
"mountpoint": "/dev/vdc",
"mode": "rw"}
body = {"os-attach": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._attach,
req,
@ -223,12 +244,13 @@ class VolumeActionsTest(test.TestCase):
messaging.RemoteError(exc_type='DBError')
with mock.patch.object(volume_api.API, 'attach',
side_effect=volume_remote_error):
id = 1
id = fake.VOLUME_ID
vol = {"instance_uuid": self.UUID,
"mountpoint": "/dev/vdc",
"mode": "rw"}
body = {"os-attach": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(messaging.RemoteError,
self.controller._attach,
req,
@ -236,13 +258,15 @@ class VolumeActionsTest(test.TestCase):
body)
def test_detach(self):
body = {'os-detach': {'attachment_id': 'fakeuuid'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
body = {'os-detach': {'attachment_id': fake.ATTACHMENT_ID}}
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_volume_detach_raises_remote_error(self):
@ -250,10 +274,11 @@ class VolumeActionsTest(test.TestCase):
messaging.RemoteError(exc_type='VolumeAttachmentNotFound')
with mock.patch.object(volume_api.API, 'detach',
side_effect=volume_remote_error):
id = 1
id = fake.VOLUME_ID
vol = {"attachment_id": self.UUID}
body = {"os-detach": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._detach,
req,
@ -266,10 +291,11 @@ class VolumeActionsTest(test.TestCase):
messaging.RemoteError(exc_type='DBError')
with mock.patch.object(volume_api.API, 'detach',
side_effect=volume_remote_error):
id = 1
id = fake.VOLUME_ID
vol = {"attachment_id": self.UUID}
body = {"os-detach": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(messaging.RemoteError,
self.controller._detach,
req,
@ -279,31 +305,37 @@ class VolumeActionsTest(test.TestCase):
def test_attach_with_invalid_arguments(self):
# Invalid request to attach volume an invalid target
body = {'os-attach': {'mountpoint': '/dev/vdc'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.headers["content-type"] = "application/json"
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
# Invalid request to attach volume with an invalid mode
body = {'os-attach': {'instance_uuid': 'fake',
'mountpoint': '/dev/vdc',
'mode': 'rr'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.headers["content-type"] = "application/json"
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
body = {'os-attach': {'host_name': 'fake_host',
'mountpoint': '/dev/vdc',
'mode': 'ww'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.headers["content-type"] = "application/json"
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_begin_detaching(self):
@ -313,12 +345,14 @@ class VolumeActionsTest(test.TestCase):
fake_begin_detaching)
body = {'os-begin_detaching': {'fake': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_roll_detaching(self):
@ -328,12 +362,14 @@ class VolumeActionsTest(test.TestCase):
fake_roll_detaching)
body = {'os-roll_detaching': {'fake': 'fake'}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_extend_volume(self):
@ -343,12 +379,14 @@ class VolumeActionsTest(test.TestCase):
fake_extend_volume)
body = {'os-extend': {'new_size': 5}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(202, res.status_int)
def test_extend_volume_invalid_status(self):
@ -359,12 +397,13 @@ class VolumeActionsTest(test.TestCase):
fake_extend_volume)
body = {'os-extend': {'new_size': 5}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_update_readonly_flag(self):
@ -377,11 +416,13 @@ class VolumeActionsTest(test.TestCase):
body = {"os-update_readonly_flag": {"readonly": readonly}}
if readonly is None:
body = {"os-update_readonly_flag": {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(return_code, res.status_int)
make_update_readonly_flag_test(self, True, 202)
@ -400,11 +441,13 @@ class VolumeActionsTest(test.TestCase):
body = {"os-set_bootable": {"bootable": bootable}}
if bootable is None:
body = {"os-set_bootable": {}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.context))
self.assertEqual(return_code, res.status_int)
make_set_bootable_test(self, True, 200)
@ -421,8 +464,12 @@ class VolumeActionsTest(test.TestCase):
class VolumeRetypeActionsTest(VolumeActionsTest):
def setUp(self):
def get_vol_type(*args, **kwargs):
d1 = {'id': 'fake', 'qos_specs_id': 'fakeqid1', 'extra_specs': {}}
d2 = {'id': 'foo', 'qos_specs_id': 'fakeqid2', 'extra_specs': {}}
d1 = {'id': fake.VOLUME_TYPE_ID,
'qos_specs_id': fake.QOS_SPEC_ID,
'extra_specs': {}}
d2 = {'id': fake.volume_type2_id,
'qos_specs_id': fake.QOS_SPEC2_ID,
'extra_specs': {}}
return d1 if d1['id'] == args[1] else d2
self.retype_patchers = {}
@ -444,14 +491,18 @@ class VolumeRetypeActionsTest(VolumeActionsTest):
self.retype_mocks['reserve'].return_value = None
super(VolumeRetypeActionsTest, self).setUp()
self.context = context.RequestContext(
fake.USER_ID, fake.PROJECT_ID, auth_token=True)
def _retype_volume_exec(self, expected_status, new_type='foo'):
req = webob.Request.blank('/v2/fake/volumes/1/action')
def _retype_volume_exec(self, expected_status,
new_type=fake.volume_type2_id):
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.headers['content-type'] = 'application/json'
retype_body = {'new_type': new_type, 'migration_policy': 'never'}
req.body = jsonutils.dump_as_bytes({'os-retype': retype_body})
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(expected_status, res.status_int)
@mock.patch('cinder.volume.qos_specs.get_qos_specs')
@ -459,28 +510,30 @@ class VolumeRetypeActionsTest(VolumeActionsTest):
# Test that the retype API works for both available and in-use
self._retype_volume_exec(202)
self.mock_volume_get.return_value['status'] = 'in-use'
specs = {'id': 'fakeqid1', 'name': 'fake_name1',
specs = {'id': fake.QOS_SPEC_ID, 'name': 'fake_name1',
'consumer': 'back-end', 'specs': {'key1': 'value1'}}
_mock_get_qspecs.return_value = specs
self._retype_volume_exec(202)
def test_retype_volume_no_body(self):
# Request with no body should fail
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.headers['content-type'] = 'application/json'
req.body = jsonutils.dump_as_bytes({'os-retype': None})
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_retype_volume_bad_policy(self):
# Request with invalid migration policy should fail
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.headers['content-type'] = 'application/json'
retype_body = {'new_type': 'foo', 'migration_policy': 'invalid'}
req.body = jsonutils.dump_as_bytes({'os-retype': retype_body})
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_retype_volume_bad_status(self):
@ -496,7 +549,7 @@ class VolumeRetypeActionsTest(VolumeActionsTest):
def test_retype_same_type(self):
# Should fail if new type and old type are the same
self._retype_volume_exec(400, new_type='fake')
self._retype_volume_exec(400, new_type=fake.VOLUME_TYPE_ID)
def test_retype_over_quota(self):
# Should fail if going over quota for new type
@ -511,9 +564,9 @@ class VolumeRetypeActionsTest(VolumeActionsTest):
def _retype_volume_diff_qos(self, vol_status, consumer, expected_status,
_mock_get_qspecs):
def fake_get_qos(ctxt, qos_id):
d1 = {'id': 'fakeqid1', 'name': 'fake_name1',
d1 = {'id': fake.QOS_SPEC_ID, 'name': 'fake_name1',
'consumer': consumer, 'specs': {'key1': 'value1'}}
d2 = {'id': 'fakeqid2', 'name': 'fake_name2',
d2 = {'id': fake.QOS_SPEC2_ID, 'name': 'fake_name2',
'consumer': consumer, 'specs': {'key1': 'value1'}}
return d1 if d1['id'] == qos_id else d2
@ -537,7 +590,7 @@ class VolumeRetypeActionsTest(VolumeActionsTest):
def stub_volume_get(self, context, volume_id):
volume = stubs.stub_volume(volume_id)
if volume_id == 5:
if volume_id == fake.VOLUME3_ID:
volume['status'] = 'in-use'
else:
volume['status'] = 'available'
@ -552,7 +605,7 @@ def stub_upload_volume_to_image_service(self, context, volume, metadata,
"display_description": volume['display_description'],
"size": volume['size'],
"volume_type": volume['volume_type'],
"image_id": 1,
"image_id": fake.IMAGE_ID,
"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name'}
@ -565,7 +618,8 @@ class VolumeImageActionsTest(test.TestCase):
self.controller = volume_actions.VolumeActionsController()
self.stubs.Set(volume_api.API, 'get', stub_volume_get)
self.context = context.RequestContext('fake', 'fake', is_admin=False)
self.context = context.RequestContext(fake.USER_ID, fake.PROJECT_ID,
is_admin=False)
def _get_os_volume_upload_image(self):
vol = {
@ -587,7 +641,7 @@ class VolumeImageActionsTest(test.TestCase):
'created_at': datetime.datetime(1, 1, 1, 1, 1, 1),
'disk_format': u'raw',
'updated_at': datetime.datetime(1, 1, 1, 1, 1, 1),
'id': 1,
'id': fake.IMAGE_ID,
'min_ram': 0,
'checksum': None,
'min_disk': 0,
@ -624,13 +678,14 @@ class VolumeImageActionsTest(test.TestCase):
"copy_volume_to_image",
stub_upload_volume_to_image_service)
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
res_dict = self.controller._volume_upload_image(req, id, body)
expected = {'os-volume_upload_image':
{'id': id,
@ -640,7 +695,7 @@ class VolumeImageActionsTest(test.TestCase):
'size': 1,
'volume_type': fake_volume.fake_db_volume_type(
name='vol_type_name'),
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'image_name'}}
@ -652,13 +707,14 @@ class VolumeImageActionsTest(test.TestCase):
self.stubs.Set(volume_api.API, 'get', stub_volume_get_raise_exc)
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.WILL_NOT_BE_FOUND_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPNotFound,
self.controller._volume_upload_image,
req,
@ -673,13 +729,14 @@ class VolumeImageActionsTest(test.TestCase):
"copy_volume_to_image",
stub_upload_volume_to_image_service_raise)
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
@ -694,13 +751,14 @@ class VolumeImageActionsTest(test.TestCase):
"copy_volume_to_image",
stub_upload_volume_to_image_service_raise)
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, fake.VOLUME_ID))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
@ -715,13 +773,14 @@ class VolumeImageActionsTest(test.TestCase):
"copy_volume_to_image",
stub_upload_volume_to_image_service_raise)
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": 'image_name',
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
@ -729,33 +788,36 @@ class VolumeImageActionsTest(test.TestCase):
body)
def test_volume_upload_image_typeerror(self):
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
body = {"os-volume_upload_image_fake": "fake"}
req = webob.Request.blank('/v2/tenant1/volumes/%s/action' % id)
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_volume_upload_image_without_type(self):
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": None,
"force": True}
body = {"": vol}
req = webob.Request.blank('/v2/tenant1/volumes/%s/action' % id)
req = webob.Request.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(fake_auth_context=self.context))
self.assertEqual(400, res.status_int)
def test_extend_volume_valueerror(self):
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
body = {'os-extend': {'new_size': 'fake'}}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._extend,
req,
@ -763,13 +825,14 @@ class VolumeImageActionsTest(test.TestCase):
body)
def test_copy_volume_to_image_notimagename(self):
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
vol = {"container_format": 'bare',
"disk_format": 'raw',
"image_name": None,
"force": True}
body = {"os-volume_upload_image": vol}
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._volume_upload_image,
req,
@ -778,11 +841,11 @@ class VolumeImageActionsTest(test.TestCase):
def test_copy_volume_to_image_with_protected_prop(self):
"""Test create image from volume with protected properties."""
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME2_ID
def fake_get_volume_image_metadata(*args):
meta_dict = {
"volume_id": id,
"volume_id": fake.VOLUME2_ID,
"key": "x_billing_code_license",
"value": "246254365"}
return meta_dict
@ -809,7 +872,8 @@ class VolumeImageActionsTest(test.TestCase):
self.fake_rpc_copy_volume_to_image
req = fakes.HTTPRequest.blank(
'/v2/tenant1/volumes/%s/action' % id)
'/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, id))
body = self._get_os_volume_upload_image()
res_dict = self.controller._volume_upload_image(req,
id,
@ -825,7 +889,7 @@ class VolumeImageActionsTest(test.TestCase):
'size': 1,
'volume_type': fake_volume.fake_db_volume_type(
name='vol_type_name'),
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'image_name'
@ -853,7 +917,7 @@ class VolumeImageActionsTest(test.TestCase):
In this case volume glance metadata will not be available for this
volume.
"""
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME_ID
def fake_get_volume_image_metadata_raise(*args):
raise exception.GlanceMetadataNotFound(id=id)
@ -880,7 +944,7 @@ class VolumeImageActionsTest(test.TestCase):
self.fake_rpc_copy_volume_to_image
req = fakes.HTTPRequest.blank(
'/v2/tenant1/volumes/%s/action' % id)
'/v2/%s/volumes/%s/action' % (fake.PROJECT_ID, id))
body = self._get_os_volume_upload_image()
res_dict = self.controller._volume_upload_image(req,
id,
@ -896,7 +960,7 @@ class VolumeImageActionsTest(test.TestCase):
'size': 1,
'volume_type': fake_volume.fake_db_volume_type(
name='vol_type_name'),
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'image_name'
@ -907,7 +971,7 @@ class VolumeImageActionsTest(test.TestCase):
def test_copy_volume_to_image_without_protected_prop(self):
"""Test protected property is not defined with the root image."""
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME_ID
def fake_get_volume_image_metadata(*args):
return []
@ -934,8 +998,8 @@ class VolumeImageActionsTest(test.TestCase):
self.fake_rpc_copy_volume_to_image
req = fakes.HTTPRequest.blank(
'/v2/tenant1/volumes/%s/action' % id)
'/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, id))
body = self._get_os_volume_upload_image()
res_dict = self.controller._volume_upload_image(req,
id,
@ -951,7 +1015,7 @@ class VolumeImageActionsTest(test.TestCase):
'size': 1,
'volume_type': fake_volume.fake_db_volume_type(
name='vol_type_name'),
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'image_name'
@ -962,7 +1026,7 @@ class VolumeImageActionsTest(test.TestCase):
def test_copy_volume_to_image_without_core_prop(self):
"""Test glance_core_properties defined in cinder.conf is empty."""
id = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
id = fake.VOLUME_ID
# Need to mock create, update, copy_volume_to_image
with mock.patch.object(glance.GlanceImageService, "create") \
@ -982,7 +1046,7 @@ class VolumeImageActionsTest(test.TestCase):
self.override_config('glance_core_properties', [])
req = fakes.HTTPRequest.blank(
'/v2/tenant1/volumes/%s/action' % id)
'/v2/%s/volumes/%s/action' % (fake.PROJECT_ID, id))
body = self._get_os_volume_upload_image()
res_dict = self.controller._volume_upload_image(req,
@ -999,7 +1063,7 @@ class VolumeImageActionsTest(test.TestCase):
'size': 1,
'volume_type': fake_volume.fake_db_volume_type(
name='vol_type_name'),
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': 'bare',
'disk_format': 'raw',
'image_name': 'image_name'
@ -1032,18 +1096,19 @@ class VolumeImageActionsTest(test.TestCase):
mock_copy_volume_to_image.side_effect = (
self.fake_rpc_copy_volume_to_image)
req = fakes.HTTPRequest.blank('/v2/tenant1/volumes/%s/action' % id)
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' %
(fake.PROJECT_ID, id))
body = self._get_os_volume_upload_image()
res_dict = self.controller._volume_upload_image(req, id, body)
expected_res = {
'os-volume_upload_image': {
'id': fake.volume_id,
'id': fake.VOLUME_ID,
'updated_at': None,
'status': 'uploading',
'display_description': None,
'size': 1,
'volume_type': None,
'image_id': 1,
'image_id': fake.IMAGE_ID,
'container_format': u'bare',
'disk_format': u'raw',
'image_name': u'image_name'

31
cinder/tests/unit/api/contrib/test_volume_host_attribute.py
View File

@ -23,13 +23,14 @@ from cinder import db
from cinder import objects
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder import volume
def fake_db_volume_get(*args, **kwargs):
return {
'id': 'fake',
'id': fake.VOLUME_ID,
'host': 'host001',
'status': 'available',
'size': 5,
@ -39,15 +40,15 @@ def fake_db_volume_get(*args, **kwargs):
'display_description': 'Just another volume!',
'volume_type_id': None,
'snapshot_id': None,
'project_id': 'fake',
'project_id': fake.PROJECT_ID,
'migration_status': None,
'_name_id': 'fake2',
'_name_id': fake.VOLUME2_ID,
'attach_status': 'detached',
}
def fake_volume_api_get(*args, **kwargs):
ctx = context.RequestContext('admin', 'fake', True)
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, True)
db_volume = fake_db_volume_get()
return fake_volume.fake_volume_obj(ctx, **db_volume)
@ -75,8 +76,9 @@ class VolumeHostAttributeTest(test.TestCase):
self.UUID = uuid.uuid4()
def test_get_volume_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -84,8 +86,9 @@ class VolumeHostAttributeTest(test.TestCase):
self.assertEqual('host001', vol['os-vol-host-attr:host'])
def test_get_volume_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -93,8 +96,8 @@ class VolumeHostAttributeTest(test.TestCase):
self.assertNotIn('os-vol-host-attr:host', vol)
def test_list_detail_volumes_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -102,8 +105,8 @@ class VolumeHostAttributeTest(test.TestCase):
self.assertEqual('host001', vol[0]['os-vol-host-attr:host'])
def test_list_detail_volumes_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -111,8 +114,8 @@ class VolumeHostAttributeTest(test.TestCase):
self.assertNotIn('os-vol-host-attr:host', vol[0])
def test_list_simple_volumes_no_host(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes')
ctx = context.RequestContext(fake.user_id, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())

99
cinder/tests/unit/api/contrib/test_volume_image_metadata.py
View File

@ -25,13 +25,14 @@ from cinder import exception
from cinder import objects
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder import volume
def fake_db_volume_get(*args, **kwargs):
return {
'id': 'fake',
'id': fake.VOLUME_ID,
'host': 'host001',
'status': 'available',
'size': 5,
@ -41,15 +42,15 @@ def fake_db_volume_get(*args, **kwargs):
'display_description': 'Just another volume!',
'volume_type_id': None,
'snapshot_id': None,
'project_id': 'fake',
'project_id': fake.PROJECT_ID,
'migration_status': None,
'_name_id': 'fake2',
'_name_id': fake.VOLUME2_ID,
'attach_status': 'detached',
}
def fake_volume_api_get(*args, **kwargs):
ctx = context.RequestContext('admin', 'fake', True)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
db_volume = fake_db_volume_get()
return fake_volume.fake_volume_obj(ctx, **db_volume)
@ -63,7 +64,7 @@ def fake_volume_get_all_empty(*args, **kwargs):
fake_image_metadata = {
'image_id': 'someid',
'image_id': fake.IMAGE_ID,
'image_name': 'fake',
'kernel_id': 'somekernel',
'ramdisk_id': 'someramdisk',
@ -109,11 +110,14 @@ class VolumeImageMetadataTest(test.TestCase):
self.UUID = uuid.uuid4()
self.controller = (volume_image_metadata.
VolumeImageMetadataController())
self.user_ctxt = context.RequestContext(
fake.USER_ID, fake.PROJECT_ID, auth_token=True)
def _make_request(self, url):
req = webob.Request.blank(url)
req.accept = self.content_type
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
return res
def _get_image_metadata(self, body):
@ -127,26 +131,29 @@ class VolumeImageMetadataTest(test.TestCase):
def _create_volume_and_glance_metadata(self):
ctxt = context.get_admin_context()
db.volume_create(ctxt, {'id': 'fake', 'status': 'available',
db.volume_create(ctxt, {'id': fake.VOLUME_ID, 'status': 'available',
'host': 'test', 'provider_location': '',
'size': 1})
db.volume_glance_metadata_create(ctxt, 'fake', 'image_id', 'someid')
db.volume_glance_metadata_create(ctxt, 'fake', 'image_name', 'fake')
db.volume_glance_metadata_create(ctxt, 'fake', 'kernel_id',
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID,
'image_id', fake.IMAGE_ID)
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID,
'image_name', 'fake')
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'kernel_id',
'somekernel')
db.volume_glance_metadata_create(ctxt, 'fake', 'ramdisk_id',
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID, 'ramdisk_id',
'someramdisk')
def test_get_volume(self):
self._create_volume_and_glance_metadata()
res = self._make_request('/v2/fake/volumes/%s' % self.UUID)
res = self._make_request('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
self._get_image_metadata(res.body))
def test_list_detail_volumes(self):
self._create_volume_and_glance_metadata()
res = self._make_request('/v2/fake/volumes/detail')
res = self._make_request('/v2/%s/volumes/detail' % fake.PROJECT_ID)
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
self._get_image_metadata_list(res.body)[0])
@ -159,18 +166,21 @@ class VolumeImageMetadataTest(test.TestCase):
fake_dont_call_this)
self.stubs.Set(volume.api.API, 'get_all', fake_volume_get_all_empty)
res = self._make_request('/v2/fake/volumes/detail')
res = self._make_request('/v2/%s/volumes/detail' % fake.PROJECT_ID)
self.assertEqual(200, res.status_int)
self.assertFalse(fake_dont_call_this.called)
def test_list_detail_volumes_with_limit(self):
ctxt = context.get_admin_context()
db.volume_create(ctxt, {'id': 'fake', 'status': 'available',
db.volume_create(ctxt, {'id': fake.VOLUME_ID, 'status': 'available',
'host': 'test', 'provider_location': '',
'size': 1})
db.volume_glance_metadata_create(ctxt, 'fake', 'key1', 'value1')
db.volume_glance_metadata_create(ctxt, 'fake', 'key2', 'value2')
res = self._make_request('/v2/fake/volumes/detail?limit=1')
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID,
'key1', 'value1')
db.volume_glance_metadata_create(ctxt, fake.VOLUME_ID,
'key2', 'value2')
res = self._make_request('/v2/%s/volumes/detail?limit=1'
% fake.PROJECT_ID)
self.assertEqual(200, res.status_int)
self.assertEqual({'key1': 'value1', 'key2': 'value2'},
self._get_image_metadata_list(res.body)[0])
@ -182,12 +192,14 @@ class VolumeImageMetadataTest(test.TestCase):
fake_create_volume_metadata)
body = {"os-set_image_metadata": {"metadata": fake_image_metadata}}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = "POST"
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
jsonutils.loads(res.body)["metadata"])
@ -211,28 +223,32 @@ class VolumeImageMetadataTest(test.TestCase):
},
}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
jsonutils.loads(res.body)["metadata"])
def test_create_empty_body(self):
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.headers["content-type"] = "application/json"
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, req, 1, None)
self.controller.create, req, fake.VOLUME_ID, None)
def test_create_nonexistent_volume(self):
self.stubs.Set(volume.api.API, 'get', return_volume_nonexistent)
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.content_type = "application/json"
body = {"os-set_image_metadata": {
@ -240,12 +256,13 @@ class VolumeImageMetadataTest(test.TestCase):
}
req.body = jsonutils.dump_as_bytes(body)
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create, req, 1, body)
self.controller.create, req, fake.VOLUME_ID, body)
def test_invalid_metadata_items_on_create(self):
self.stubs.Set(db, 'volume_metadata_update',
fake_create_volume_metadata)
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.headers["content-type"] = "application/json"
@ -256,7 +273,7 @@ class VolumeImageMetadataTest(test.TestCase):
# Test for long key
req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create, req, 1, data)
self.controller.create, req, fake.VOLUME_ID, data)
# Test for long value
data = {"os-set_image_metadata": {
@ -264,7 +281,7 @@ class VolumeImageMetadataTest(test.TestCase):
}
req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPRequestEntityTooLarge,
self.controller.create, req, 1, data)
self.controller.create, req, fake.VOLUME_ID, data)
# Test for empty key.
data = {"os-set_image_metadata": {
@ -272,7 +289,7 @@ class VolumeImageMetadataTest(test.TestCase):
}
req.body = jsonutils.dump_as_bytes(data)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create, req, 1, data)
self.controller.create, req, fake.VOLUME_ID, data)
def test_delete(self):
self.stubs.Set(db, 'volume_metadata_delete',
@ -281,25 +298,28 @@ class VolumeImageMetadataTest(test.TestCase):
body = {"os-unset_image_metadata": {
"key": "ramdisk_id"}
}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
self.assertEqual(200, res.status_int)
def test_delete_meta_not_found(self):
data = {"os-unset_image_metadata": {
"key": "invalid_id"}
}
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(data)
req.headers["content-type"] = "application/json"
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.delete, req, 1, data)
self.controller.delete, req, fake.VOLUME_ID, data)
def test_delete_nonexistent_volume(self):
self.stubs.Set(db, 'volume_metadata_delete',
@ -308,22 +328,25 @@ class VolumeImageMetadataTest(test.TestCase):
body = {"os-unset_image_metadata": {
"key": "fake"}
}
req = fakes.HTTPRequest.blank('/v2/fake/volumes/1/action')
req = fakes.HTTPRequest.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.delete, req, 1, body)
self.controller.delete, req, fake.VOLUME_ID, body)
def test_show_image_metadata(self):
body = {"os-show_image_metadata": None}
req = webob.Request.blank('/v2/fake/volumes/1/action')
req = webob.Request.blank('/v2/%s/volumes/%s/action' % (
fake.PROJECT_ID, fake.VOLUME_ID))
req.method = 'POST'
req.body = jsonutils.dump_as_bytes(body)
req.headers["content-type"] = "application/json"
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
self.assertEqual(200, res.status_int)
self.assertEqual(fake_image_metadata,
jsonutils.loads(res.body)["metadata"])

23
cinder/tests/unit/api/contrib/test_volume_manage.py
View File

@ -20,6 +20,7 @@ from cinder import context
from cinder import exception
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
@ -46,7 +47,7 @@ def db_service_get_by_host_and_topic(context, host, topic):
# Some of the tests check that volume types are correctly validated during a
# volume manage operation. This data structure represents an existing volume
# type.
fake_vt = {'id': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
fake_vt = {'id': fake.VOLUME_TYPE_ID,
'name': 'good_fakevt'}
@ -83,16 +84,16 @@ def api_manage(*args, **kwargs):
Note that we don't try to replicate any passed-in information (e.g. name,
volume type) in the returned structure.
"""
ctx = context.RequestContext('admin', 'fake', True)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
vol = {
'status': 'creating',
'display_name': 'fake_name',
'availability_zone': 'nova',
'tenant_id': 'fake',
'id': 'ffffffff-0000-ffff-0000-ffffffffffff',
'tenant_id': fake.PROJECT_ID,
'id': fake.VOLUME_ID,
'volume_type': None,
'snapshot_id': None,
'user_id': 'fake',
'user_id': fake.USER_ID,
'size': 0,
'attach_status': 'detached',
'volume_type_id': None}
@ -124,11 +125,11 @@ class VolumeManageTest(test.TestCase):
def _get_resp(self, body):
"""Helper to execute an os-volume-manage API call."""
req = webob.Request.blank('/v2/fake/os-volume-manage')
req = webob.Request.blank('/v2/%s/os-volume-manage' % fake.PROJECT_ID)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.environ['cinder.context'] = context.RequestContext('admin',
'fake',
req.environ['cinder.context'] = context.RequestContext(fake.USER_ID,
fake.PROJECT_ID,
True)
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(app())
@ -181,8 +182,7 @@ class VolumeManageTest(test.TestCase):
"""
body = {'volume': {'host': 'host_ok',
'ref': 'fake_ref',
'volume_type':
'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'}}
'volume_type': fake.VOLUME_TYPE_ID}}
res = self._get_resp(body)
self.assertEqual(202, res.status_int, res)
self.assertTrue(mock_validate.called)
@ -209,8 +209,7 @@ class VolumeManageTest(test.TestCase):
"""Test failure on nonexistent volume type specified by ID."""
body = {'volume': {'host': 'host_ok',
'ref': 'fake_ref',
'volume_type':
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'}}
'volume_type': fake.WILL_NOT_BE_FOUND_ID}}
res = self._get_resp(body)
self.assertEqual(404, res.status_int, res)
pass

43
cinder/tests/unit/api/contrib/test_volume_migration_status_attribute.py
View File

@ -22,13 +22,14 @@ from cinder import context
from cinder import objects
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder import volume
def fake_db_volume_get(*args, **kwargs):
return {
'id': 'fake',
'id': fake.VOLUME_ID,
'host': 'host001',
'status': 'available',
'size': 5,
@ -39,14 +40,14 @@ def fake_db_volume_get(*args, **kwargs):
'display_description': 'Just another volume!',
'volume_type_id': None,
'snapshot_id': None,
'project_id': 'fake',
'project_id': fake.PROJECT_ID,
'migration_status': 'migrating',
'_name_id': 'fake2',
'_name_id': fake.VOLUME2_ID,
}
def fake_volume_api_get(*args, **kwargs):
ctx = context.RequestContext('admin', 'fake', True)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
db_volume = fake_db_volume_get()
return fake_volume.fake_volume_obj(ctx, **db_volume)
@ -72,18 +73,22 @@ class VolumeMigStatusAttributeTest(test.TestCase):
self.UUID = uuid.uuid4()
def test_get_volume_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
vol = jsonutils.loads(res.body)['volume']
self.assertEqual('migrating', vol['os-vol-mig-status-attr:migstat'])
self.assertEqual('fake2', vol['os-vol-mig-status-attr:name_id'])
self.assertEqual('migrating',
vol['os-vol-mig-status-attr:migstat'])
self.assertEqual(fake.VOLUME2_ID,
vol['os-vol-mig-status-attr:name_id'])
def test_get_volume_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -92,18 +97,20 @@ class VolumeMigStatusAttributeTest(test.TestCase):
self.assertNotIn('os-vol-mig-status-attr:name_id', vol)
def test_list_detail_volumes_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
vol = jsonutils.loads(res.body)['volumes']
self.assertEqual('migrating', vol[0]['os-vol-mig-status-attr:migstat'])
self.assertEqual('fake2', vol[0]['os-vol-mig-status-attr:name_id'])
self.assertEqual('migrating',
vol[0]['os-vol-mig-status-attr:migstat'])
self.assertEqual(fake.VOLUME2_ID,
vol[0]['os-vol-mig-status-attr:name_id'])
def test_list_detail_volumes_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -112,8 +119,8 @@ class VolumeMigStatusAttributeTest(test.TestCase):
self.assertNotIn('os-vol-mig-status-attr:name_id', vol[0])
def test_list_simple_volumes_no_migration_status(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())

27
cinder/tests/unit/api/contrib/test_volume_tenant_attribute.py
View File

@ -21,6 +21,7 @@ from cinder import context
from cinder import objects
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import fake_volume
from cinder import volume
@ -29,9 +30,9 @@ PROJECT_ID = '88fd1da4-f464-4a87-9ce5-26f2f40743b9'
def fake_volume_get(*args, **kwargs):
ctx = context.RequestContext('non-admin', 'fake', False)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, False)
vol = {
'id': 'fake',
'id': fake.volume_id,
'project_id': PROJECT_ID,
}
return fake_volume.fake_volume_obj(ctx, **vol)
@ -58,8 +59,9 @@ class VolumeTenantAttributeTest(test.TestCase):
self.UUID = uuid.uuid4()
def test_get_volume_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -67,8 +69,9 @@ class VolumeTenantAttributeTest(test.TestCase):
self.assertEqual(PROJECT_ID, vol['os-vol-tenant-attr:tenant_id'])
def test_get_volume_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/%s' % self.UUID)
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/%s' % (
fake.PROJECT_ID, self.UUID))
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -76,8 +79,8 @@ class VolumeTenantAttributeTest(test.TestCase):
self.assertNotIn('os-vol-tenant-attr:tenant_id', vol)
def test_list_detail_volumes_allowed(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -85,8 +88,8 @@ class VolumeTenantAttributeTest(test.TestCase):
self.assertEqual(PROJECT_ID, vol[0]['os-vol-tenant-attr:tenant_id'])
def test_list_detail_volumes_unallowed(self):
ctx = context.RequestContext('non-admin', 'fake', False)
req = webob.Request.blank('/v2/fake/volumes/detail')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, False)
req = webob.Request.blank('/v2/%s/volumes/detail' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())
@ -94,8 +97,8 @@ class VolumeTenantAttributeTest(test.TestCase):
self.assertNotIn('os-vol-tenant-attr:tenant_id', vol[0])
def test_list_simple_volumes_no_tenant_id(self):
ctx = context.RequestContext('admin', 'fake', True)
req = webob.Request.blank('/v2/fake/volumes')
ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
req = webob.Request.blank('/v2/%s/volumes' % fake.PROJECT_ID)
req.method = 'GET'
req.environ['cinder.context'] = ctx
res = req.get_response(app())

147
cinder/tests/unit/api/contrib/test_volume_transfer.py
View File

@ -28,6 +28,7 @@ from cinder import db
from cinder import exception
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
import cinder.transfer
@ -38,8 +39,10 @@ class VolumeTransferAPITestCase(test.TestCase):
super(VolumeTransferAPITestCase, self).setUp()
self.volume_transfer_api = cinder.transfer.API()
self.controller = volume_transfer.VolumeTransferController()
self.user_ctxt = context.RequestContext(
fake.USER_ID, fake.PROJECT_ID, auth_token=True, admin=True)
def _create_transfer(self, volume_id=1,
def _create_transfer(self, volume_id=fake.VOLUME_ID,
display_name='test_transfer'):
"""Create a transfer object."""
return self.volume_transfer_api.create(context.get_admin_context(),
@ -51,12 +54,12 @@ class VolumeTransferAPITestCase(test.TestCase):
display_description='this is a test volume',
status='available',
size=1,
project_id='fake'):
project_id=fake.PROJECT_ID):
"""Create a volume object."""
vol = {}
vol['host'] = 'fake_host'
vol['size'] = size
vol['user_id'] = 'fake'
vol['user_id'] = fake.USER_ID
vol['project_id'] = project_id
vol['status'] = status
vol['display_name'] = display_name
@ -68,11 +71,12 @@ class VolumeTransferAPITestCase(test.TestCase):
def test_show_transfer(self):
volume_id = self._create_volume(size=5)
transfer = self._create_transfer(volume_id)
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
self.assertEqual('test_transfer', res_dict['transfer']['name'])
@ -83,15 +87,18 @@ class VolumeTransferAPITestCase(test.TestCase):
db.volume_destroy(context.get_admin_context(), volume_id)
def test_show_transfer_with_transfer_NotFound(self):
req = webob.Request.blank('/v2/fake/os-volume-transfer/1234')
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s' % (
fake.PROJECT_ID, fake.WILL_NOT_BE_FOUND_ID))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
self.assertEqual('Transfer 1234 could not be found.',
self.assertEqual('Transfer %s could not be found.' %
fake.WILL_NOT_BE_FOUND_ID,
res_dict['itemNotFound']['message'])
def test_list_transfers_json(self):
@ -100,10 +107,12 @@ class VolumeTransferAPITestCase(test.TestCase):
transfer1 = self._create_transfer(volume_id_1)
transfer2 = self._create_transfer(volume_id_2)
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
@ -124,11 +133,13 @@ class VolumeTransferAPITestCase(test.TestCase):
transfer1 = self._create_transfer(volume_id_1)
transfer2 = self._create_transfer(volume_id_2)
req = webob.Request.blank('/v2/fake/os-volume-transfer/detail')
req = webob.Request.blank('/v2/%s/os-volume-transfer/detail' %
fake.PROJECT_ID)
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(200, res.status_int)
@ -151,12 +162,12 @@ class VolumeTransferAPITestCase(test.TestCase):
def test_list_transfers_with_all_tenants(self):
volume_id_1 = self._create_volume(size=5)
volume_id_2 = self._create_volume(size=5, project_id='fake1')
volume_id_2 = self._create_volume(size=5, project_id=fake.PROJECT_ID)
transfer1 = self._create_transfer(volume_id_1)
transfer2 = self._create_transfer(volume_id_2)
req = fakes.HTTPRequest.blank('/v2/fake/os-volume-transfer?'
'all_tenants=1',
req = fakes.HTTPRequest.blank('/v2/%s/os-volume-transfer?'
'all_tenants=1' % fake.PROJECT_ID,
use_admin_context=True)
res_dict = self.controller.index(req)
@ -178,11 +189,13 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"transfer": {"name": "transfer1",
"volume_id": volume_id}}
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
@ -197,12 +210,14 @@ class VolumeTransferAPITestCase(test.TestCase):
db.volume_destroy(context.get_admin_context(), volume_id)
def test_create_transfer_with_no_body(self):
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.body = jsonutils.dump_as_bytes(None)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
@ -213,11 +228,13 @@ class VolumeTransferAPITestCase(test.TestCase):
def test_create_transfer_with_body_KeyError(self):
body = {"transfer": {"name": "transfer1"}}
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
@ -229,11 +246,13 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"transfer": {"name": "transfer1",
"volume_id": 1234}}
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
@ -245,11 +264,13 @@ class VolumeTransferAPITestCase(test.TestCase):
volume_id = self._create_volume(status='attached')
body = {"transfer": {"name": "transfer1",
"volume_id": volume_id}}
req = webob.Request.blank('/v2/fake/os-volume-transfer')
req = webob.Request.blank('/v2/%s/os-volume-transfer' %
fake.PROJECT_ID)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
@ -262,20 +283,22 @@ class VolumeTransferAPITestCase(test.TestCase):
def test_delete_transfer_awaiting_transfer(self):
volume_id = self._create_volume()
transfer = self._create_transfer(volume_id)
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
self.assertEqual(202, res.status_int)
# verify transfer has been deleted
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'GET'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
@ -288,15 +311,18 @@ class VolumeTransferAPITestCase(test.TestCase):
db.volume_destroy(context.get_admin_context(), volume_id)
def test_delete_transfer_with_transfer_NotFound(self):
req = webob.Request.blank('/v2/fake/os-volume-transfer/9999')
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s' % (
fake.PROJECT_ID, fake.WILL_NOT_BE_FOUND_ID))
req.method = 'DELETE'
req.headers['Content-Type'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
self.assertEqual('Transfer 9999 could not be found.',
self.assertEqual('Transfer %s could not be found.' %
fake.WILL_NOT_BE_FOUND_ID,
res_dict['itemNotFound']['message'])
def test_accept_transfer_volume_id_specified_json(self):
@ -306,12 +332,13 @@ class VolumeTransferAPITestCase(test.TestCase):
svc = self.start_service('volume', host='fake_host')
body = {"accept": {"id": transfer['id'],
"auth_key": transfer['auth_key']}}
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(202, res.status_int)
@ -324,13 +351,14 @@ class VolumeTransferAPITestCase(test.TestCase):
volume_id = self._create_volume(size=5)
transfer = self._create_transfer(volume_id)
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
req.body = jsonutils.dump_as_bytes(None)
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
@ -344,14 +372,15 @@ class VolumeTransferAPITestCase(test.TestCase):
volume_id = self._create_volume(size=5)
transfer = self._create_transfer(volume_id)
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
body = {"": {}}
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.headers['Accept'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
@ -366,12 +395,13 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"accept": {"id": transfer['id'],
"auth_key": 1}}
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(400, res.status_int)
@ -389,16 +419,19 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"accept": {"id": transfer['id'],
"auth_key": 1}}
req = webob.Request.blank('/v2/fake/os-volume-transfer/1/accept')
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, fake.WILL_NOT_BE_FOUND_ID))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(404, res.status_int)
self.assertEqual(404, res_dict['itemNotFound']['code'])
self.assertEqual('TransferNotFound: Transfer 1 could not be found.',
self.assertEqual('TransferNotFound: Transfer %s could not be found.' %
fake.WILL_NOT_BE_FOUND_ID,
res_dict['itemNotFound']['message'])
db.transfer_destroy(context.get_admin_context(), transfer['id'])
@ -422,13 +455,14 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"accept": {"id": transfer['id'],
"auth_key": transfer['auth_key']}}
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)
@ -454,13 +488,14 @@ class VolumeTransferAPITestCase(test.TestCase):
body = {"accept": {"id": transfer['id'],
"auth_key": transfer['auth_key']}}
req = webob.Request.blank('/v2/fake/os-volume-transfer/%s/accept' %
transfer['id'])
req = webob.Request.blank('/v2/%s/os-volume-transfer/%s/accept' % (
fake.PROJECT_ID, transfer['id']))
req.method = 'POST'
req.headers['Content-Type'] = 'application/json'
req.body = jsonutils.dump_as_bytes(body)
res = req.get_response(fakes.wsgi_app())
res = req.get_response(fakes.wsgi_app(
fake_auth_context=self.user_ctxt))
res_dict = jsonutils.loads(res.body)
self.assertEqual(413, res.status_int)

5
cinder/tests/unit/api/contrib/test_volume_unmanage.py
View File

@ -21,6 +21,7 @@ from cinder import db
from cinder import objects
from cinder import test
from cinder.tests.unit.api import fakes
from cinder.tests.unit import fake_constants as fake
from cinder.tests.unit import utils
@ -40,7 +41,7 @@ class VolumeUnmanageTest(test.TestCase):
def setUp(self):
super(VolumeUnmanageTest, self).setUp()
self.ctxt = context.RequestContext('admin', 'fake_project', True)
self.ctxt = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
api = fakes.router.APIRouter()
self.app = fakes.urlmap.URLMap()
@ -72,7 +73,7 @@ class VolumeUnmanageTest(test.TestCase):
def test_unmanage_volume_bad_volume_id(self):
"""Return 404 if the volume does not exist."""
res = self._get_resp('nonexistent-volume-id')
res = self._get_resp(fake.WILL_NOT_BE_FOUND_ID)
self.assertEqual(404, res.status_int, res)
def test_unmanage_volume_attached(self):