class VolumeTypeEncryptionController(wsgi.Controller):
- """The volume type encryption API controller for the OpenStack API """
+ """The volume type encryption API controller for the OpenStack API."""
def _get_volume_type_encryption(self, context, type_id):
encryption_ref = db.volume_type_encryption_get(context, type_id)
'updated_at': literal_column('updated_at')})
-# TODO(joel-coffman): split into two functions -- update and create
@require_admin_context
def volume_type_encryption_update_or_create(context, volume_type_id,
values):
####################
-@require_admin_context
+@require_context
def volume_encryption_metadata_get(context, volume_id, session=None):
"""Return the encryption key id for a given volume."""
self.stubs.Set(db.sqlalchemy.api, 'volume_type_encryption_get',
return_volume_type_encryption_metadata)
- self.ctxt = context.RequestContext('fake', 'fake', is_admin=True)
+ self.ctxt = context.RequestContext('fake', 'fake')
self.volume_id = self._create_volume(self.ctxt)
def tearDown(self):
- db.volume_destroy(self.ctxt, self.volume_id)
+ db.volume_destroy(self.ctxt.elevated(), self.volume_id)
super(VolumeEncryptionMetadataTest, self).tearDown()
def test_index(self):
% bad_volume_id}}
self.assertEqual(expected, res_dict)
- def test_retrieve_key_not_admin(self):
+ def test_retrieve_key_admin(self):
self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: True)
- ctxt = self.ctxt.deepcopy()
- ctxt.is_admin = False
+ ctxt = context.RequestContext('fake', 'fake', is_admin=True)
req = webob.Request.blank('/v2/fake/volumes/%s/encryption/'
'encryption_key_id' % self.volume_id)
res = req.get_response(fakes.wsgi_app(fake_auth_context=ctxt))
- self.assertEqual(403, res.status_code)
- res_dict = json.loads(res.body)
+ self.assertEqual(200, res.status_code)
- expected = {
- 'forbidden': {
- 'code': 403,
- 'message': ("Policy doesn't allow volume_extension:"
- "volume_encryption_metadata to be performed.")
- }
- }
- self.assertEqual(expected, res_dict)
+ self.assertEqual('fake_key', res.body)
def test_show_volume_not_encrypted_type(self):
self.stubs.Set(volume_types, 'is_encrypted', lambda *a, **kw: False)
{
"context_is_admin": [["role:admin"]],
"admin_api": [["is_admin:True"]],
+ "admin_or_owner": [["is_admin:True"], ["project_id:%(project_id)s"]],
"volume:create": [],
"volume:get": [],
"volume_extension:types_manage": [],
"volume_extension:types_extra_specs": [],
"volume_extension:volume_type_encryption": [["rule:admin_api"]],
- "volume_extension:volume_encryption_metadata": [["rule:admin_api"]],
+ "volume_extension:volume_encryption_metadata": [["rule:admin_or_owner"]],
"volume_extension:extended_snapshot_attributes": [],
"volume_extension:volume_image_metadata": [],
"volume_extension:volume_host_attribute": [["rule:admin_api"]],
"volume_extension:types_manage": [["rule:admin_api"]],
"volume_extension:types_extra_specs": [["rule:admin_api"]],
"volume_extension:volume_type_encryption": [["rule:admin_api"]],
- "volume_extension:volume_encryption_metadata": [["rule:admin_api"]],
+ "volume_extension:volume_encryption_metadata": [["rule:admin_or_owner"]],
"volume_extension:extended_snapshot_attributes": [],
"volume_extension:volume_image_metadata": [],