Migrate to override_role for volume module (part 3)
Now that override_role has supplanted switch_role (which has
been deprecated) in [0], the RBAC tests need to switch to use
override_role.
This PS switches to override_role for the volume module. This
PS handles the remaining modules for volume.
This PS also removes unnecessary indexing into response bodies.
[0] I670fba358bf321eae0d22d18cea6d2f530f00716
Partially Implements: blueprint rbac-utils-contextmanager
Change-Id: I7451b7b098b9f751af777fcfad947ebc7a72f5ae
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_actions_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_actions_rbac.py
index e9ebb99..a755d48 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_actions_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_actions_rbac.py
@@ -28,13 +28,11 @@
class VolumesActionsV3RbacTest(rbac_base.BaseVolumeRbacTest):
- credentials = ['primary', 'admin']
@classmethod
def setup_clients(cls):
super(VolumesActionsV3RbacTest, cls).setup_clients()
- cls.admin_image_client = cls.os_admin.image_client_v2
- cls.admin_volumes_client = cls.os_admin.volumes_client_latest
+ cls.image_client = cls.os_primary.image_client_v2
@classmethod
def resource_setup(cls):
@@ -56,7 +54,7 @@
server['id'], volumeId=volume_id,
device='/dev/%s' % CONF.compute.volume_device_name)
waiters.wait_for_volume_resource_status(
- self.admin_volumes_client, volume_id, 'in-use')
+ self.volumes_client, volume_id, 'in-use')
self.addCleanup(self._detach_volume, volume_id)
def _detach_volume(self, volume_id=None):
@@ -65,7 +63,7 @@
self.volumes_client.detach_volume(volume_id)
waiters.wait_for_volume_resource_status(
- self.admin_volumes_client, volume_id, 'available')
+ self.volumes_client, volume_id, 'available')
@utils.services('compute')
@rbac_rule_validation.action(
@@ -74,8 +72,15 @@
@decorators.idempotent_id('f97b10e4-2eed-4f8b-8632-71c02cb9fe42')
def test_attach_volume_to_instance(self):
server = self._create_server()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._attach_volume(server)
+ volume_id = self.volume['id']
+
+ with self.rbac_utils.override_role(self):
+ self.servers_client.attach_volume(
+ server['id'], volumeId=volume_id,
+ device='/dev/%s' % CONF.compute.volume_device_name)
+ waiters.wait_for_volume_resource_status(
+ self.volumes_client, volume_id, 'in-use')
+ self.addCleanup(self._detach_volume, volume_id)
@utils.services('compute')
@decorators.attr(type='slow')
@@ -86,9 +91,12 @@
def test_detach_volume_from_instance(self):
server = self._create_server()
self._attach_volume(server)
+ volume_id = self.volume['id']
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._detach_volume()
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.detach_volume(volume_id)
+ waiters.wait_for_volume_resource_status(
+ self.volumes_client, volume_id, 'available')
@decorators.attr(type=["slow"])
@utils.services('image')
@@ -102,26 +110,26 @@
# Cinder's policy.json.
image_name = data_utils.rand_name(self.__class__.__name__ + '-Image')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.volumes_client.upload_volume(
- self.volume['id'], image_name=image_name, visibility="private",
- disk_format=CONF.volume.disk_format)['os-volume_upload_image']
+ with self.rbac_utils.override_role(self):
+ body = self.volumes_client.upload_volume(
+ self.volume['id'], image_name=image_name, visibility="private",
+ disk_format=CONF.volume.disk_format)['os-volume_upload_image']
image_id = body["image_id"]
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
- self.admin_image_client.delete_image,
+ self.image_client.delete_image,
image_id)
- waiters.wait_for_image_status(self.admin_image_client, image_id,
+ waiters.wait_for_image_status(self.image_client, image_id,
'active')
- waiters.wait_for_volume_resource_status(self.admin_volumes_client,
+ waiters.wait_for_volume_resource_status(self.volumes_client,
self.volume['id'], 'available')
@rbac_rule_validation.action(service="cinder",
rule="volume:update_readonly_flag")
@decorators.idempotent_id('2750717a-f250-4e41-9e09-02624aad6ff8')
def test_volume_readonly_update(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.update_volume_readonly(self.volume['id'],
- readonly=True)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.update_volume_readonly(self.volume['id'],
+ readonly=True)
self.addCleanup(self.volumes_client.update_volume_readonly,
self.volume['id'], readonly=False)
@@ -132,32 +140,32 @@
def test_unmanage_volume(self):
volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.unmanage_volume(volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.unmanage_volume(volume['id'])
@decorators.idempotent_id('59b783c0-f4ef-430c-8a90-1bad97d4ec5c')
@rbac_rule_validation.action(service="cinder",
rule="volume:update")
def test_volume_set_bootable(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.set_bootable_volume(self.volume['id'],
- bootable=True)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.set_bootable_volume(self.volume['id'],
+ bootable=True)
@decorators.idempotent_id('41566922-75a1-4484-99c7-9c8782ee99ac')
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_actions:reserve")
def test_volume_reserve(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.reserve_volume(self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.reserve_volume(self.volume['id'])
@decorators.idempotent_id('e5fa9564-77d9-4e57-b0c0-3e0ae4d08535')
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_actions:unreserve")
def test_volume_unreserve(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.unreserve_volume(self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.unreserve_volume(self.volume['id'])
@decorators.idempotent_id('c015c82f-7010-48cc-bd71-4ef542046f20')
@rbac_rule_validation.action(service="cinder",
@@ -166,10 +174,10 @@
vol_type = self.create_volume_type()['name']
volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.retype_volume(volume['id'], new_type=vol_type)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.retype_volume(volume['id'], new_type=vol_type)
waiters.wait_for_volume_retype(
- self.admin_volumes_client, volume['id'], vol_type)
+ self.volumes_client, volume['id'], vol_type)
@rbac_rule_validation.action(
service="cinder",
@@ -178,8 +186,9 @@
def test_volume_reset_status(self):
volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.reset_volume_status(volume['id'], status='error')
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.reset_volume_status(
+ volume['id'], status='error')
@rbac_rule_validation.action(
service="cinder",
@@ -189,8 +198,8 @@
volume = self.create_volume()
self.volumes_client.reset_volume_status(volume['id'], status='error')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.force_delete_volume(volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.force_delete_volume(volume['id'])
self.volumes_client.wait_for_resource_deletion(volume['id'])
@decorators.idempotent_id('48bd302b-950a-4830-840c-3158246ecdcc')
@@ -208,11 +217,11 @@
# Reset volume's status to error.
self.volumes_client.reset_volume_status(volume['id'], status='error')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.force_detach_volume(
- volume['id'], connector=None,
- attachment_id=attachment['attachment_id'])
- waiters.wait_for_volume_resource_status(self.admin_volumes_client,
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.force_detach_volume(
+ volume['id'], connector=None,
+ attachment_id=attachment['attachment_id'])
+ waiters.wait_for_volume_resource_status(self.volumes_client,
volume['id'], 'available')
@@ -221,13 +230,10 @@
min_microversion = '3.10'
max_microversion = 'latest'
- credentials = ['primary', 'admin']
-
@classmethod
def setup_clients(cls):
super(VolumesActionsV310RbacTest, cls).setup_clients()
- cls.admin_image_client = cls.os_admin.image_client_v2
- cls.admin_volumes_client = cls.os_admin.volumes_client_latest
+ cls.image_client = cls.os_primary.image_client_v2
@decorators.attr(type=["slow"])
@utils.services('image')
@@ -240,17 +246,17 @@
volume = self.create_volume()
image_name = data_utils.rand_name(self.__class__.__name__ + '-Image')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.volumes_client.upload_volume(
- volume['id'], image_name=image_name, visibility="public",
- disk_format=CONF.volume.disk_format)['os-volume_upload_image']
- image_id = body["image_id"]
+ with self.rbac_utils.override_role(self):
+ body = self.volumes_client.upload_volume(
+ volume['id'], image_name=image_name, visibility="public",
+ disk_format=CONF.volume.disk_format)['os-volume_upload_image']
+ image_id = body["image_id"]
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
- self.admin_image_client.delete_image,
+ self.image_client.delete_image,
image_id)
- waiters.wait_for_image_status(self.admin_image_client, image_id,
+ waiters.wait_for_image_status(self.image_client, image_id,
'active')
- waiters.wait_for_volume_resource_status(self.admin_volumes_client,
+ waiters.wait_for_volume_resource_status(self.volumes_client,
volume['id'], 'available')
@@ -262,5 +268,5 @@
@decorators.idempotent_id('a654833d-4811-4acd-93ef-5ac4a34c75bc')
@rbac_rule_validation.action(service="cinder", rule="volume:get_all")
def test_show_volume_summary(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.show_volume_summary()['volume-summary']
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.show_volume_summary()
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_basic_crud_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_basic_crud_rbac.py
index 244f333..1bd87d2 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_basic_crud_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_basic_crud_rbac.py
@@ -31,42 +31,42 @@
rule="volume:create")
@decorators.idempotent_id('426b08ef-6394-4d06-9128-965d5a6c38ef')
def test_create_volume(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_volume()
+ with self.rbac_utils.override_role(self):
+ self.create_volume()
@rbac_rule_validation.action(service="cinder",
rule="volume:delete")
@decorators.idempotent_id('6de9f9c2-509f-4558-867b-af21c7163be4')
def test_delete_volume(self):
volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.delete_volume(volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.delete_volume(volume['id'])
@rbac_rule_validation.action(service="cinder", rule="volume:get")
@decorators.idempotent_id('c4c3fdd5-b1b1-49c3-b977-a9f40ee9257a')
def test_get_volume(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.show_volume(self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.show_volume(self.volume['id'])
@rbac_rule_validation.action(service="cinder",
rule="volume:get_all")
@decorators.idempotent_id('e3ab7906-b04b-4c45-aa11-1104d302f940')
def test_volume_list(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.list_volumes()
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.list_volumes()
@rbac_rule_validation.action(service="cinder", rule="volume:update")
@decorators.idempotent_id('b751b889-9a9b-40d8-ae7d-4b0f65e71ac7')
def test_update_volume(self):
update_name = data_utils.rand_name(self.__class__.__name__ + 'volume')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.update_volume(self.volume['id'],
- name=update_name)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.update_volume(self.volume['id'],
+ name=update_name)
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_image_metadata")
@decorators.idempotent_id('3d48ca91-f02b-4616-a69d-4a8b296c8529')
def test_volume_list_image_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.list_volumes(detail=True)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.list_volumes(detail=True)
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_hosts_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_hosts_rbac.py
index 9519cea..c21c40e 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_hosts_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_hosts_rbac.py
@@ -25,8 +25,8 @@
rule="volume_extension:hosts")
@decorators.idempotent_id('64e837f5-5452-4e26-b934-c721ea7a8644')
def test_list_hosts(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_hosts_client.list_hosts()
+ with self.rbac_utils.override_role(self):
+ self.volume_hosts_client.list_hosts()
@decorators.idempotent_id('9ddf321e-788f-4787-b8cc-dfa59e264143')
@rbac_rule_validation.action(service="cinder",
@@ -37,5 +37,5 @@
self.assertNotEmpty(host_names, "No available volume host was found, "
"all hosts found were: %s" % hosts)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_hosts_client.show_host(host_names[0])
+ with self.rbac_utils.override_role(self):
+ self.volume_hosts_client.show_host(host_names[0])
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_metadata_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_metadata_rbac.py
index 5866934..768372f 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_metadata_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_metadata_rbac.py
@@ -49,50 +49,51 @@
rule="volume:create_volume_metadata")
@decorators.idempotent_id('232bbb8b-4c29-44dc-9077-b1398c20b738')
def test_create_volume_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._add_metadata(self.volume)
+ with self.rbac_utils.override_role(self):
+ self._add_metadata(self.volume)
@rbac_rule_validation.action(service="cinder",
rule="volume:get_volume_metadata")
@decorators.idempotent_id('87ea37d9-23ab-47b2-a59c-16fc4d2c6dfa')
def test_show_volume_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.show_volume_metadata(self.volume['id'])['metadata']
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.show_volume_metadata(
+ self.volume['id'])['metadata']
@rbac_rule_validation.action(service="cinder",
rule="volume:delete_volume_metadata")
@decorators.idempotent_id('7498dfc1-9db2-4423-ad20-e6dcb25d1beb')
def test_delete_volume_metadata_item(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.delete_volume_metadata_item(self.volume['id'],
- "key1")
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.delete_volume_metadata_item(self.volume['id'],
+ "key1")
@rbac_rule_validation.action(service="cinder",
rule="volume:update_volume_metadata")
@decorators.idempotent_id('8ce2ff80-99ba-49ae-9bb1-7e96729ee5af')
def test_update_volume_metadata_item(self):
updated_metadata_item = {"key1": "value1_update"}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.update_volume_metadata_item(
- self.volume['id'], "key1", updated_metadata_item)['meta']
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.update_volume_metadata_item(
+ self.volume['id'], "key1", updated_metadata_item)['meta']
@decorators.idempotent_id('a231b445-97a5-4657-b05f-245895e88da9')
@rbac_rule_validation.action(service="cinder",
rule="volume:update_volume_metadata")
def test_update_volume_metadata(self):
updated_metadata = {"key1": "value1"}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.update_volume_metadata(self.volume['id'],
- updated_metadata)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.update_volume_metadata(self.volume['id'],
+ updated_metadata)
@decorators.idempotent_id('a9d9e825-5ea3-42e6-96f3-7ac4e97b2ed0')
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_image_metadata")
def test_update_volume_image_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.update_volume_image_metadata(
- self.volume['id'], image_id=self.image_id)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.update_volume_image_metadata(
+ self.volume['id'], image_id=self.image_id)
self.addCleanup(self.volumes_client.delete_volume_image_metadata,
self.volume['id'], 'image_id')
@@ -107,6 +108,6 @@
self.volumes_client.delete_volume_image_metadata,
self.volume['id'], 'image_id')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_client.delete_volume_image_metadata(self.volume['id'],
- 'image_id')
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.delete_volume_image_metadata(self.volume['id'],
+ 'image_id')
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_quotas_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_quotas_rbac.py
index 6a79345..32cc48c 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_quotas_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_quotas_rbac.py
@@ -47,24 +47,24 @@
@rbac_rule_validation.action(service="cinder",
rule="volume_extension:quotas:show")
def test_list_quotas(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.quotas_client.show_quota_set(self.demo_tenant_id)
+ with self.rbac_utils.override_role(self):
+ self.quotas_client.show_quota_set(self.demo_tenant_id)
@decorators.idempotent_id('e47cf444-2753-4983-be6d-fc0d6523720f')
@rbac_rule_validation.action(service="cinder",
rule="volume_extension:quotas:show")
def test_list_quotas_usage_true(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.quotas_client.show_quota_set(self.demo_tenant_id,
- params={'usage': True})
+ with self.rbac_utils.override_role(self):
+ self.quotas_client.show_quota_set(self.demo_tenant_id,
+ params={'usage': True})
@rbac_rule_validation.action(service="cinder",
rule="volume_extension:quotas:show")
@decorators.idempotent_id('b3c7177e-b6b1-4d0f-810a-fc95606964dd')
def test_list_default_quotas(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.quotas_client.show_default_quota_set(
- self.demo_tenant_id)
+ with self.rbac_utils.override_role(self):
+ self.quotas_client.show_default_quota_set(
+ self.demo_tenant_id)
@rbac_rule_validation.action(service="cinder",
rule="volume_extension:quotas:update")
@@ -75,9 +75,9 @@
'volumes': 11,
'snapshots': 11}
# Update limits for all quota resources.
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.quotas_client.update_quota_set(
- self.demo_tenant_id, **new_quota_set)
+ with self.rbac_utils.override_role(self):
+ self.quotas_client.update_quota_set(
+ self.demo_tenant_id, **new_quota_set)
@decorators.idempotent_id('329bdb88-5132-4810-b1fc-350d181577e3')
@rbac_rule_validation.action(service="cinder",
@@ -85,5 +85,5 @@
def test_delete_quota_set(self):
self._restore_default_quota_set()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.quotas_client.delete_quota_set(self.demo_tenant_id)
+ with self.rbac_utils.override_role(self):
+ self.quotas_client.delete_quota_set(self.demo_tenant_id)
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_transfers_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_transfers_rbac.py
index a33ebe0..ad0d031 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_transfers_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_transfers_rbac.py
@@ -22,13 +22,11 @@
class VolumesTransfersV3RbacTest(rbac_base.BaseVolumeRbacTest):
- credentials = ['primary', 'admin']
@classmethod
def setup_clients(cls):
super(VolumesTransfersV3RbacTest, cls).setup_clients()
cls.transfers_client = cls.os_primary.volume_transfers_v2_client
- cls.admin_volumes_client = cls.os_admin.volumes_client_latest
@classmethod
def resource_setup(cls):
@@ -42,7 +40,7 @@
test_utils.call_and_ignore_notfound_exc(
self.transfers_client.delete_volume_transfer, transfer['id'])
waiters.wait_for_volume_resource_status(
- self.admin_volumes_client, self.volume['id'], 'available')
+ self.volumes_client, self.volume['id'], 'available')
def _create_transfer(self):
transfer = self.transfers_client.create_volume_transfer(
@@ -54,23 +52,23 @@
rule="volume:create_transfer")
@decorators.idempotent_id('25413af4-468d-48ff-94ca-4436f8526b3e')
def test_create_volume_transfer(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._create_transfer()
+ with self.rbac_utils.override_role(self):
+ self._create_transfer()
@rbac_rule_validation.action(service="cinder",
rule="volume:get_transfer")
@decorators.idempotent_id('7a0925d3-ed97-4c25-8299-e5cdabe2eb55')
def test_get_volume_transfer(self):
transfer = self._create_transfer()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.transfers_client.show_volume_transfer(transfer['id'])
+ with self.rbac_utils.override_role(self):
+ self.transfers_client.show_volume_transfer(transfer['id'])
@rbac_rule_validation.action(service="cinder",
rule="volume:get_all_transfers")
@decorators.idempotent_id('02a06f2b-5040-49e2-b2b7-619a7db59603')
def test_list_volume_transfers(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.transfers_client.list_volume_transfers()
+ with self.rbac_utils.override_role(self):
+ self.transfers_client.list_volume_transfers()
@decorators.idempotent_id('e84e45b0-9872-40bf-bf44-971266161a86')
@rbac_rule_validation.action(service="cinder",
@@ -84,14 +82,16 @@
@decorators.idempotent_id('987f2a11-d657-4984-a6c9-28f06c1cd014')
def test_accept_volume_transfer(self):
transfer = self._create_transfer()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.transfers_client.accept_volume_transfer(
- transfer['id'], auth_key=transfer['auth_key'])
+ with self.rbac_utils.override_role(self):
+ self.transfers_client.accept_volume_transfer(
+ transfer['id'], auth_key=transfer['auth_key'])
@rbac_rule_validation.action(service="cinder",
rule="volume:delete_transfer")
@decorators.idempotent_id('4672187e-7fff-454b-832a-5c8865dda868')
def test_delete_volume_transfer(self):
transfer = self._create_transfer()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.transfers_client.delete_volume_transfer(transfer['id'])
+ with self.rbac_utils.override_role(self):
+ self.transfers_client.delete_volume_transfer(transfer['id'])
+ waiters.wait_for_volume_resource_status(
+ self.volumes_client, self.volume['id'], 'available')
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_types_access_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_types_access_rbac.py
index f4aeee8..89dc0a2 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_types_access_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_types_access_rbac.py
@@ -56,17 +56,17 @@
def test_list_type_access(self):
self._add_type_access()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.list_type_access(self.vol_type['id'])[
- 'volume_type_access']
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.list_type_access(self.vol_type['id'])[
+ 'volume_type_access']
@decorators.idempotent_id('b462eeba-45d0-4d6e-945a-a1d27708d367')
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_type_access:addProjectAccess")
def test_add_type_access(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._add_type_access(ignore_not_found=True)
+ with self.rbac_utils.override_role(self):
+ self._add_type_access(ignore_not_found=True)
@decorators.idempotent_id('8f848aeb-636a-46f1-aeeb-e2a60e9d2bfe')
@rbac_rule_validation.action(
@@ -75,6 +75,6 @@
def test_remove_type_access(self):
self._add_type_access(ignore_not_found=True)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.remove_type_access(
- self.vol_type['id'], project=self.project_id)
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.remove_type_access(
+ self.vol_type['id'], project=self.project_id)
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volume_types_extra_specs_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volume_types_extra_specs_rbac.py
index 2abfd32..8d4c265 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volume_types_extra_specs_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volume_types_extra_specs_rbac.py
@@ -57,17 +57,17 @@
service="cinder",
rule="volume_extension:types_extra_specs:index")
def test_list_volume_types_extra_specs(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.list_volume_types_extra_specs(
- self.vol_type['id'])['extra_specs']
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.list_volume_types_extra_specs(
+ self.vol_type['id'])['extra_specs']
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:types_extra_specs:create")
@decorators.idempotent_id('eea40251-990b-49b0-99ae-10e4585b479b')
def test_create_volume_type_extra_specs(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._create_volume_type_extra_specs(ignore_not_found=True)
+ with self.rbac_utils.override_role(self):
+ self._create_volume_type_extra_specs(ignore_not_found=True)
@decorators.idempotent_id('e2dcc9c6-2fef-431d-afaf-92b45bc76d1a')
@rbac_rule_validation.action(
@@ -76,9 +76,9 @@
def test_show_volume_type_extra_specs(self):
self._create_volume_type_extra_specs()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.show_volume_type_extra_specs(
- self.vol_type['id'], self.spec_key)
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.show_volume_type_extra_specs(
+ self.vol_type['id'], self.spec_key)
@decorators.idempotent_id('93001912-f938-41c7-8787-62dc7010fd52')
@rbac_rule_validation.action(
@@ -87,9 +87,9 @@
def test_delete_volume_type_extra_specs(self):
self._create_volume_type_extra_specs(ignore_not_found=True)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.delete_volume_type_extra_specs(
- self.vol_type['id'], self.spec_key)
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.delete_volume_type_extra_specs(
+ self.vol_type['id'], self.spec_key)
@decorators.idempotent_id('0a444437-7402-4fbe-a18a-93af2ee00618')
@rbac_rule_validation.action(
@@ -99,6 +99,6 @@
self._create_volume_type_extra_specs()
update_extra_specs = {self.spec_key: "val2"}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volume_types_client.update_volume_type_extra_specs(
- self.vol_type['id'], self.spec_key, update_extra_specs)
+ with self.rbac_utils.override_role(self):
+ self.volume_types_client.update_volume_type_extra_specs(
+ self.vol_type['id'], self.spec_key, update_extra_specs)
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volumes_backup_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volumes_backup_rbac.py
index 51ee925..7f5f566 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volumes_backup_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volumes_backup_rbac.py
@@ -30,7 +30,6 @@
class VolumesBackupsV3RbacTest(rbac_base.BaseVolumeRbacTest):
- credentials = ['primary', 'admin']
@classmethod
def skip_checks(cls):
@@ -39,11 +38,6 @@
raise cls.skipException("Cinder backup feature disabled")
@classmethod
- def setup_clients(cls):
- super(VolumesBackupsV3RbacTest, cls).setup_clients()
- cls.admin_backups_client = cls.os_admin.backups_v2_client
-
- @classmethod
def resource_setup(cls):
super(VolumesBackupsV3RbacTest, cls).resource_setup()
cls.volume = cls.create_volume()
@@ -65,8 +59,8 @@
rule="backup:create")
@decorators.idempotent_id('6887ec94-0bcf-4ab7-b30f-3808a4b5a2a5')
def test_create_backup(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_backup(volume_id=self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.create_backup(volume_id=self.volume['id'])
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
@@ -75,22 +69,22 @@
def test_show_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.show_backup(backup['id'])
+ with self.rbac_utils.override_role(self):
+ self.backups_client.show_backup(backup['id'])
@rbac_rule_validation.action(service="cinder",
rule="backup:get_all")
@decorators.idempotent_id('4d18f0f0-7e01-4007-b622-dedc859b22f6')
def test_list_backups(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.list_backups()
+ with self.rbac_utils.override_role(self):
+ self.backups_client.list_backups()
@decorators.idempotent_id('dbd69865-876f-4835-b70e-7341153fb162')
@rbac_rule_validation.action(service="cinder",
rule="backup:get_all")
def test_list_backups_with_details(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.list_backups(detail=True)
+ with self.rbac_utils.override_role(self):
+ self.backups_client.list_backups(detail=True)
@decorators.attr(type='slow')
@decorators.idempotent_id('50f43bde-205e-438e-9a05-5eac07fc3d63')
@@ -100,10 +94,10 @@
def test_reset_backup_status(self):
backup = self.create_backup(volume_id=self.volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.reset_backup_status(backup_id=backup['id'],
- status='error')
- waiters.wait_for_volume_resource_status(self.admin_backups_client,
+ with self.rbac_utils.override_role(self):
+ self.backups_client.reset_backup_status(backup_id=backup['id'],
+ status='error')
+ waiters.wait_for_volume_resource_status(self.backups_client,
backup['id'], 'error')
@decorators.attr(type='slow')
@@ -113,10 +107,11 @@
def test_restore_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- restore = self.backups_client.restore_backup(backup['id'])['restore']
+ with self.rbac_utils.override_role(self):
+ restore = self.backups_client.restore_backup(
+ backup['id'])['restore']
waiters.wait_for_volume_resource_status(
- self.admin_backups_client, restore['backup_id'], 'available')
+ self.backups_client, restore['backup_id'], 'available')
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
@@ -130,13 +125,13 @@
volume_id=self.volume['id'])['backup']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.backups_client.delete_backup, backup['id'])
- waiters.wait_for_volume_resource_status(self.admin_backups_client,
+ waiters.wait_for_volume_resource_status(self.backups_client,
backup['id'], 'available')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.delete_backup(backup['id'])
+ with self.rbac_utils.override_role(self):
+ self.backups_client.delete_backup(backup['id'])
# Wait for deletion so error isn't thrown during clean up.
- self.admin_backups_client.wait_for_resource_deletion(backup['id'])
+ self.backups_client.wait_for_resource_deletion(backup['id'])
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
@@ -145,8 +140,8 @@
def test_export_backup(self):
backup = self.create_backup(volume_id=self.volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.export_backup(backup['id'])['backup-record']
+ with self.rbac_utils.override_role(self):
+ self.backups_client.export_backup(backup['id'])['backup-record']
@decorators.attr(type='slow')
@rbac_rule_validation.action(service="cinder",
@@ -160,10 +155,10 @@
new_url = self._modify_backup_url(
export_backup['backup_url'], {'id': new_id})
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- import_backup = self.backups_client.import_backup(
- backup_service=export_backup['backup_service'],
- backup_url=new_url)['backup']
+ with self.rbac_utils.override_role(self):
+ import_backup = self.backups_client.import_backup(
+ backup_service=export_backup['backup_service'],
+ backup_url=new_url)['backup']
self.addCleanup(self.backups_client.delete_backup, import_backup['id'])
@@ -188,8 +183,8 @@
backup = self.create_backup(volume_id=volume['id'])
expected_attr = 'os-backup-project-attr:project_id'
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.backups_client.show_backup(backup['id'])['backup']
+ with self.rbac_utils.override_role(self):
+ body = self.backups_client.show_backup(backup['id'])['backup']
# Show backup API attempts to inject the attribute below into the
# response body but only if policy enforcement succeeds.
@@ -221,6 +216,6 @@
'name': data_utils.rand_name(self.__class__.__name__ + '-Backup'),
'description': data_utils.rand_name("volume-backup-description")
}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.backups_client.update_backup(backup['id'],
- **update_kwargs)
+ with self.rbac_utils.override_role(self):
+ self.backups_client.update_backup(backup['id'],
+ **update_kwargs)
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volumes_manage_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volumes_manage_rbac.py
index 1365b79..852d81e 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volumes_manage_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volumes_manage_rbac.py
@@ -26,7 +26,6 @@
class VolumesManageV3RbacTest(rbac_base.BaseVolumeRbacTest):
- credentials = ['primary', 'admin']
@classmethod
def skip_checks(cls):
@@ -43,7 +42,6 @@
def setup_clients(cls):
super(VolumesManageV3RbacTest, cls).setup_clients()
cls.volume_manage_client = cls.os_primary.volume_manage_v2_client
- cls.admin_volumes_client = cls.os_admin.volumes_client_latest
def _manage_volume(self, org_volume):
# Manage volume
@@ -61,15 +59,11 @@
new_volume_id = self.volume_manage_client.manage_volume(
**new_volume_ref)['volume']['id']
- waiters.wait_for_volume_resource_status(self.admin_volumes_client,
+ waiters.wait_for_volume_resource_status(self.volumes_client,
new_volume_id, 'available')
self.addCleanup(self.delete_volume,
self.volumes_client, new_volume_id)
- def _unmanage_volume(self, volume):
- self.volumes_client.unmanage_volume(volume['id'])
- self.admin_volumes_client.wait_for_resource_deletion(volume['id'])
-
@rbac_rule_validation.action(
service="cinder",
rule="volume_extension:volume_manage")
@@ -80,19 +74,37 @@
# By default, the volume is managed after creation. We need to
# unmanage the volume first before testing manage volume.
- self._unmanage_volume(volume)
+ self.volumes_client.unmanage_volume(volume['id'])
+ self.volumes_client.wait_for_resource_deletion(volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- try:
- self._manage_volume(volume)
- except exceptions.Forbidden as e:
- # Since the test role under test does not have permission to
- # manage the volume, Forbidden exception is thrown and the
- # manageable list will not be cleaned up. Therefore, we need to
- # re-manage the volume at the end of the test case for proper
- # resource clean up.
- self.addCleanup(self._manage_volume, volume)
- raise exceptions.Forbidden(e)
+ new_volume_name = data_utils.rand_name(
+ self.__class__.__name__ + '-volume')
+
+ new_volume_ref = {
+ 'name': new_volume_name,
+ 'host': volume['os-vol-host-attr:host'],
+ 'ref': {CONF.volume.manage_volume_ref[0]:
+ CONF.volume.manage_volume_ref[1] % volume['id']},
+ 'volume_type': volume['volume_type'],
+ 'availability_zone': volume['availability_zone']}
+
+ with self.rbac_utils.override_role(self):
+ try:
+ new_volume_id = self.volume_manage_client.manage_volume(
+ **new_volume_ref)['volume']['id']
+ except exceptions.Forbidden as e:
+ # Since the test role under test does not have permission to
+ # manage the volume, Forbidden exception is thrown and the
+ # manageable list will not be cleaned up. Therefore, we need to
+ # re-manage the volume at the end of the test case for proper
+ # resource clean up.
+ self.addCleanup(self._manage_volume, volume)
+ raise exceptions.Forbidden(e)
+
+ waiters.wait_for_volume_resource_status(self.volumes_client,
+ new_volume_id, 'available')
+ self.addCleanup(
+ self.delete_volume, self.volumes_client, new_volume_id)
@rbac_rule_validation.action(
service="cinder",
@@ -102,8 +114,9 @@
volume_id = self.create_volume()['id']
volume = self.volumes_client.show_volume(volume_id)['volume']
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._unmanage_volume(volume)
+ with self.rbac_utils.override_role(self):
+ self.volumes_client.unmanage_volume(volume['id'])
+ self.volumes_client.wait_for_resource_deletion(volume['id'])
# In order to clean up the manageable list, we need to re-manage the
# volume after the test. The _manage_volume method will set up the
diff --git a/patrole_tempest_plugin/tests/api/volume/test_volumes_snapshots_rbac.py b/patrole_tempest_plugin/tests/api/volume/test_volumes_snapshots_rbac.py
index 7491820..df4fd10 100644
--- a/patrole_tempest_plugin/tests/api/volume/test_volumes_snapshots_rbac.py
+++ b/patrole_tempest_plugin/tests/api/volume/test_volumes_snapshots_rbac.py
@@ -24,7 +24,6 @@
class VolumesSnapshotV3RbacTest(rbac_base.BaseVolumeRbacTest):
- credentials = ['primary', 'admin']
@classmethod
def skip_checks(cls):
@@ -33,11 +32,6 @@
raise cls.skipException("Cinder volume snapshots are disabled")
@classmethod
- def setup_clients(cls):
- super(VolumesSnapshotV3RbacTest, cls).setup_clients()
- cls.admin_snapshots_client = cls.os_admin.snapshots_v2_client
-
- @classmethod
def resource_setup(cls):
super(VolumesSnapshotV3RbacTest, cls).resource_setup()
# Create a test shared volume for tests
@@ -60,17 +54,17 @@
@decorators.idempotent_id('ac7b2ee5-fbc0-4360-afc2-de8fa4881ede')
def test_snapshot_create(self):
# Create a temp snapshot
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_snapshot(self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.create_snapshot(self.volume['id'])
@rbac_rule_validation.action(service="cinder",
rule="volume:get_snapshot")
@decorators.idempotent_id('93a11b40-1ba8-44d6-a196-f8d97220f796')
def test_snapshot_get(self):
# Get the snapshot
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.snapshots_client.show_snapshot(self.snapshot
- ['id'])['snapshot']
+ with self.rbac_utils.override_role(self):
+ self.snapshots_client.show_snapshot(self.snapshot
+ ['id'])['snapshot']
@rbac_rule_validation.action(service="cinder",
rule="volume:update_snapshot")
@@ -79,11 +73,11 @@
new_desc = 'This is the new description of snapshot.'
params = {'description': new_desc}
# Updates snapshot with new values
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.snapshots_client.update_snapshot(
- self.snapshot['id'], **params)['snapshot']
+ with self.rbac_utils.override_role(self):
+ self.snapshots_client.update_snapshot(
+ self.snapshot['id'], **params)['snapshot']
waiters.wait_for_volume_resource_status(
- self.admin_snapshots_client, self.snapshot['id'], 'available')
+ self.snapshots_client, self.snapshot['id'], 'available')
@rbac_rule_validation.action(service="cinder",
rule="volume:get_all_snapshots")
@@ -92,8 +86,8 @@
"""list snapshots with params."""
# Verify list snapshots by display_name filter
params = {'name': self.snapshot['name']}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._list_by_param_values(params)
+ with self.rbac_utils.override_role(self):
+ self._list_by_param_values(params)
@rbac_rule_validation.action(service="cinder",
rule="volume:delete_snapshot")
@@ -101,8 +95,8 @@
def test_snapshot_delete(self):
# Create a temp snapshot
temp_snapshot = self.create_snapshot(self.volume['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- # Delete the snapshot
- self.snapshots_client.delete_snapshot(temp_snapshot['id'])
- self.admin_snapshots_client.wait_for_resource_deletion(
+ with self.rbac_utils.override_role(self):
+ # Delete the snapshot
+ self.snapshots_client.delete_snapshot(temp_snapshot['id'])
+ self.snapshots_client.wait_for_resource_deletion(
temp_snapshot['id'])