openstack/tempest
Code Issues Proposed changes

Use service user for nova service-to-service API tests

Browse Source 
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>
This commit is contained in:
Ghanshyam Maan
2025-08-26 02:41:04 +00:00
parent 290069b157
commit 538f6af8da
4 changed files with 44 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tempest/api/compute/admin/test_assisted_volume_snapshots.py
+7 -5
View File
@@ -26,11 +26,6 @@ class VolumesAssistedSnapshotsTest(base.BaseV2ComputeAdminTest):
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 @@ class VolumesAssistedSnapshotsTest(base.BaseV2ComputeAdminTest):
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):
tempest/api/compute/admin/test_server_external_events.py
-5
View File
@@ -19,11 +19,6 @@ from tempest.lib import decorators
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')
tempest/api/compute/admin/test_volume_swap.py
+21 -16
View File
@@ -24,6 +24,8 @@ CONF = config.CONF
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 @@ class TestVolumeSwapBase(base.BaseV2ComputeAdminTest):
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 TestVolumeSwapBase(base.BaseV2ComputeAdminTest):
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 @@ class TestVolumeSwap(TestVolumeSwapBase):
@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 @@ class TestVolumeSwap(TestVolumeSwapBase):
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 @@ class TestVolumeSwap(TestVolumeSwapBase):
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 @@ class TestMultiAttachVolumeSwap(TestVolumeSwapBase):
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 @@ class TestMultiAttachVolumeSwap(TestVolumeSwapBase):
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(
tempest/api/compute/admin/test_volumes_negative.py
+16 -4
View File
@@ -26,6 +26,7 @@ class VolumesAdminNegativeTest(base.BaseV2ComputeAdminTest):
"""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 @@ class VolumesAdminNegativeTest(base.BaseV2ComputeAdminTest):
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 @@ class VolumesAdminNegativeTest(base.BaseV2ComputeAdminTest):
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 @@ class VolumesAdminNegativeTest(base.BaseV2ComputeAdminTest):
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 @@ class UpdateMultiattachVolumeNegativeTest(base.BaseV2ComputeAdminTest):
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 @@ class UpdateMultiattachVolumeNegativeTest(base.BaseV2ComputeAdminTest):
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 @@ class UpdateMultiattachVolumeNegativeTest(base.BaseV2ComputeAdminTest):
# 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.