diff --git a/cinder_tempest_plugin/rbac/v3/base.py b/cinder_tempest_plugin/rbac/v3/base.py index 17644f4..af724a3 100644 --- a/cinder_tempest_plugin/rbac/v3/base.py +++ b/cinder_tempest_plugin/rbac/v3/base.py @@ -1,3 +1,4 @@ + # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at @@ -25,7 +26,6 @@ CONF = config.CONF class VolumeV3RbacBaseTests( api_version_utils.BaseMicroversionTest, test.BaseTestCase ): - identity_version = 'v3' @classmethod @@ -40,7 +40,6 @@ class VolumeV3RbacBaseTests( if not CONF.service_available.cinder: skip_msg = ("%s skipped as Cinder is not available" % cls.__name__) raise cls.skipException(skip_msg) - api_version_utils.check_skip_with_microversion( cls.min_microversion, cls.max_microversion, CONF.volume.min_microversion, CONF.volume.max_microversion) @@ -160,3 +159,29 @@ class VolumeV3RbacBaseTests( test_utils.call_and_ignore_notfound_exc(del_action, resource_id) test_utils.call_and_ignore_notfound_exc( client.wait_for_resource_deletion, resource_id) + + @classmethod + def create_backup( + cls, volume_id, backup_client=None, add_cleanup=True, **kwargs + ): + """Wrapper utility that returns a test backup.""" + if backup_client is None: + backup_client = cls.backups_client + if 'name' not in kwargs: + name = data_utils.rand_name(cls.__class__.__name__ + '-Backup') + kwargs['name'] = name + + backup = backup_client.create_backup( + volume_id=volume_id, **kwargs + )['backup'] + if add_cleanup: + cls.addClassResourceCleanup( + test_utils.call_and_ignore_notfound_exc, + cls.delete_resource, + client=backup_client, + backup_id=backup['id'] + ) + waiters.wait_for_volume_resource_status( + backup_client, backup['id'], 'available' + ) + return backup diff --git a/cinder_tempest_plugin/rbac/v3/test_backups.py b/cinder_tempest_plugin/rbac/v3/test_backups.py new file mode 100644 index 0000000..d309b3e --- /dev/null +++ b/cinder_tempest_plugin/rbac/v3/test_backups.py @@ -0,0 +1,455 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +from oslo_serialization import base64 +from oslo_serialization import jsonutils as json + +from tempest.common import waiters +from tempest import config +from tempest.lib.common.utils import data_utils +from tempest.lib import decorators +from tempest.lib import exceptions + +from cinder_tempest_plugin.rbac.v3 import base as rbac_base + +CONF = config.CONF + + +class RbacV3BackupsTests(rbac_base.VolumeV3RbacBaseTests): + @classmethod + def skip_checks(cls): + super(RbacV3BackupsTests, cls).skip_checks() + if not CONF.volume_feature_enabled.backup: + raise cls.skipException("Cinder backup feature disabled") + + @classmethod + def setup_clients(cls): + super().setup_clients() + admin_client = cls.os_project_admin + cls.admin_backups_client = admin_client.backups_client_latest + cls.admin_volumes_client = admin_client.volumes_client_latest + + @classmethod + def resource_setup(cls): + super(RbacV3BackupsTests, cls).resource_setup() + cls.volume_id = cls.create_volume(client=cls.admin_volumes_client) + backup = cls.create_backup( + volume_id=cls.volume_id, backup_client=cls.admin_backups_client + ) + cls.backup_id = backup['id'] + cls.backup_name = backup['name'] + + +class RbacV3BackupsTests33(RbacV3BackupsTests): + """Test API with microversion greater than 3.3""" + min_microversion = '3.3' + + def _encode_backup(self, backup): + retval = json.dumps(backup) + return base64.encode_as_text(retval) + + def _decode_url(self, backup_url): + return json.loads(base64.decode_as_text(backup_url)) + + def _modify_backup_url(self, backup_url, changes): + backup = self._decode_url(backup_url) + backup.update(changes) + return self._encode_backup(backup) + + def _list_backups(self, expected_status): + """List all backups""" + backups = self.do_request( + method='list_backups', expected_status=expected_status + )['backups'] + backup_list = [ + b['id'] for b in backups if b['name'] == self.backup_name + ] + + self.assertNotEmpty( + backup_list, f"Backup {self.backup_name} not found" + ) + + def _list_project_backups(self, expected_status): + """List all backups for a project""" + backups = self.do_request( + method='list_backups', + expected_status=expected_status, + project_id=self.client.project_id + )['backups'] + backup_list = [ + b['id'] for b in backups if b['name'] == self.backup_name + ] + + self.assertNotEmpty( + backup_list, f"Backup {self.backup_name} not found" + ) + + def _show_backup(self, expected_status): + """Show backup details""" + backup = self.do_request( + method='show_backup', + expected_status=expected_status, + backup_id=self.backup_id + )['backup'] + self.assertNotEmpty(backup, f"Backup {self.backup_name} not found") + + def _delete_backup(self, expected_status): + """Delete a backup""" + add_cleanup = True if expected_status == exceptions.Forbidden\ + else False + volume_id = self.create_volume(client=self.admin_volumes_client) + backup = self.create_backup( + volume_id=volume_id, + backup_client=self.admin_backups_client, + add_cleanup=add_cleanup + ) + + self.do_request( + method='delete_backup', + expected_status=expected_status, + backup_id=backup['id'] + ) + + def _restore_backup(self, expected_status): + """Restore a backup""" + res = self.do_request( + method='restore_backup', + expected_status=expected_status, + backup_id=self.backup_id, + name='new-backup-vol' + ) + if expected_status != exceptions.Forbidden: + waiters.wait_for_volume_resource_status( + self.admin_backups_client, + self.backup_id, 'available' + ) + self.delete_resource( + client=self.admin_volumes_client, + volume_id=res['restore']['volume_id'] + ) + + def _create_backup(self, expected_status): + """Create a backup""" + res = self.do_request( + method='create_backup', + expected_status=expected_status, + volume_id=self.volume_id + ) + if expected_status != exceptions.Forbidden: + backup = res['backup'] + waiters.wait_for_volume_resource_status( + self.admin_backups_client, backup['id'], 'available' + ) + self.admin_backups_client.delete_backup(backup_id=backup['id']) + + def _export_backup(self, expected_status): + """Export a backup""" + self.do_request( + method='export_backup', + expected_status=expected_status, + backup_id=self.backup_id + ) + + def _import_backup(self, expected_status): + """Import a backup""" + volume_id = self.create_volume(client=self.admin_volumes_client) + backup = self.create_backup( + volume_id=volume_id, + backup_client=self.admin_backups_client + ) + + export_backup = ( + self.admin_backups_client.export_backup( + backup['id'] + )['backup-record'] + ) + waiters.wait_for_volume_resource_status( + self.admin_backups_client, backup['id'], 'available' + ) + self.assertTrue( + export_backup['backup_service'].startswith('cinder.backup.drivers') + ) + # NOTE(ybenshim): Backups are imported with the same backup id + # (important for incremental backups among other things), so we cannot + # import the exported backup information as it is, because that Backup + # ID already exists. So we'll fake the data by changing the backup id + # in the exported backup DB info we have retrieved before importing it + # back. + + new_id = data_utils.rand_uuid() + new_url = self._modify_backup_url( + export_backup['backup_url'], {'id': new_id}) + + res = self.do_request( + method='import_backup', + expected_status=expected_status, + backup_service=export_backup['backup_service'], + backup_url=new_url + ) + if expected_status != exceptions.Forbidden: + new_backup = res['backup'] + waiters.wait_for_volume_resource_status( + self.client, new_backup['id'], 'available' + ) + self.delete_resource( + client=self.admin_backups_client, + backup_id=new_backup['id'] + ) + + def _reset_backup_status(self, expected_status): + """Reset a backup status""" + new_status = 'error' + volume_id = self.create_volume(client=self.admin_volumes_client) + backup = self.create_backup( + volume_id=volume_id, + backup_client=self.admin_backups_client + ) + + self.do_request( + method='reset_backup_status', + expected_status=expected_status, + backup_id=backup['id'], + status=new_status + ) + + +class ProjectReaderTests33(RbacV3BackupsTests33): + credentials = ['project_reader', 'project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_reader.backups_client_latest + + @decorators.idempotent_id('9dd02d4b-d6f8-45ca-a95e-534dbd586aab') + def test_list_backups(self): + """List all backups""" + self._list_backups(expected_status=200) + + @decorators.idempotent_id('9ba2e970-c08b-4c1c-b912-2f3b1373ae6e') + def test_list_project_backups(self): + """List all backups for a project""" + self._list_project_backups(expected_status=200) + + @decorators.idempotent_id('e88f8971-2892-4a54-80bb-dd21b18f19e9') + def test_show_backup(self): + """Show backup details""" + self._show_backup(expected_status=200) + + @decorators.skip_because(bug='2017110') + @decorators.idempotent_id('a9ab3279-aa5e-4ad8-b740-b80a7769d3f9') + def test_delete_backup(self): + """Delete a backup""" + self._delete_backup(expected_status=exceptions.Forbidden) + + @decorators.skip_because(bug='2017110') + @decorators.idempotent_id('0566fa4a-4e03-4cca-822f-d5a4922da2ab') + def test_restore_backup(self): + """Restore a backup""" + self._restore_backup(expected_status=exceptions.Forbidden) + + @decorators.skip_because(bug='2017110') + @decorators.idempotent_id('bad2514e-18c0-4fa0-9e35-221182ee24cf') + def test_create_backup(self): + """Create a backup""" + self._create_backup(expected_status=exceptions.Forbidden) + + @decorators.idempotent_id('ab74b8cc-5005-49b4-94f4-994567171b07') + def test_export_backup(self): + """Export a backup""" + self._export_backup(expected_status=exceptions.Forbidden) + + @decorators.idempotent_id('caaa5756-261a-4d9c-bfc2-788719630a06') + def test_import_backup(self): + """Import a backup""" + self._import_backup(expected_status=exceptions.Forbidden) + + @decorators.idempotent_id('c832ff77-8f22-499f-a7a3-0834972a1507') + def test_reset_backup_status(self): + """Reset a backup status""" + self._reset_backup_status(expected_status=exceptions.Forbidden) + + +class ProjectMemberTests33(RbacV3BackupsTests33): + credentials = ['project_member', 'project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_member.backups_client_latest + + @decorators.idempotent_id('5a23c53c-924b-47f6-a5d1-ab6327391c12') + def test_list_backups(self): + """List all backups""" + self._list_backups(expected_status=200) + + @decorators.idempotent_id('c737bd7b-293c-4d8f-ada9-3b00f7e1adce') + def test_list_project_backups(self): + """List all backups for a project""" + self._list_project_backups(expected_status=200) + + @decorators.idempotent_id('9944bb15-02fa-4321-97a4-ef8cb5b5fec2') + def test_show_backup(self): + """Show backup details""" + self._show_backup(expected_status=200) + + @decorators.idempotent_id('c98dfea8-b9f2-4a84-947b-1d857c707789') + def test_delete_backup(self): + """Delete a backup""" + self._delete_backup(expected_status=202) + + @decorators.idempotent_id('7a6fd066-00e7-4140-866c-8195fbd71e87') + def test_restore_backup(self): + """Restore a backup""" + self._restore_backup(expected_status=202) + + @decorators.idempotent_id('44644140-4d05-4725-9a4b-6d1a71eda9b7') + def test_create_backup(self): + """Create a backup""" + self._create_backup(expected_status=202) + + @decorators.idempotent_id('71c7cfaf-7809-4872-b1b2-3feb90b939d4') + def test_export_backup(self): + """Export a backup""" + self._export_backup(expected_status=exceptions.Forbidden) + + @decorators.idempotent_id('f1c03c1b-2b48-4be0-8b6a-81df8a75f78c') + def test_import_backup(self): + """Import a backup""" + self._import_backup(expected_status=exceptions.Forbidden) + + @decorators.idempotent_id('307f6fe9-81ed-444a-9aae-99a571d24bf5') + def test_reset_backup_status(self): + """Reset a backup status""" + self._reset_backup_status(expected_status=exceptions.Forbidden) + + +class ProjectAdminTests33(RbacV3BackupsTests33): + credentials = ['project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_admin.backups_client_latest + + @decorators.idempotent_id('81c579bc-db98-4773-9590-b742d0b00b89') + def test_list_backups(self): + """List all backups""" + self._list_backups(expected_status=200) + + @decorators.idempotent_id('602dd42d-10df-4eb2-9664-3c9c44e3b35e') + def test_list_project_backups(self): + """List all backups for a project""" + self._list_project_backups(expected_status=200) + + @decorators.idempotent_id('2094dcee-9585-4745-b045-a0f8c79fbe52') + def test_show_backup(self): + """Show backup details""" + self._show_backup(expected_status=200) + + @decorators.idempotent_id('b77a8d69-1d12-480d-a83e-5f712d7c2b74') + def test_delete_backup(self): + """Delete a backup""" + self._delete_backup(expected_status=202) + + @decorators.idempotent_id('7221d2df-338c-4932-be40-ad7166c03db1') + def test_restore_backup(self): + """Restore a backup""" + self._restore_backup(expected_status=202) + + @decorators.idempotent_id('d347fa21-a5bf-4ce5-ab6b-246c3a06a735') + def test_create_backup(self): + """Create a backup""" + self._create_backup(expected_status=202) + + @decorators.idempotent_id('e179a062-47d5-4fa8-b359-dedab2afddd8') + def test_export_backup(self): + """Export a backup""" + self._export_backup(expected_status=200) + + @decorators.idempotent_id('1be80834-2463-49fb-a763-906e8c672fd5') + def test_import_backup(self): + """Import a backup""" + self._import_backup(expected_status=201) + + @decorators.idempotent_id('88db5943-0053-489a-af30-12b139a38a0b') + def test_reset_backup_status(self): + """Reset a backup status""" + self._reset_backup_status(expected_status=202) + + +class RbacV3BackupsTests39(RbacV3BackupsTests): + """Test API with microversion greater than 3.3""" + min_microversion = '3.9' + + def _update_backup(self, expected_status): + """Update a backup""" + new_description = "Updated backup description" + update_kwargs = {"description": new_description} + self.do_request( + method='update_backup', + expected_status=expected_status, + backup_id=self.backup_id, + **update_kwargs + ) + if expected_status != exceptions.Forbidden: + backup = self.admin_backups_client.show_backup( + backup_id=self.backup_id + )['backup'] + self.assertEqual( + backup['description'], new_description, + f"Backup {backup['name']} description should be " + f"{new_description}" + ) + + +class ProjectReaderTests39(RbacV3BackupsTests39): + credentials = ['project_reader', 'project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_reader.backups_client_latest + + @decorators.idempotent_id('50ccc892-6ed0-4015-b181-9f64ffa45f33') + @decorators.skip_because(bug='2017110') + def test_update_backup(self): + """Update a backup""" + self._update_backup(expected_status=exceptions.Forbidden) + + +class ProjectMemberTests39(RbacV3BackupsTests39): + credentials = ['project_member', 'project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_member.backups_client_latest + + @decorators.idempotent_id('a1cdd6f2-e9bc-4f6a-a0e6-2493ac6f9f27') + def test_update_backup(self): + """Update a backup""" + self._update_backup(expected_status=200) + + +class ProjectAdminTests39(RbacV3BackupsTests39): + credentials = ['project_admin'] + + @classmethod + def setup_clients(cls): + super().setup_clients() + cls.client = cls.os_project_admin.backups_client_latest + + @decorators.idempotent_id('2686eecf-e3cd-4f23-8771-aa040ed9be4b') + def test_update_backup(self): + """Update a backup""" + self._update_backup(expected_status=200)