Migrate to override_role for compute 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 compute module. This
PS handles the last 11 modules for compute.
This PS also removes unnecessary indexing into response bodies.
[0] I670fba358bf321eae0d22d18cea6d2f530f00716
Partially Implements: blueprint rbac-utils-contextmanager
Change-Id: I0f8d249d58d7c6ad6f1d2fde2b52aefda10c64c4
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_consoles_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_consoles_rbac.py
index 7744263..fa2f359 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_consoles_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_consoles_rbac.py
@@ -40,8 +40,8 @@
rule="os_compute_api:os-console-output")
@decorators.idempotent_id('90fd80f6-456c-11e7-a919-92ebcb67fe33')
def test_get_console_output(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.get_console_output(self.server_id)
+ with self.rbac_utils.override_role(self):
+ self.servers_client.get_console_output(self.server_id)
class ServerConsolesMaxV25RbacTest(rbac_base.BaseV2ComputeRbacTest):
@@ -64,8 +64,8 @@
rule="os_compute_api:os-remote-consoles")
@decorators.idempotent_id('b0a72c02-9b15-4dcb-b186-efe8753370ab')
def test_get_vnc_console_output(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.get_vnc_console(self.server_id, type="novnc")
+ with self.rbac_utils.override_role(self):
+ self.servers_client.get_vnc_console(self.server_id, type="novnc")
class ServerConsolesV26RbacTest(rbac_base.BaseV2ComputeRbacTest):
@@ -89,6 +89,6 @@
rule="os_compute_api:os-remote-consoles")
@decorators.idempotent_id('879597de-87e0-4da9-a60a-28c8088dc508')
def test_get_remote_console_output(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.get_remote_console(self.server_id,
- "novnc", "vnc")
+ with self.rbac_utils.override_role(self):
+ self.servers_client.get_remote_console(self.server_id,
+ "novnc", "vnc")
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_groups_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_groups_rbac.py
index be24569..1674b1a 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_groups_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_groups_rbac.py
@@ -34,8 +34,8 @@
rule="os_compute_api:os-server-groups:create")
@decorators.idempotent_id('7f3eae94-6130-47e9-81ac-34009f55be2f')
def test_create_server_group(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_test_server_group()
+ with self.rbac_utils.override_role(self):
+ self.create_test_server_group()
@rbac_rule_validation.action(
service="nova",
@@ -43,16 +43,16 @@
@decorators.idempotent_id('832d9be3-632e-47b2-93d2-5897db43e3e2')
def test_delete_server_group(self):
server_group = self.create_test_server_group()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.server_groups_client.delete_server_group(server_group['id'])
+ with self.rbac_utils.override_role(self):
+ self.server_groups_client.delete_server_group(server_group['id'])
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-server-groups:index")
@decorators.idempotent_id('5eccd67f-5945-483b-b1c8-de851ebfc1c1')
def test_list_server_groups(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.server_groups_client.list_server_groups()
+ with self.rbac_utils.override_role(self):
+ self.server_groups_client.list_server_groups()
@rbac_rule_validation.action(
service="nova",
@@ -60,5 +60,5 @@
@decorators.idempotent_id('62534e3f-7e99-4a3d-a08e-33e056460cf2')
def test_show_server_group(self):
server_group = self.create_test_server_group()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.server_groups_client.show_server_group(server_group['id'])
+ with self.rbac_utils.override_role(self):
+ self.server_groups_client.show_server_group(server_group['id'])
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_metadata_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_metadata_rbac.py
index ac2dcb0..05b1758 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_metadata_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_metadata_rbac.py
@@ -37,51 +37,48 @@
service="nova",
rule="os_compute_api:server-metadata:index")
def test_list_server_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_server_metadata(self.server['id'])['metadata']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_server_metadata(self.server['id'])
@decorators.idempotent_id('6e76748b-2417-4fa2-b41a-c0cc4bff356b')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:server-metadata:update_all")
def test_set_server_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.set_server_metadata(self.server['id'], {})[
- 'metadata']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.set_server_metadata(self.server['id'], {})
@decorators.idempotent_id('1060bac4-fe16-4a77-be64-d8e482a06eab')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:server-metadata:create")
def test_update_server_metadata(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.update_server_metadata(self.server['id'], {})[
- 'metadata']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.update_server_metadata(self.server['id'], {})
@decorators.idempotent_id('93dd8323-d3fa-48d1-8bd6-91c1b62fc341')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:server-metadata:show")
def test_show_server_metadata_item(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.show_server_metadata_item(
- self.server['id'], 'default_key')['meta']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.show_server_metadata_item(
+ self.server['id'], 'default_key')
@decorators.idempotent_id('79511293-4bd7-447d-ba7e-634d0f4da70c')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:server-metadata:update")
def test_set_server_metadata_item(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.set_server_metadata_item(
- self.server['id'], 'default_key', {'default_key': 'value2'})[
- 'meta']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.set_server_metadata_item(
+ self.server['id'], 'default_key', {'default_key': 'value2'})
@decorators.idempotent_id('feec5064-678d-40bc-a88f-c856e18d1e31')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:server-metadata:delete")
def test_delete_server_metadata_item(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.delete_server_metadata_item(
- self.server['id'], 'delete_key')
+ with self.rbac_utils.override_role(self):
+ self.servers_client.delete_server_metadata_item(
+ self.server['id'], 'delete_key')
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_migrations_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_migrations_rbac.py
index 1913159..a867b81 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_migrations_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_migrations_rbac.py
@@ -30,8 +30,6 @@
max_microversion = 'latest'
block_migration = 'auto'
- credentials = ['primary', 'admin']
-
@classmethod
def skip_checks(cls):
super(MigrateServerV225RbacTest, cls).skip_checks()
@@ -39,11 +37,6 @@
raise cls.skipException(
"Less than 2 compute nodes, skipping migration tests.")
- @classmethod
- def setup_clients(cls):
- super(MigrateServerV225RbacTest, cls).setup_clients()
- cls.admin_servers_client = cls.os_admin.servers_client
-
def _get_server_details(self, server_id):
body = self.servers_client.show_server(server_id)['server']
return body
@@ -73,9 +66,9 @@
@decorators.idempotent_id('c6f1607c-9fed-4c00-807e-9ba675b98b1b')
def test_cold_migration(self):
server = self.create_test_server(wait_until="ACTIVE")
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.migrate_server(server['id'])
- waiters.wait_for_server_status(self.admin_servers_client,
+ with self.rbac_utils.override_role(self):
+ self.servers_client.migrate_server(server['id'])
+ waiters.wait_for_server_status(self.servers_client,
server['id'], 'VERIFY_RESIZE')
@decorators.attr(type='slow')
@@ -91,8 +84,9 @@
actual_host = self._get_host_for_server(server_id)
target_host = self._get_host_other_than(actual_host)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.live_migrate_server(
- server_id, host=target_host, block_migration=self.block_migration)
- waiters.wait_for_server_status(self.admin_servers_client,
+ with self.rbac_utils.override_role(self):
+ self.servers_client.live_migrate_server(
+ server_id, host=target_host,
+ block_migration=self.block_migration)
+ waiters.wait_for_server_status(self.servers_client,
server_id, "ACTIVE")
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_misc_policy_actions_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_misc_policy_actions_rbac.py
index 48df4a3..fabcb4c 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_misc_policy_actions_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_misc_policy_actions_rbac.py
@@ -83,8 +83,8 @@
@decorators.idempotent_id('ae84dd0b-f364-462e-b565-3457f9c019ef')
def test_reset_server_state(self):
"""Test reset server state, part of os-admin-actions."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.reset_state(self.server['id'], state='error')
+ with self.rbac_utils.override_role(self):
+ self.servers_client.reset_state(self.server['id'], state='error')
self.addCleanup(self.servers_client.reset_state, self.server['id'],
state='active')
@@ -95,8 +95,8 @@
@decorators.idempotent_id('ce48c340-51c1-4cff-9b6e-0cc5ef008630')
def test_inject_network_info(self):
"""Test inject network info, part of os-admin-actions."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.inject_network_info(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.inject_network_info(self.server['id'])
@utils.requires_ext(extension='os-admin-actions', service='compute')
@rbac_rule_validation.action(
@@ -105,8 +105,8 @@
@decorators.idempotent_id('2911a242-15c4-4fcb-80d5-80a8930661b0')
def test_reset_network(self):
"""Test reset network, part of os-admin-actions."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.reset_network(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.reset_network(self.server['id'])
@testtools.skipUnless(CONF.compute_feature_enabled.change_password,
'Change password not available.')
@@ -119,13 +119,13 @@
original_password = self.servers_client.show_password(
self.server['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.change_password(
- self.server['id'], adminPass=data_utils.rand_password())
+ with self.rbac_utils.override_role(self):
+ self.servers_client.change_password(
+ self.server['id'], adminPass=data_utils.rand_password())
self.addCleanup(self.servers_client.change_password, self.server['id'],
adminPass=original_password)
waiters.wait_for_server_status(
- self.os_admin.servers_client, self.server['id'], 'ACTIVE')
+ self.servers_client, self.server['id'], 'ACTIVE')
@utils.requires_ext(extension='os-config-drive', service='compute')
@decorators.idempotent_id('2c82e819-382d-4d6f-87f0-a45954cbbc64')
@@ -134,8 +134,8 @@
rule="os_compute_api:os-config-drive")
def test_list_servers_with_details_config_drive(self):
"""Test list servers with config_drive property in response body."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.list_servers(detail=True)['servers']
expected_attr = 'config_drive'
# If the first server contains "config_drive", then all the others do.
if expected_attr not in body[0]:
@@ -149,9 +149,8 @@
rule="os_compute_api:os-config-drive")
def test_show_server_config_drive(self):
"""Test show server with config_drive property in response body."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
-
- body = self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.show_server(self.server['id'])['server']
expected_attr = 'config_drive'
if expected_attr not in body:
raise rbac_exceptions.RbacMalformedResponse(
@@ -164,9 +163,9 @@
rule="os_compute_api:os-deferred-delete")
def test_force_delete_server(self):
"""Test force delete server, part of os-deferred-delete."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- # Force-deleting a server enforces os-deferred-delete.
- self.servers_client.force_delete_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ # Force-deleting a server enforces os-deferred-delete.
+ self.servers_client.force_delete_server(self.server['id'])
@decorators.idempotent_id('d873740a-7b10-40a9-943d-7cc18115370e')
@utils.requires_ext(extension='OS-EXT-AZ', service='compute')
@@ -177,8 +176,8 @@
"""Test list servers OS-EXT-AZ:availability_zone attr in resp body."""
expected_attr = 'OS-EXT-AZ:availability_zone'
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.list_servers(detail=True)['servers']
# If the first server contains `expected_attr`, then all the others do.
if expected_attr not in body[0]:
raise rbac_exceptions.RbacMalformedResponse(
@@ -193,8 +192,8 @@
"""Test show server OS-EXT-AZ:availability_zone attr in resp body."""
expected_attr = 'OS-EXT-AZ:availability_zone'
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.show_server(self.server['id'])['server']
if expected_attr not in body:
raise rbac_exceptions.RbacMalformedResponse(
attribute=expected_attr)
@@ -208,8 +207,8 @@
"""Test list servers with details, with extended server attributes in
response body.
"""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.list_servers(detail=True)['servers']
# NOTE(felipemonteiro): The attributes included below should be
# returned by all microversions. We don't include tests for other
@@ -230,8 +229,8 @@
"""Test show server with extended server attributes in response
body.
"""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.show_server(self.server['id'])['server']
# NOTE(felipemonteiro): The attributes included below should be
# returned by all microversions. We don't include tests for other
@@ -250,8 +249,8 @@
rule="os_compute_api:os-extended-status")
def test_list_servers_extended_status(self):
"""Test list servers with extended properties in response body."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.list_servers(detail=True)['servers']
expected_attrs = ('OS-EXT-STS:task_state', 'OS-EXT-STS:vm_state',
'OS-EXT-STS:power_state')
@@ -267,8 +266,8 @@
rule="os_compute_api:os-extended-status")
def test_show_server_extended_status(self):
"""Test show server with extended properties in response body."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.show_server(self.server['id'])['server']
expected_attrs = ('OS-EXT-STS:task_state', 'OS-EXT-STS:vm_state',
'OS-EXT-STS:power_state')
@@ -288,8 +287,8 @@
"""
expected_attr = 'os-extended-volumes:volumes_attached'
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.list_servers(detail=True)['servers']
if expected_attr not in body[0]:
raise rbac_exceptions.RbacMalformedResponse(
attribute=expected_attr)
@@ -305,8 +304,8 @@
"""
expected_attr = 'os-extended-volumes:volumes_attached'
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- body = self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ body = self.servers_client.show_server(self.server['id'])['server']
if expected_attr not in body:
raise rbac_exceptions.RbacMalformedResponse(
attribute=expected_attr)
@@ -318,8 +317,8 @@
rule="os_compute_api:os-instance-actions")
def test_list_instance_actions(self):
"""Test list instance actions, part of os-instance-actions."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_instance_actions(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_instance_actions(self.server['id'])
@utils.requires_ext(extension='os-instance-actions', service='compute')
@decorators.idempotent_id('eb04c439-4215-4029-9ccb-5b3c041bfc25')
@@ -334,9 +333,9 @@
# NOTE: "os_compute_api:os-instance-actions" is also enforced.
request_id = self.server.response['x-compute-request-id']
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- instance_action = self.servers_client.show_instance_action(
- self.server['id'], request_id)['instanceAction']
+ with self.rbac_utils.override_role(self):
+ instance_action = self.servers_client.show_instance_action(
+ self.server['id'], request_id)['instanceAction']
if 'events' not in instance_action:
raise rbac_exceptions.RbacMalformedResponse(
@@ -352,9 +351,9 @@
rule="os_compute_api:os-keypairs")
@decorators.idempotent_id('81e6fa34-c06b-42ca-b195-82bf8699b940')
def test_show_server_keypair(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- result =\
- self.servers_client.show_server(self.server['id'])['server']
+ with self.rbac_utils.override_role(self):
+ result = self.servers_client.show_server(self.server['id'])[
+ 'server']
if 'key_name' not in result:
raise rbac_exceptions.RbacMalformedResponse(
attribute='key_name')
@@ -364,8 +363,8 @@
rule="os_compute_api:os-keypairs")
@decorators.idempotent_id('41ca4280-ec59-4b80-a9b1-6bc6366faf39')
def test_list_servers_keypairs(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- result = self.servers_client.list_servers(detail=True)['servers']
+ with self.rbac_utils.override_role(self):
+ result = self.servers_client.list_servers(detail=True)['servers']
if 'key_name' not in result[0]:
raise rbac_exceptions.RbacMalformedResponse(
attribute='key_name')
@@ -376,8 +375,8 @@
@decorators.idempotent_id('b81e10fb-1864-498f-8c1d-5175c6fec5fb')
def test_lock_server(self):
"""Test lock server, part of os-lock-server."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.lock_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.lock_server(self.server['id'])
self.addCleanup(self.servers_client.unlock_server, self.server['id'])
@rbac_rule_validation.action(
@@ -389,8 +388,8 @@
self.servers_client.lock_server(self.server['id'])
self.addCleanup(self.servers_client.unlock_server, self.server['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.unlock_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.unlock_server(self.server['id'])
@rbac_rule_validation.action(
service="nova",
@@ -406,8 +405,8 @@
self.os_admin.servers_client.lock_server(self.server['id'])
self.addCleanup(self.servers_client.unlock_server, self.server['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.unlock_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.unlock_server(self.server['id'])
@utils.requires_ext(extension='os-rescue', service='compute')
@rbac_rule_validation.action(
@@ -416,8 +415,8 @@
@decorators.idempotent_id('fbbb2afc-ed0e-4552-887d-ac00fb5d436e')
def test_rescue_server(self):
"""Test rescue server, part of os-rescue."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.rescue_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.rescue_server(self.server['id'])
@decorators.idempotent_id('ac2d956f-d6a3-4184-b814-b44d05c9574c')
@utils.requires_ext(extension='os-rescue', service='compute')
@@ -428,10 +427,10 @@
"""Test unrescue server, part of os-rescue."""
self.servers_client.rescue_server(self.server['id'])
waiters.wait_for_server_status(
- self.os_admin.servers_client, self.server['id'], 'RESCUE')
+ self.servers_client, self.server['id'], 'RESCUE')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.unrescue_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.unrescue_server(self.server['id'])
# `setUp` will wait for the server to reach 'ACTIVE' for next test.
@utils.requires_ext(extension='os-server-diagnostics', service='compute')
@@ -441,8 +440,8 @@
@decorators.idempotent_id('5dabfcc4-bedb-417b-8247-b3ee7c5c0f3e')
def test_show_server_diagnostics(self):
"""Test show server diagnostics, part of os-server-diagnostics."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.show_server_diagnostics(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.show_server_diagnostics(self.server['id'])
@utils.requires_ext(extension='os-server-password', service='compute')
@decorators.idempotent_id('aaf43f78-c178-4581-ac18-14afd3f1f6ba')
@@ -451,8 +450,8 @@
rule="os_compute_api:os-server-password")
def test_delete_server_password(self):
"""Test delete server password, part of os-server-password."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.delete_password(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.delete_password(self.server['id'])
@utils.requires_ext(extension='os-server-password', service='compute')
@rbac_rule_validation.action(
@@ -461,8 +460,8 @@
@decorators.idempotent_id('f677971a-7d20-493c-977f-6ff0a74b5b2c')
def test_get_server_password(self):
"""Test show server password, part of os-server-password."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.show_password(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.show_password(self.server['id'])
@utils.requires_ext(extension='OS-SRV-USG', service='compute')
@rbac_rule_validation.action(
@@ -475,8 +474,8 @@
TODO(felipemonteiro): Once multiple policy testing is supported, this
test can be combined with the generic test for showing a server.
"""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.show_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.show_server(self.server['id'])
@utils.requires_ext(extension='os-simple-tenant-usage', service='compute')
@rbac_rule_validation.action(
@@ -485,8 +484,8 @@
@decorators.idempotent_id('2aef094f-0452-4df6-a66a-0ec22a92b16e')
def test_list_simple_tenant_usages(self):
"""Test list tenant usages, part of os-simple-tenant-usage."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.tenant_usages_client.list_tenant_usages()
+ with self.rbac_utils.override_role(self):
+ self.tenant_usages_client.list_tenant_usages()
@utils.requires_ext(extension='os-simple-tenant-usage', service='compute')
@rbac_rule_validation.action(
@@ -497,8 +496,8 @@
"""Test show tenant usage, part of os-simple-tenant-usage."""
tenant_id = self.os_primary.credentials.tenant_id
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.tenant_usages_client.show_tenant_usage(tenant_id=tenant_id)
+ with self.rbac_utils.override_role(self):
+ self.tenant_usages_client.show_tenant_usage(tenant_id=tenant_id)
@testtools.skipUnless(CONF.compute_feature_enabled.suspend,
"Suspend compute feature is not available.")
@@ -508,11 +507,11 @@
rule="os_compute_api:os-suspend-server:suspend")
def test_suspend_server(self):
"""Test suspend server, part of os-suspend-server."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.suspend_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.suspend_server(self.server['id'])
self.addCleanup(self.servers_client.resume_server, self.server['id'])
waiters.wait_for_server_status(
- self.os_admin.servers_client, self.server['id'], 'SUSPENDED')
+ self.servers_client, self.server['id'], 'SUSPENDED')
@testtools.skipUnless(CONF.compute_feature_enabled.suspend,
"Suspend compute feature is not available.")
@@ -524,12 +523,12 @@
"""Test resume server, part of os-suspend-server."""
self.servers_client.suspend_server(self.server['id'])
waiters.wait_for_server_status(
- self.os_admin.servers_client, self.server['id'], 'SUSPENDED')
+ self.servers_client, self.server['id'], 'SUSPENDED')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.resume_server(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.resume_server(self.server['id'])
waiters.wait_for_server_status(
- self.os_admin.servers_client, self.server['id'], 'ACTIVE')
+ self.servers_client, self.server['id'], 'ACTIVE')
class MiscPolicyActionsNetworkRbacTest(rbac_base.BaseV2ComputeRbacTest):
@@ -541,8 +540,6 @@
* tests that require network resources
"""
- credentials = ['primary', 'admin']
-
@classmethod
def skip_checks(cls):
super(MiscPolicyActionsNetworkRbacTest, cls).skip_checks()
@@ -565,7 +562,7 @@
interface = self.interfaces_client.create_interface(
self.server['id'])['interfaceAttachment']
waiters.wait_for_interface_status(
- self.os_admin.interfaces_client, self.server['id'],
+ self.interfaces_client, self.server['id'],
interface['port_id'], 'ACTIVE')
self.addCleanup(
test_utils.call_and_ignore_notfound_exc,
@@ -582,9 +579,8 @@
rule="os_compute_api:os-attach-interfaces")
def test_list_interfaces(self):
"""Test list interfaces, part of os-attach-interfaces."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.interfaces_client.list_interfaces(
- self.server['id'])['interfaceAttachments']
+ with self.rbac_utils.override_role(self):
+ self.interfaces_client.list_interfaces(self.server['id'])
@decorators.idempotent_id('1b9cf7db-dc50-48a2-8eb9-8c25af5e934a')
@testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,
@@ -596,9 +592,9 @@
def test_show_interface(self):
"""Test show interfaces, part of os-attach-interfaces."""
interface = self._attach_interface_to_server()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.interfaces_client.show_interface(
- self.server['id'], interface['port_id'])['interfaceAttachment']
+ with self.rbac_utils.override_role(self):
+ self.interfaces_client.show_interface(
+ self.server['id'], interface['port_id'])
@testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,
"Interface attachment is not available.")
@@ -609,8 +605,16 @@
rule="os_compute_api:os-attach-interfaces:create")
def test_create_interface(self):
"""Test create interface, part of os-attach-interfaces."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._attach_interface_to_server()
+ with self.rbac_utils.override_role(self):
+ interface = self.interfaces_client.create_interface(
+ self.server['id'])['interfaceAttachment']
+ waiters.wait_for_interface_status(
+ self.interfaces_client, self.server['id'],
+ interface['port_id'], 'ACTIVE')
+ self.addCleanup(
+ test_utils.call_and_ignore_notfound_exc,
+ self.interfaces_client.delete_interface,
+ self.server['id'], interface['port_id'])
@testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,
"Interface attachment is not available.")
@@ -623,9 +627,9 @@
"""Test delete interface, part of os-attach-interfaces."""
interface = self._attach_interface_to_server()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.interfaces_client.delete_interface(self.server['id'],
- interface['port_id'])
+ with self.rbac_utils.override_role(self):
+ self.interfaces_client.delete_interface(self.server['id'],
+ interface['port_id'])
@decorators.idempotent_id('6886d360-0d86-4760-b1a3-882d81fbebcc')
@utils.requires_ext(extension='os-ips', service='compute')
@@ -634,8 +638,8 @@
rule="os_compute_api:ips:index")
def test_list_addresses(self):
"""Test list server addresses, part of ips policy family."""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_addresses(self.server['id'])['addresses']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_addresses(self.server['id'])
@decorators.idempotent_id('fa43e7e5-0db9-48eb-9c6b-c11eb766b8e4')
@utils.requires_ext(extension='os-ips', service='compute')
@@ -648,9 +652,9 @@
'addresses']
address = next(iter(addresses))
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_addresses_by_network(
- self.server['id'], address)[address]
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_addresses_by_network(
+ self.server['id'], address)
@testtools.skipUnless(CONF.compute_feature_enabled.interface_attach,
"Interface attachment is not available.")
@@ -668,9 +672,9 @@
network_id = self.interfaces_client.create_interface(
self.server['id'])['interfaceAttachment']['net_id']
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.add_fixed_ip(self.server['id'],
- networkId=network_id)
+ with self.rbac_utils.override_role(self):
+ self.servers_client.add_fixed_ip(self.server['id'],
+ networkId=network_id)
@rbac_rule_validation.action(
service="nova",
@@ -685,10 +689,12 @@
For more information, see:
https://developer.openstack.org/api-ref/compute/#servers-virtual-interfaces-servers-os-virtual-interfaces-deprecated
"""
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- if CONF.service_available.neutron:
- msg = "Listing virtual interfaces is not supported by this cloud."
- with self.assertRaisesRegex(lib_exc.BadRequest, msg):
+ with self.rbac_utils.override_role(self):
+ if CONF.service_available.neutron:
+ msg = ("Listing virtual interfaces is not supported by this "
+ "cloud.")
+ with self.assertRaisesRegex(lib_exc.BadRequest, msg):
+ self.servers_client.list_virtual_interfaces(
+ self.server['id'])
+ else:
self.servers_client.list_virtual_interfaces(self.server['id'])
- else:
- self.servers_client.list_virtual_interfaces(self.server['id'])
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_rbac.py
index 0fe92a9..b95ebd5 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_rbac.py
@@ -33,15 +33,12 @@
class ComputeServersRbacTest(base.BaseV2ComputeRbacTest):
- credentials = ['primary', 'admin']
-
@classmethod
def setup_clients(cls):
super(ComputeServersRbacTest, cls).setup_clients()
cls.networks_client = cls.os_primary.networks_client
cls.ports_client = cls.os_primary.ports_client
cls.subnets_client = cls.os_primary.subnets_client
- cls.admin_servers_client = cls.os_admin.servers_client
@classmethod
def resource_setup(cls):
@@ -53,8 +50,8 @@
rule="os_compute_api:servers:create")
@decorators.idempotent_id('4f34c73a-6ddc-4677-976f-71320fa855bd')
def test_create_server(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_test_server(wait_until='ACTIVE')
+ with self.rbac_utils.override_role(self):
+ self.create_test_server(wait_until='ACTIVE')
@rbac_rule_validation.action(
service="nova",
@@ -72,9 +69,9 @@
host = list(hosts[0].keys())[0]
availability_zone = 'nova:' + host
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.create_test_server(wait_until='ACTIVE',
- availability_zone=availability_zone)
+ with self.rbac_utils.override_role(self):
+ self.create_test_server(wait_until='ACTIVE',
+ availability_zone=availability_zone)
@utils.services('volume')
@rbac_rule_validation.action(
@@ -97,10 +94,10 @@
'delete_on_termination': True}]
device_mapping = {'block_device_mapping_v2': bd_map_v2}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- # Use image_id='' to avoid using the default image in tempest.conf.
- server = self.create_test_server(name=server_name, image_id='',
- **device_mapping)
+ with self.rbac_utils.override_role(self):
+ # Use image_id='' to avoid using the default image in tempest.conf.
+ server = self.create_test_server(name=server_name, image_id='',
+ **device_mapping)
# Delete the server and wait for the volume to become available to
# avoid clean up errors.
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
@@ -140,9 +137,9 @@
network = _create_network_resources()
network_id = {'uuid': network['id']}
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- server = self.create_test_server(wait_until='ACTIVE',
- networks=[network_id])
+ with self.rbac_utils.override_role(self):
+ server = self.create_test_server(wait_until='ACTIVE',
+ networks=[network_id])
self.addCleanup(waiters.wait_for_server_termination,
self.servers_client, server['id'])
self.addCleanup(self.servers_client.delete_server, server['id'])
@@ -154,10 +151,10 @@
def test_delete_server(self):
server = self.create_test_server(wait_until='ACTIVE')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.delete_server(server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.delete_server(server['id'])
waiters.wait_for_server_termination(
- self.admin_servers_client, server['id'])
+ self.servers_client, server['id'])
@rbac_rule_validation.action(
service="nova",
@@ -165,13 +162,14 @@
@decorators.idempotent_id('077b17cb-5621-43b9-8adf-5725f0d7a863')
def test_update_server(self):
new_name = data_utils.rand_name(self.__class__.__name__ + '-server')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- try:
- self.servers_client.update_server(self.server['id'], name=new_name)
- waiters.wait_for_server_status(self.admin_servers_client,
- self.server['id'], 'ACTIVE')
- except exceptions.ServerFault as e:
- # Some other policy may have blocked it.
- LOG.info("ServerFault exception caught. Some other policy "
- "blocked updating of server")
- raise rbac_exceptions.RbacConflictingPolicies(e)
+ with self.rbac_utils.override_role(self):
+ try:
+ self.servers_client.update_server(self.server['id'],
+ name=new_name)
+ waiters.wait_for_server_status(self.servers_client,
+ self.server['id'], 'ACTIVE')
+ except exceptions.ServerFault as e:
+ # Some other policy may have blocked it.
+ LOG.info("ServerFault exception caught. Some other policy "
+ "blocked updating of server")
+ raise rbac_exceptions.RbacConflictingPolicies(e)
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_tags_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_tags_rbac.py
index 0f49095..70e7da9 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_tags_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_tags_rbac.py
@@ -49,8 +49,8 @@
service="nova",
rule="os_compute_api:os-server-tags:index")
def test_list_tags(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_tags(self.server['id'])['tags']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_tags(self.server['id'])
@decorators.idempotent_id('9297c99e-94eb-429f-93cf-9b1838e33622')
@rbac_rule_validation.action(
@@ -58,16 +58,17 @@
rule="os_compute_api:os-server-tags:show")
def test_check_tag_existence(self):
tag_name = self._add_tag_to_server()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.check_tag_existence(self.server['id'], tag_name)
+ with self.rbac_utils.override_role(self):
+ self.servers_client.check_tag_existence(self.server['id'],
+ tag_name)
@decorators.idempotent_id('0d84ee94-d3ca-4635-8edf-b7f67ab8e4a3')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-server-tags:update")
def test_update_tag(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._add_tag_to_server()
+ with self.rbac_utils.override_role(self):
+ self._add_tag_to_server()
@decorators.idempotent_id('115c2694-00aa-41ee-99f6-9eab4040c182')
@rbac_rule_validation.action(
@@ -75,23 +76,23 @@
rule="os_compute_api:os-server-tags:delete")
def test_delete_tag(self):
tag_name = self._add_tag_to_server()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.delete_tag(self.server['id'], tag_name)
+ with self.rbac_utils.override_role(self):
+ self.servers_client.delete_tag(self.server['id'], tag_name)
@decorators.idempotent_id('a8e19b87-6580-4bc8-9933-e62561ff667d')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-server-tags:update_all")
def test_update_all_tags(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
new_tag_name = data_utils.rand_name(self.__class__.__name__ + '-tag')
- self.servers_client.update_all_tags(self.server['id'],
- [new_tag_name])['tags']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.update_all_tags(self.server['id'],
+ [new_tag_name])
@decorators.idempotent_id('89d51936-e333-42f9-a045-132a4865ba1a')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-server-tags:delete_all")
def test_delete_all_tags(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.delete_all_tags(self.server['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.delete_all_tags(self.server['id'])
diff --git a/patrole_tempest_plugin/tests/api/compute/test_server_volume_attachments_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_server_volume_attachments_rbac.py
index 65d9edb..ed0c7a2 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_server_volume_attachments_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_server_volume_attachments_rbac.py
@@ -49,17 +49,16 @@
rule="os_compute_api:os-volumes-attachments:index")
@decorators.idempotent_id('529b668b-6edb-41d5-8886-d7dbd0614678')
def test_list_volume_attachments(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.list_volume_attachments(self.server['id'])[
- 'volumeAttachments']
+ with self.rbac_utils.override_role(self):
+ self.servers_client.list_volume_attachments(self.server['id'])
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-volumes-attachments:create")
@decorators.idempotent_id('21c2c3fd-fbe8-41b1-8ef8-115ec47d54c1')
def test_create_volume_attachment(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.attach_volume(self.server, self.volume)
+ with self.rbac_utils.override_role(self):
+ self.attach_volume(self.server, self.volume)
@rbac_rule_validation.action(
service="nova",
@@ -68,9 +67,9 @@
def test_show_volume_attachment(self):
attachment = self.attach_volume(self.server, self.volume)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.show_volume_attachment(
- self.server['id'], attachment['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.show_volume_attachment(
+ self.server['id'], attachment['id'])
@decorators.attr(type='slow')
@rbac_rule_validation.action(
@@ -81,9 +80,9 @@
attachment = self.attach_volume(self.server, self.volume)
alt_volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.update_attached_volume(
- self.server['id'], attachment['id'], volumeId=alt_volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.update_attached_volume(
+ self.server['id'], attachment['id'], volumeId=alt_volume['id'])
waiters.wait_for_volume_resource_status(self.volumes_client,
alt_volume['id'], 'in-use')
# On teardown detach the volume and wait for it to be available. This
@@ -105,7 +104,8 @@
def test_delete_volume_attachment(self):
self.attach_volume(self.server, self.volume)
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.servers_client.detach_volume(self.server['id'], self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.servers_client.detach_volume(self.server['id'],
+ self.volume['id'])
waiters.wait_for_volume_resource_status(self.volumes_client,
self.volume['id'], 'available')
diff --git a/patrole_tempest_plugin/tests/api/compute/test_services_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_services_rbac.py
index 316da00..183d990 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_services_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_services_rbac.py
@@ -34,5 +34,5 @@
rule="os_compute_api:os-services")
@decorators.idempotent_id('7472261b-9c6d-453a-bcb3-aecaa29ad281')
def test_list_services(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.services_client.list_services()['services']
+ with self.rbac_utils.override_role(self):
+ self.services_client.list_services()['services']
diff --git a/patrole_tempest_plugin/tests/api/compute/test_tenant_networks_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_tenant_networks_rbac.py
index 830e19a..19c7e33 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_tenant_networks_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_tenant_networks_rbac.py
@@ -57,5 +57,5 @@
service="nova",
rule="os_compute_api:os-tenant-networks")
def test_list_show_tenant_networks(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.tenant_networks_client.list_tenant_networks()['networks']
+ with self.rbac_utils.override_role(self):
+ self.tenant_networks_client.list_tenant_networks()
diff --git a/patrole_tempest_plugin/tests/api/compute/test_volume_rbac.py b/patrole_tempest_plugin/tests/api/compute/test_volume_rbac.py
index 9321fd1..b07fb3f 100644
--- a/patrole_tempest_plugin/tests/api/compute/test_volume_rbac.py
+++ b/patrole_tempest_plugin/tests/api/compute/test_volume_rbac.py
@@ -33,7 +33,6 @@
# For more information, see:
# https://developer.openstack.org/api-ref/compute/#volume-extension-os-volumes-os-snapshots-deprecated
max_microversion = '2.35'
- credentials = ['primary', 'admin']
@classmethod
def skip_checks(cls):
@@ -46,21 +45,16 @@
raise cls.skipException(skip_msg)
@classmethod
- def setup_clients(cls):
- super(VolumeRbacTest, cls).setup_clients()
- cls.admin_volumes_client = cls.os_admin.volumes_client_latest
-
- @classmethod
def resource_setup(cls):
super(VolumeRbacTest, cls).resource_setup()
cls.volume = cls.create_volume()
def _delete_snapshot(self, snapshot_id):
waiters.wait_for_volume_resource_status(
- self.os_admin.snapshots_extensions_client, snapshot_id,
+ self.snapshots_extensions_client, snapshot_id,
'available')
self.snapshots_extensions_client.delete_snapshot(snapshot_id)
- self.os_admin.snapshots_extensions_client.wait_for_resource_deletion(
+ self.snapshots_extensions_client.wait_for_resource_deletion(
snapshot_id)
@decorators.idempotent_id('2402013e-a624-43e3-9518-44a5d1dbb32d')
@@ -68,12 +62,10 @@
service="nova",
rule="os_compute_api:os-volumes")
def test_create_volume(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- volume = self.volumes_extensions_client.create_volume(
- size=CONF.volume.volume_size)['volume']
- # Use the admin volumes client to wait, because waiting involves
- # calling show API action which enforces a different policy.
- waiters.wait_for_volume_resource_status(self.admin_volumes_client,
+ with self.rbac_utils.override_role(self):
+ volume = self.volumes_extensions_client.create_volume(
+ size=CONF.volume.volume_size)['volume']
+ waiters.wait_for_volume_resource_status(self.volumes_client,
volume['id'], 'available')
# Use non-deprecated volumes_client for deletion.
self.addCleanup(self.volumes_client.delete_volume, volume['id'])
@@ -83,16 +75,16 @@
service="nova",
rule="os_compute_api:os-volumes")
def test_list_volumes(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_extensions_client.list_volumes()
+ with self.rbac_utils.override_role(self):
+ self.volumes_extensions_client.list_volumes()
@decorators.idempotent_id('4ba0a820-040f-488b-86bb-be2e920ea12c')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-volumes")
def test_show_volume(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_extensions_client.show_volume(self.volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_extensions_client.show_volume(self.volume['id'])
@decorators.idempotent_id('6e7870f2-1bb2-4b58-96f8-6782071ef327')
@rbac_rule_validation.action(
@@ -100,18 +92,18 @@
rule="os_compute_api:os-volumes")
def test_delete_volume(self):
volume = self.create_volume()
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.volumes_extensions_client.delete_volume(volume['id'])
+ with self.rbac_utils.override_role(self):
+ self.volumes_extensions_client.delete_volume(volume['id'])
@decorators.idempotent_id('0c3eaa4f-69d6-4a13-9dda-19585f36b1c1')
@rbac_rule_validation.action(
service="nova",
rule="os_compute_api:os-volumes")
def test_create_snapshot(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
s_name = data_utils.rand_name(self.__class__.__name__ + '-Snapshot')
- snapshot = self.snapshots_extensions_client.create_snapshot(
- volume_id=self.volume['id'], display_name=s_name)['snapshot']
+ with self.rbac_utils.override_role(self):
+ snapshot = self.snapshots_extensions_client.create_snapshot(
+ volume_id=self.volume['id'], display_name=s_name)['snapshot']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self._delete_snapshot, snapshot['id'])
@@ -120,8 +112,8 @@
service="nova",
rule="os_compute_api:os-volumes")
def test_list_snapshots(self):
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.snapshots_extensions_client.list_snapshots()
+ with self.rbac_utils.override_role(self):
+ self.snapshots_extensions_client.list_snapshots()
@decorators.idempotent_id('19c2e6bd-585b-472f-a8d7-71ea9299c655')
@rbac_rule_validation.action(
@@ -133,8 +125,8 @@
volume_id=self.volume['id'], display_name=s_name)['snapshot']
self.addCleanup(self._delete_snapshot, snapshot['id'])
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self.snapshots_extensions_client.show_snapshot(snapshot['id'])
+ with self.rbac_utils.override_role(self):
+ self.snapshots_extensions_client.show_snapshot(snapshot['id'])
@decorators.idempotent_id('f4f5635c-416c-11e7-a919-92ebcb67fe33')
@rbac_rule_validation.action(
@@ -146,6 +138,11 @@
volume_id=self.volume['id'], display_name=s_name)['snapshot']
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self._delete_snapshot, snapshot['id'])
+ waiters.wait_for_volume_resource_status(
+ self.snapshots_extensions_client, snapshot['id'],
+ 'available')
- self.rbac_utils.switch_role(self, toggle_rbac_role=True)
- self._delete_snapshot(snapshot['id'])
+ with self.rbac_utils.override_role(self):
+ self.snapshots_extensions_client.delete_snapshot(snapshot['id'])
+ self.snapshots_extensions_client.wait_for_resource_deletion(
+ snapshot['id'])