From 538f6af8da510e9286ddc762182cfa7170ad7b03 Mon Sep 17 00:00:00 2001 From: Ghanshyam Maan Date: Tue, 26 Aug 2025 02:41:04 +0000 Subject: [PATCH] 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 --- .../admin/test_assisted_volume_snapshots.py | 12 +++--- .../admin/test_server_external_events.py | 5 --- tempest/api/compute/admin/test_volume_swap.py | 37 +++++++++++-------- .../compute/admin/test_volumes_negative.py | 20 ++++++++-- 4 files changed, 44 insertions(+), 30 deletions(-) diff --git a/tempest/api/compute/admin/test_assisted_volume_snapshots.py b/tempest/api/compute/admin/test_assisted_volume_snapshots.py index b7be796ad6..034efc9353 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 @@ 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): diff --git a/tempest/api/compute/admin/test_server_external_events.py b/tempest/api/compute/admin/test_server_external_events.py index d867a399bf..6b26c68500 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 @@ 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') diff --git a/tempest/api/compute/admin/test_volume_swap.py b/tempest/api/compute/admin/test_volume_swap.py index 24b3ee4025..423407e04a 100644 --- a/tempest/api/compute/admin/test_volume_swap.py +++ b/tempest/api/compute/admin/test_volume_swap.py @@ -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( diff --git a/tempest/api/compute/admin/test_volumes_negative.py b/tempest/api/compute/admin/test_volumes_negative.py index e3c33595ef..3d42388051 100644 --- a/tempest/api/compute/admin/test_volumes_negative.py +++ b/tempest/api/compute/admin/test_volumes_negative.py @@ -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.