return xmlutil.MasterTemplate(root, 1)
-class VolumeTypeExtraSpecsController(object):
+class VolumeTypeExtraSpecsController(wsgi.Controller):
""" The volume type extra specs API controller for the OpenStack API """
def _get_extra_specs(self, context, type_id):
specs_dict[key] = value
return dict(extra_specs=specs_dict)
- def _check_body(self, body):
- if not body:
- expl = _('No Request Body')
- raise webob.exc.HTTPBadRequest(explanation=expl)
-
def _check_type(self, context, type_id):
try:
volume_types.get_volume_type(context, type_id)
def create(self, req, type_id, body=None):
context = req.environ['cinder.context']
authorize(context)
+
+ if not self.is_valid_body(body, 'extra_specs'):
+ raise webob.exc.HTTPUnprocessableEntity()
+
self._check_type(context, type_id)
- self._check_body(body)
- specs = body.get('extra_specs')
- if not isinstance(specs, dict):
- expl = _('Malformed extra specs')
- raise webob.exc.HTTPBadRequest(explanation=expl)
+
+ specs = body['extra_specs']
db.volume_type_extra_specs_update_or_create(context,
type_id,
specs)
def update(self, req, type_id, id, body=None):
context = req.environ['cinder.context']
authorize(context)
+ if not body:
+ raise webob.exc.HTTPUnprocessableEntity()
self._check_type(context, type_id)
- self._check_body(body)
if not id in body:
expl = _('Request body and URI mismatch')
raise webob.exc.HTTPBadRequest(explanation=expl)
context = req.environ['cinder.context']
authorize(context)
- if not body or body == "":
- raise webob.exc.HTTPUnprocessableEntity()
-
- vol_type = body.get('volume_type', None)
- if vol_type is None or vol_type == "":
+ if not self.is_valid_body(body, 'volume_type'):
raise webob.exc.HTTPUnprocessableEntity()
+ vol_type = body['volume_type']
name = vol_type.get('name', None)
specs = vol_type.get('extra_specs', {})
return xmlutil.MasterTemplate(root, 1)
-class SnapshotsController(object):
+class SnapshotsController(wsgi.Controller):
"""The Volumes API controller for the OpenStack API."""
def __init__(self, ext_mgr=None):
"""Creates a new snapshot."""
context = req.environ['cinder.context']
- if not body:
- return exc.HTTPUnprocessableEntity()
+ if not self.is_valid_body(body, 'snapshot'):
+ raise exc.HTTPUnprocessableEntity()
snapshot = body['snapshot']
volume_id = snapshot['volume_id']
return {'body': {'volume': volume}}
-class VolumeController(object):
+class VolumeController(wsgi.Controller):
"""The Volumes API controller for the OpenStack API."""
def __init__(self, ext_mgr):
@wsgi.deserializers(xml=CreateDeserializer)
def create(self, req, body):
"""Creates a new volume."""
- context = req.environ['cinder.context']
-
- if not body:
+ if not self.is_valid_body(body, 'volume'):
raise exc.HTTPUnprocessableEntity()
+ context = req.environ['cinder.context']
volume = body['volume']
kwargs = {}
self.assertEqual('value1', res_dict['extra_specs']['key1'])
- def test_create_empty_body(self):
- self.stubs.Set(cinder.db,
- 'volume_type_extra_specs_update_or_create',
- return_create_volume_type_extra_specs)
-
- req = fakes.HTTPRequest.blank(self.api_path)
- self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
- req, 1, '')
-
def test_update_item(self):
self.stubs.Set(cinder.db,
'volume_type_extra_specs_update_or_create',
self.assertEqual('value1', res_dict['key1'])
- def test_update_item_empty_body(self):
- self.stubs.Set(cinder.db,
- 'volume_type_extra_specs_update_or_create',
- return_create_volume_type_extra_specs)
-
- req = fakes.HTTPRequest.blank(self.api_path + '/key1')
- self.assertRaises(webob.exc.HTTPBadRequest, self.controller.update,
- req, 1, 'key1', '')
-
def test_update_item_too_many_keys(self):
self.stubs.Set(cinder.db,
'volume_type_extra_specs_update_or_create',
self.assertEqual('key1', tree.tag)
self.assertEqual('value1', tree.text)
self.assertEqual(0, len(tree))
+
+
+class VolumeTypeExtraSpecsUnprocessableEntityTestCase(test.TestCase):
+
+ """
+ Tests of places we throw 422 Unprocessable Entity from
+ """
+
+ def setUp(self):
+ super(VolumeTypeExtraSpecsUnprocessableEntityTestCase, self).setUp()
+ self.controller = types_extra_specs.VolumeTypeExtraSpecsController()
+
+ def _unprocessable_extra_specs_create(self, body):
+ req = fakes.HTTPRequest.blank('/v2/fake/types/1/extra_specs')
+ req.method = 'POST'
+
+ self.assertRaises(webob.exc.HTTPUnprocessableEntity,
+ self.controller.create, req, '1', body)
+
+ def test_create_no_body(self):
+ self._unprocessable_extra_specs_create(body=None)
+
+ def test_create_missing_volume(self):
+ body = {'foo': {'a': 'b'}}
+ self._unprocessable_extra_specs_create(body=body)
+
+ def test_create_malformed_entity(self):
+ body = {'extra_specs': 'string'}
+ self._unprocessable_extra_specs_create(body=body)
+
+ def _unprocessable_extra_specs_update(self, body):
+ req = fakes.HTTPRequest.blank('/v2/fake/types/1/extra_specs')
+ req.method = 'POST'
+
+ self.assertRaises(webob.exc.HTTPUnprocessableEntity,
+ self.controller.update, req, '1', body)
+
+ def test_update_no_body(self):
+ self._unprocessable_extra_specs_update(body=None)
+
+ def test_update_empty_body(self):
+ self._unprocessable_extra_specs_update(body={})
self.assertEqual(1, len(res_dict))
self.assertEqual('vol_type_1', res_dict['volume_type']['name'])
- def test_create_empty_body(self):
- self.stubs.Set(volume_types, 'create',
- return_volume_types_create)
- self.stubs.Set(volume_types, 'get_volume_type_by_name',
- return_volume_types_get_by_name)
- req = fakes.HTTPRequest.blank('/v1/fake/types')
+class VolumeTypesUnprocessableEntityTestCase(test.TestCase):
+
+ """
+ Tests of places we throw 422 Unprocessable Entity from
+ """
+
+ def setUp(self):
+ super(VolumeTypesUnprocessableEntityTestCase, self).setUp()
+ self.controller = types_manage.VolumeTypesManageController()
+
+ def _unprocessable_volume_type_create(self, body):
+ req = fakes.HTTPRequest.blank('/v2/fake/types')
+ req.method = 'POST'
+
self.assertRaises(webob.exc.HTTPUnprocessableEntity,
- self.controller._create, req, '')
+ self.controller._create, req, body)
+
+ def test_create_no_body(self):
+ self._unprocessable_volume_type_create(body=None)
+
+ def test_create_missing_volume(self):
+ body = {'foo': {'a': 'b'}}
+ self._unprocessable_volume_type_create(body=body)
+
+ def test_create_malformed_entity(self):
+ body = {'volume_type': 'string'}
+ self._unprocessable_volume_type_create(body=body)
self.assertEqual(len(raw_snapshots), len(tree))
for idx, child in enumerate(tree):
self._verify_snapshot(raw_snapshots[idx], child)
+
+
+class SnapshotsUnprocessableEntityTestCase(test.TestCase):
+
+ """
+ Tests of places we throw 422 Unprocessable Entity from
+ """
+
+ def setUp(self):
+ super(SnapshotsUnprocessableEntityTestCase, self).setUp()
+ self.controller = snapshots.SnapshotsController()
+
+ def _unprocessable_snapshot_create(self, body):
+ req = fakes.HTTPRequest.blank('/v2/fake/snapshots')
+ req.method = 'POST'
+
+ self.assertRaises(webob.exc.HTTPUnprocessableEntity,
+ self.controller.create, req, body)
+
+ def test_create_no_body(self):
+ self._unprocessable_snapshot_create(body=None)
+
+ def test_create_missing_snapshot(self):
+ body = {'foo': {'a': 'b'}}
+ self._unprocessable_snapshot_create(body=body)
+
+ def test_create_malformed_entity(self):
+ body = {'snapshot': 'string'}
+ self._unprocessable_snapshot_create(body=body)
req,
body)
- def test_volume_create_no_body(self):
- body = {}
- req = fakes.HTTPRequest.blank('/v1/volumes')
- self.assertRaises(webob.exc.HTTPUnprocessableEntity,
- self.controller.create,
- req,
- body)
-
def test_volume_create_with_image_id(self):
self.stubs.Set(volume_api.API, "create", fakes.stub_volume_create)
self.ext_mgr.extensions = {'os-image-create': 'fake'}
},
}
self.assertEquals(request['body'], expected)
+
+
+class VolumesUnprocessableEntityTestCase(test.TestCase):
+
+ """
+ Tests of places we throw 422 Unprocessable Entity from
+ """
+
+ def setUp(self):
+ super(VolumesUnprocessableEntityTestCase, self).setUp()
+ self.ext_mgr = extensions.ExtensionManager()
+ self.ext_mgr.extensions = {}
+ self.controller = volumes.VolumeController(self.ext_mgr)
+
+ def _unprocessable_volume_create(self, body):
+ req = fakes.HTTPRequest.blank('/v2/fake/volumes')
+ req.method = 'POST'
+
+ self.assertRaises(webob.exc.HTTPUnprocessableEntity,
+ self.controller.create, req, body)
+
+ def test_create_no_body(self):
+ self._unprocessable_volume_create(body=None)
+
+ def test_create_missing_volume(self):
+ body = {'foo': {'a': 'b'}}
+ self._unprocessable_volume_create(body=body)
+
+ def test_create_malformed_entity(self):
+ body = {'volume': 'string'}
+ self._unprocessable_volume_create(body=body)
Code Review / openstack-build / cinder-build.git / commitdiff