openstack/patrole
Code Issues Proposed changes

Merge "Increase test stability of backup RBAC tests"

Browse Source
This commit is contained in:
Zuul 2018-07-27 03:35:48 +00:00 committed by Gerrit Code Review
commit 05a73499ce
2 changed files with 66 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
patrole_tempest_plugin/tests/api/volume/rbac_base.py
View File

@ -12,6 +12,7 @@
# under the License.
from tempest.api.volume import base as vol_base
from tempest.common import waiters
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
@ -58,6 +59,29 @@ class BaseVolumeRbacTest(rbac_utils.RbacUtilsMixin,
cls.volume_types_client.delete_volume_type, volume_type['id'])
return volume_type
@classmethod
def _create_backup(cls, volume_id, backup_client=None, **kwargs):
"""Wrapper utility that returns a test backup.
Tempest has an instance-level version. This is a class-level version.
"""
if backup_client is None:
backup_client = cls.backups_client
if 'name' not in kwargs:
name = data_utils.rand_name(cls.__name__ + '-Backup')
kwargs['name'] = name
backup = backup_client.create_backup(
volume_id=volume_id, **kwargs)['backup']
cls.addClassResourceCleanup(
test_utils.call_and_ignore_notfound_exc,
backup_client.delete_backup, backup['id'])
waiters.wait_for_volume_resource_status(backup_client, backup['id'],
'available')
waiters.wait_for_volume_resource_status(cls.volumes_client, volume_id,
'available')
return backup
def create_group_type(self, name=None, ignore_notfound=False, **kwargs):
"""Create a test group-type"""
name = name or data_utils.rand_name(

64
patrole_tempest_plugin/tests/api/volume/test_volumes_backup_rbac.py
View File

@ -41,6 +41,7 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
def resource_setup(cls):
super(VolumesBackupsV3RbacTest, cls).resource_setup()
cls.volume = cls.create_volume()
cls.backup = cls._create_backup(volume_id=cls.volume['id'])
def _decode_url(self, backup_url):
return json.loads(base64.decode_as_text(backup_url))
@ -59,18 +60,24 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
rule="backup:create")
@decorators.idempotent_id('6887ec94-0bcf-4ab7-b30f-3808a4b5a2a5')
def test_create_backup(self):
backup_name = data_utils.rand_name(self.__class__.__name__ + '-Backup')
with self.rbac_utils.override_role(self):
self.create_backup(volume_id=self.volume['id'])
backup = self.backups_client.create_backup(
volume_id=self.volume['id'], name=backup_name)['backup']
self.addCleanup(self.backups_client.delete_backup, backup['id'])
waiters.wait_for_volume_resource_status(
self.backups_client, backup['id'], 'available')
waiters.wait_for_volume_resource_status(self.volumes_client,
self.volume['id'], 'available')
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
rule="backup:get")
@decorators.idempotent_id('abd92bdd-b0fb-4dc4-9cfc-de9e968f8c8a')
def test_show_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
with self.rbac_utils.override_role(self):
self.backups_client.show_backup(backup['id'])
self.backups_client.show_backup(self.backup['id'])
@rbac_rule_validation.action(service="cinder",
rule="backup:get_all")
@ -92,7 +99,10 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
service="cinder",
rule="volume_extension:backup_admin_actions:reset_status")
def test_reset_backup_status(self):
# Use instance-level create_backup for easier debugging.
backup = self.create_backup(volume_id=self.volume['id'])
waiters.wait_for_volume_resource_status(self.volumes_client,
self.volume['id'], 'available')
with self.rbac_utils.override_role(self):
self.backups_client.reset_backup_status(backup_id=backup['id'],
@ -105,11 +115,11 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
rule="backup:restore")
@decorators.idempotent_id('9c794bf9-2446-4f41-8fe0-80b71e757f9d')
def test_restore_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
with self.rbac_utils.override_role(self):
restore = self.backups_client.restore_backup(
backup['id'])['restore']
self.backup['id'])['restore']
self.addCleanup(self.volumes_client.delete_volume,
restore['volume_id'])
waiters.wait_for_volume_resource_status(
self.backups_client, restore['backup_id'], 'available')
@ -127,6 +137,8 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
self.backups_client.delete_backup, backup['id'])
waiters.wait_for_volume_resource_status(self.backups_client,
backup['id'], 'available')
waiters.wait_for_volume_resource_status(self.volumes_client,
self.volume['id'], 'available')
with self.rbac_utils.override_role(self):
self.backups_client.delete_backup(backup['id'])
@ -138,19 +150,17 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
rule="backup:export-import")
@decorators.idempotent_id('e984ec8d-e8eb-485c-98bc-f1856020303c')
def test_export_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
with self.rbac_utils.override_role(self):
self.backups_client.export_backup(backup['id'])['backup-record']
self.backups_client.export_backup(self.backup['id'])[
'backup-record']
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
rule="backup:backup-import")
@decorators.idempotent_id('1e70f039-4556-44cc-9cc1-edf2b7ed648b')
def test_import_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
export_backup = self.backups_client.export_backup(
backup['id'])['backup-record']
self.backup['id'])['backup-record']
new_id = data_utils.rand_uuid()
new_url = self._modify_backup_url(
export_backup['backup_url'], {'id': new_id})
@ -160,6 +170,9 @@ class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
backup_service=export_backup['backup_service'],
backup_url=new_url)['backup']
self.addCleanup(self.backups_client.delete_backup, import_backup['id'])
waiters.wait_for_volume_resource_status(self.backups_client,
import_backup['id'],
'available')
class VolumesBackupsV318RbacTest(rbac_base.BaseVolumeRbacTest):
@ -175,22 +188,25 @@ class VolumesBackupsV318RbacTest(rbac_base.BaseVolumeRbacTest):
if not CONF.volume_feature_enabled.backup:
raise cls.skipException("Cinder backup feature disabled")
@classmethod
def resource_setup(cls):
super(VolumesBackupsV318RbacTest, cls).resource_setup()
cls.volume = cls.create_volume()
cls.backup = cls._create_backup(volume_id=cls.volume['id'])
cls.expected_attr = 'os-backup-project-attr:project_id'
@decorators.idempotent_id('69801485-d5be-4e75-bbb4-168d50b5a8c2')
@rbac_rule_validation.action(service="cinder",
rule="backup:backup_project_attribute")
def test_show_backup_project_attribute(self):
volume = self.create_volume()
backup = self.create_backup(volume_id=volume['id'])
expected_attr = 'os-backup-project-attr:project_id'
with self.rbac_utils.override_role(self):
body = self.backups_client.show_backup(backup['id'])['backup']
body = self.backups_client.show_backup(self.backup['id'])['backup']
# Show backup API attempts to inject the attribute below into the
# response body but only if policy enforcement succeeds.
if expected_attr not in body:
if self.expected_attr not in body:
raise rbac_exceptions.RbacMalformedResponse(
attribute=expected_attr)
attribute=self.expected_attr)
class VolumesBackupsV39RbacTest(rbac_base.BaseVolumeRbacTest):
@ -204,18 +220,22 @@ class VolumesBackupsV39RbacTest(rbac_base.BaseVolumeRbacTest):
if not CONF.volume_feature_enabled.backup:
raise cls.skipException("Cinder backup feature disabled")
@classmethod
def resource_setup(cls):
super(VolumesBackupsV39RbacTest, cls).resource_setup()
cls.volume = cls.create_volume()
cls.backup = cls._create_backup(volume_id=cls.volume['id'])
@decorators.attr(type='slow')
@decorators.idempotent_id('b45b0e98-6eb8-4c62-aa53-0f8c7c09faa6')
@rbac_rule_validation.action(
service="cinder",
rule="backup:update")
def test_backup_update(self):
volume = self.create_volume()
backup = self.create_backup(volume_id=volume['id'])
update_kwargs = {
'name': data_utils.rand_name(self.__class__.__name__ + '-Backup'),
'description': data_utils.rand_name("volume-backup-description")
}
with self.rbac_utils.override_role(self):
self.backups_client.update_backup(backup['id'],
self.backups_client.update_backup(self.backup['id'],
**update_kwargs)