From fa7130388f3fd3862bea3d58bed02ecc46fbf4f9 Mon Sep 17 00:00:00 2001 From: Yikun Jiang Date: Fri, 21 Dec 2018 17:35:31 +0800 Subject: [PATCH] Add policy test for volume extend/retype/update_ro policy Add policy test for volume extend/retype/update_readonly_flag policy, and remove them from test policy file. This is one of the series patches of policy-in-code test, see more information on [1] and [2]. [1] 530fb9319ce21b7ff99e55f095c04f13f0785842 [2] f207bac80924ffaf6d4c2a500c295d0e2e71966e Change-Id: I52f2cea01a84cb398fedddb8dbeaca377baaeb95 --- .../unit/api/v3/test_volume_protection.py | 203 +++++++++++++++++- cinder/tests/unit/policy.json | 4 - 2 files changed, 200 insertions(+), 7 deletions(-) diff --git a/cinder/tests/unit/api/v3/test_volume_protection.py b/cinder/tests/unit/api/v3/test_volume_protection.py index 48bb5332d45..b59b88f3447 100644 --- a/cinder/tests/unit/api/v3/test_volume_protection.py +++ b/cinder/tests/unit/api/v3/test_volume_protection.py @@ -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) diff --git a/cinder/tests/unit/policy.json b/cinder/tests/unit/policy.json index 6edf75ff0d8..3d85a2c488c 100644 --- a/cinder/tests/unit/policy.json +++ b/cinder/tests/unit/policy.json @@ -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": "",