Test srbac on backups

Testing reader, member and admin roles.
This file contains test 2 classes for API version 3.3 and 3.9

Signed-off-by: Yosi Ben Shimon <ybenshim@redhat.com>
Change-Id: I9e6ea838fa7a93fcfc3d4ef7f807224aadee187b

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

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)