Use service user for nova service-to-service API tests
Nova has service-to-service internal APIs which are now
default to service role[1]. Token passed to
Nova needs to have 'service' role so that Nova use
service token auth data to validate if request is from
the OpenStack services or not.
Tempest test for server external event and volume swap
needs to add 'admin' role also in service user because
Nova need to access the server or volume resource of other
project which need 'admin' role.
Depends-On: https://review.opendev.org/c/openstack/devstack/+/958612
[1] https://review.opendev.org/c/openstack/nova/+/957578
Change-Id: I56ed7663ae1ffe4377a644d307e02921c3982fa5
Signed-off-by: Ghanshyam Maan <gmaan@ghanshyammann.com>
diff --git a/tempest/api/compute/admin/test_assisted_volume_snapshots.py b/tempest/api/compute/admin/test_assisted_volume_snapshots.py
index b7be796..034efc9 100644
--- a/tempest/api/compute/admin/test_assisted_volume_snapshots.py
+++ b/tempest/api/compute/admin/test_assisted_volume_snapshots.py
@@ -26,11 +26,6 @@
create_default_network = True
- # TODO(gmann): Remove the admin access to service user
- # once nova change the default of this API to service
- # role. To merge the nova changing the policy default
- # we need to use token with admin as well as service
- # role and later we can use only service token.
credentials = ['primary', 'admin', ['service_user', 'admin', 'service']]
@classmethod
@@ -39,6 +34,13 @@
if not CONF.service_available.cinder:
skip_msg = ("%s skipped as Cinder is not available" % cls.__name__)
raise cls.skipException(skip_msg)
+ # NOTE(gmaan): If new policy is enforced and and service role
+ # is present in nova then use service user (no admin role) for
+ # assisted volume snapshots APIs.
+ if (CONF.enforce_scope.nova and 'service' in
+ CONF.compute_feature_enabled.nova_policy_roles):
+ cls.credentials = [
+ 'primary', 'admin', ['service_user', 'service']]
@classmethod
def setup_clients(cls):
diff --git a/tempest/api/compute/admin/test_server_external_events.py b/tempest/api/compute/admin/test_server_external_events.py
index d867a39..6b26c68 100644
--- a/tempest/api/compute/admin/test_server_external_events.py
+++ b/tempest/api/compute/admin/test_server_external_events.py
@@ -19,11 +19,6 @@
class ServerExternalEventsTest(base.BaseV2ComputeAdminTest):
"""Test server external events test"""
- # TODO(gmann): Remove the admin access to service user
- # once nova change the default of this API to service
- # role. To merge the nova changing the policy default
- # we need to use token with admin as well as service
- # role and later we can use only service token.
credentials = ['primary', 'admin', ['service_user', 'admin', 'service']]
@decorators.idempotent_id('6bbf4723-61d2-4372-af55-7ba27f1c9ba6')
diff --git a/tempest/api/compute/admin/test_volume_swap.py b/tempest/api/compute/admin/test_volume_swap.py
index 24b3ee4..423407e 100644
--- a/tempest/api/compute/admin/test_volume_swap.py
+++ b/tempest/api/compute/admin/test_volume_swap.py
@@ -24,6 +24,8 @@
class TestVolumeSwapBase(base.BaseV2ComputeAdminTest):
create_default_network = True
+ credentials = ['primary', 'admin', ['service_user', 'admin', 'service']]
+
@classmethod
def setup_credentials(cls):
cls.prepare_instance_network()
@@ -37,6 +39,11 @@
if not CONF.compute_feature_enabled.swap_volume:
raise cls.skipException("Swapping volumes is not supported.")
+ @classmethod
+ def setup_clients(cls):
+ super(TestVolumeSwapBase, cls).setup_clients()
+ cls.service_client = cls.os_service_user.servers_client
+
def wait_for_server_volume_swap(self, server_id, old_volume_id,
new_volume_id):
"""Waits for a server to swap the old volume to a new one."""
@@ -74,7 +81,7 @@
class TestVolumeSwap(TestVolumeSwapBase):
- """The test suite for swapping of volume with admin user"""
+ """The test suite for swapping of volume with service user"""
# NOTE(mriedem): This is an uncommon scenario to call the compute API
# to swap volumes directly; swap volume is primarily only for volume
@@ -84,7 +91,7 @@
@decorators.attr(type='slow')
@decorators.idempotent_id('1769f00d-a693-4d67-a631-6a3496773813')
def test_volume_swap(self):
- """Test swapping of volume attached to server with admin user
+ """Test swapping of volume attached to server with service user
The following is the scenario outline:
@@ -92,7 +99,7 @@
2. Create a volume "volume2" with non-admin.
3. Boot an instance "instance1" with non-admin.
4. Attach "volume1" to "instance1" with non-admin.
- 5. Swap volume from "volume1" to "volume2" as admin.
+ 5. Swap volume from "volume1" to "volume2" as service.
6. Check the swap volume is rejected and "volume1"
is attached to "instance1" and "volume2" is in available state.
"""
@@ -117,7 +124,7 @@
self.attach_volume(server, volume1)
# Swap volume from "volume1" to "volume2"
self.assertRaises(
- lib_exc.Conflict, self.admin_servers_client.update_attached_volume,
+ lib_exc.Conflict, self.service_client.update_attached_volume,
server['id'], volume1['id'], volumeId=volume2['id'])
# Verify "volume1" is attached to the server
vol_attachments = self.servers_client.list_volume_attachments(
@@ -177,9 +184,8 @@
4. Attach "volume1" to "server1" with non-admin.
5. Attach "volume1" to "server2" with non-admin.
6. Swap "volume1" to "volume2" on "server1"
- 7. Check "volume1" is attached to "server2" and not attached to
- "server1"
- 8. Check "volume2" is attached to "server1".
+ 7. Check the swap volume is rejected and "volume1" is attached to
+ "server1".
"""
multiattach_vol_type = CONF.volume.volume_type_multiattach
# Create two volumes.
@@ -220,23 +226,22 @@
server2 = servers[1]
self.attach_volume(server2, volume1)
- # Swap volume1 to volume2 on server1, volume1 should remain attached
- # to server 2
- self.admin_servers_client.update_attached_volume(
+ # Swap volume is rejected by nova
+ self.assertRaises(
+ lib_exc.Conflict, self.service_client.update_attached_volume,
server1['id'], volume1['id'], volumeId=volume2['id'])
- # volume1 will return to in-use after the swap
+
+ # volume1 remains in in-use and volume2 in available
waiters.wait_for_volume_resource_status(self.volumes_client,
volume1['id'], 'in-use')
waiters.wait_for_volume_resource_status(self.volumes_client,
- volume2['id'], 'in-use')
- self.wait_for_server_volume_swap(server1['id'], volume1['id'],
- volume2['id'])
+ volume2['id'], 'available')
- # Verify volume2 is attached to server1
+ # Verify volume1 is attached to server1
vol_attachments = self.servers_client.list_volume_attachments(
server1['id'])['volumeAttachments']
self.assertEqual(1, len(vol_attachments))
- self.assertIn(volume2['id'], vol_attachments[0]['volumeId'])
+ self.assertIn(volume1['id'], vol_attachments[0]['volumeId'])
# Verify volume1 is still attached to server2
vol_attachments = self.servers_client.list_volume_attachments(
diff --git a/tempest/api/compute/admin/test_volumes_negative.py b/tempest/api/compute/admin/test_volumes_negative.py
index e3c3359..3d42388 100644
--- a/tempest/api/compute/admin/test_volumes_negative.py
+++ b/tempest/api/compute/admin/test_volumes_negative.py
@@ -26,6 +26,7 @@
"""Negative tests of volume swapping"""
create_default_network = True
+ credentials = ['primary', 'admin', ['service_user', 'admin', 'service']]
@classmethod
def setup_credentials(cls):
@@ -39,6 +40,11 @@
skip_msg = ("%s skipped as Cinder is not available" % cls.__name__)
raise cls.skipException(skip_msg)
+ @classmethod
+ def setup_clients(cls):
+ super(VolumesAdminNegativeTest, cls).setup_clients()
+ cls.service_client = cls.os_service_user.servers_client
+
@decorators.attr(type=['negative'])
@decorators.idempotent_id('309b5ecd-0585-4a7e-a36f-d2b2bf55259d')
def test_update_attached_volume_with_nonexistent_volume_in_uri(self):
@@ -47,7 +53,7 @@
volume = self.create_volume()
nonexistent_volume = data_utils.rand_uuid()
self.assertRaises(lib_exc.NotFound,
- self.admin_servers_client.update_attached_volume,
+ self.service_client.update_attached_volume,
self.server['id'], nonexistent_volume,
volumeId=volume['id'])
@@ -73,7 +79,7 @@
nonexistent_volume = data_utils.rand_uuid()
self.assertRaises(lib_exc.Conflict,
- self.admin_servers_client.update_attached_volume,
+ self.service_servers_client.update_attached_volume,
self.server['id'], volume['id'],
volumeId=nonexistent_volume)
@@ -90,6 +96,7 @@
volume_min_microversion = '3.27'
create_default_network = True
+ credentials = ['primary', 'admin', ['service_user', 'admin', 'service']]
@classmethod
def setup_credentials(cls):
@@ -102,6 +109,11 @@
if not CONF.compute_feature_enabled.volume_multiattach:
raise cls.skipException('Volume multi-attach is not available.')
+ @classmethod
+ def setup_clients(cls):
+ super(UpdateMultiattachVolumeNegativeTest, cls).setup_clients()
+ cls.service_client = cls.os_service_user.servers_client
+
@decorators.attr(type=['negative'])
@decorators.idempotent_id('7576d497-b7c6-44bd-9cc5-c5b4e50fec71')
@utils.services('volume')
@@ -160,10 +172,10 @@
# Assert that a BadRequest is raised when we attempt to update volume1
# to volume2 on server1 or server2.
self.assertRaises(lib_exc.BadRequest,
- self.admin_servers_client.update_attached_volume,
+ self.service_client.update_attached_volume,
server1['id'], vol1['id'], volumeId=vol2['id'])
self.assertRaises(lib_exc.BadRequest,
- self.admin_servers_client.update_attached_volume,
+ self.service_client.update_attached_volume,
server2['id'], vol1['id'], volumeId=vol2['id'])
# Fetch the volume 1 to check the current attachments.