openstack
/
cinder
Code Issues Proposed changes

Merge "Add policy test for volume extend/retype/update_ro policy"

This commit is contained in:
Zuul 2019-01-17 04:35:34 +00:00 committed by Gerrit Code Review
parent 395dab3d33 fa7130388f
commit cd2b00246b
2 changed files with 200 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

203
cinder/tests/unit/api/v3/test_volume_protection.py
View File

@ -48,10 +48,11 @@ class VolumeProtectionTests(test.TestCase):
)
fake_image.mock_image_service(self)
def _get_request_response(self, context, path, method, body=None):
def _get_request_response(self, context, path, method, body=None,
microversion=mv.BASE_VERSION):
request = webob.Request.blank(path)
request.content_type = 'application/json'
request.headers = mv.get_mv_header(mv.BASE_VERSION)
request.headers = mv.get_mv_header(microversion)
request.method = method
if body:
request.headers["content-type"] = "application/json"
@ -61,7 +62,7 @@ class VolumeProtectionTests(test.TestCase):
)
def _create_fake_volume(self, context, status=None, attach_status=None,
metadata=None):
metadata=None, admin_metadata=None):
vol = {
'display_name': 'fake_volume1',
'status': 'available',
@ -73,10 +74,24 @@ class VolumeProtectionTests(test.TestCase):
vol['attach_status'] = attach_status
if metadata:
vol['metadata'] = metadata
if admin_metadata:
vol['admin_metadata'] = admin_metadata
volume = objects.Volume(context=context, **vol)
volume.create()
return volume
def _create_fake_type(self, context):
vol_type = {
'name': 'fake_volume1',
'extra_specs': {},
'is_public': True,
'projects': [],
'description': 'A fake volume type'
}
volume_type = objects.VolumeType(context=context, **vol_type)
volume_type.create()
return volume_type
@mock.patch.object(volume_api.API, 'get_volume')
def test_admin_can_show_volumes(self, mock_volume):
# Make sure administrators are authorized to list volumes
@ -830,3 +845,185 @@ class VolumeProtectionTests(test.TestCase):
response = self._get_request_response(non_owner_context, path,
'DELETE')
self.assertEqual(http_client.FORBIDDEN, response.status_int)
def test_admin_can_extend_volume(self):
admin_context = self.admin_context
volume = self._create_fake_volume(admin_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': admin_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(admin_context, path, 'POST',
body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
def test_owner_can_extend_volume(self):
user_context = self.user_context
volume = self._create_fake_volume(user_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': user_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(user_context, path, 'POST',
body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
@mock.patch.object(volume_api.API, 'get')
def test_owner_cannot_extend_volume_for_others(self, mock_volume):
user_context = self.user_context
non_owner_context = self.other_user_context
volume = self._create_fake_volume(user_context)
mock_volume.return_value = volume
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': non_owner_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(non_owner_context, path, 'POST',
body=body)
self.assertEqual(http_client.FORBIDDEN, response.status_int)
def test_admin_can_extend_attached_volume(self):
admin_context = self.admin_context
volume = self._create_fake_volume(admin_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': admin_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(
admin_context, path, 'POST', body=body,
microversion=mv.VOLUME_EXTEND_INUSE)
self.assertEqual(http_client.ACCEPTED, response.status_int)
def test_owner_can_extend_attached_volume(self):
user_context = self.user_context
volume = self._create_fake_volume(user_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': user_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(
user_context, path, 'POST', body=body,
microversion=mv.VOLUME_EXTEND_INUSE)
self.assertEqual(http_client.ACCEPTED, response.status_int)
@mock.patch.object(volume_api.API, 'get')
def test_owner_cannot_extend_attached_volume_for_others(self, mock_volume):
user_context = self.user_context
non_owner_context = self.other_user_context
volume = self._create_fake_volume(user_context)
mock_volume.return_value = volume
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': non_owner_context.project_id, 'volume_id': volume.id
}
body = {"os-extend": {"new_size": "2"}}
response = self._get_request_response(
non_owner_context, path, 'POST', body=body,
microversion=mv.VOLUME_EXTEND_INUSE)
self.assertEqual(http_client.FORBIDDEN, response.status_int)
def test_admin_can_retype_volume(self):
admin_context = self.admin_context
volume = self._create_fake_volume(admin_context)
vol_type = self._create_fake_type(admin_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': admin_context.project_id, 'volume_id': volume.id
}
body = {"os-retype": {"new_type": "%s" % vol_type.name,
"migration_policy": "never"}}
response = self._get_request_response(
admin_context, path, 'POST', body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
def test_owner_can_retype_volume(self):
user_context = self.user_context
volume = self._create_fake_volume(user_context)
vol_type = self._create_fake_type(user_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': user_context.project_id, 'volume_id': volume.id
}
body = {"os-retype": {"new_type": "%s" % vol_type.name,
"migration_policy": "never"}}
response = self._get_request_response(
user_context, path, 'POST', body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
@mock.patch.object(volume_api.API, 'get')
def test_owner_cannot_retype_volume_for_others(self, mock_volume):
user_context = self.user_context
non_owner_context = self.other_user_context
volume = self._create_fake_volume(user_context)
mock_volume.return_value = volume
vol_type = self._create_fake_type(user_context)
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': non_owner_context.project_id, 'volume_id': volume.id
}
body = {"os-retype": {"new_type": "%s" % vol_type.name,
"migration_policy": "never"}}
response = self._get_request_response(
non_owner_context, path, 'POST', body=body)
self.assertEqual(http_client.FORBIDDEN, response.status_int)
def test_admin_can_update_readonly(self):
admin_context = self.admin_context
volume = self._create_fake_volume(
admin_context, admin_metadata={"readonly": "False"})
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': admin_context.project_id, 'volume_id': volume.id
}
body = {"os-update_readonly_flag": {"readonly": "True"}}
response = self._get_request_response(
admin_context, path, 'POST', body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
def test_owner_can_update_readonly(self):
user_context = self.user_context
volume = self._create_fake_volume(
user_context, admin_metadata={"readonly": "False"})
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': user_context.project_id, 'volume_id': volume.id
}
body = {"os-update_readonly_flag": {"readonly": "True"}}
response = self._get_request_response(
user_context, path, 'POST', body=body)
self.assertEqual(http_client.ACCEPTED, response.status_int)
@mock.patch.object(volume_api.API, 'get')
def test_owner_cannot_update_readonly_for_others(self, mock_volume):
user_context = self.user_context
non_owner_context = self.other_user_context
volume = self._create_fake_volume(
user_context, admin_metadata={"readonly": "False"})
mock_volume.return_value = volume
path = '/v3/%(project_id)s/volumes/%(volume_id)s/action' % {
'project_id': non_owner_context.project_id, 'volume_id': volume.id
}
body = {"os-update_readonly_flag": {"readonly": "True"}}
response = self._get_request_response(
non_owner_context, path, 'POST', body=body)
self.assertEqual(http_client.FORBIDDEN, response.status_int)

4
cinder/tests/unit/policy.json
View File

@ -10,10 +10,6 @@
"volume:get_snapshot_metadata": "",
"volume:delete_snapshot_metadata": "",
"volume:update_snapshot_metadata": "",
"volume:extend": "",
"volume:extend_attached_volume": "",
"volume:update_readonly_flag": "",
"volume:retype": "",
"volume:revert_to_snapshot": "",
"volume_extension:volume_actions:upload_image": "",
"volume_extension:types_manage": "",