From 6250b98b5b0ec64b25ec9adfe4850a0f89c09e61 Mon Sep 17 00:00:00 2001 From: Stephen Finucane Date: Thu, 27 Feb 2025 13:36:22 +0000 Subject: [PATCH] tests: Drop support for testing identity v2 This is effectively impossible to support, given identity v2 was removed in Queens, way back in March 2018 (feeling old yet?). We also rename openstack/tests/functional/cloud/test_identity.py to openstack/tests/functional/cloud/test_roles.py to better capture its purpose. Change-Id: I997fc3553846ce1739119e6db4fe30c67a421a22 Signed-off-by: Stephen Finucane --- openstack/tests/functional/base.py | 17 ++--- .../tests/functional/cloud/test_domain.py | 7 +- .../tests/functional/cloud/test_endpoints.py | 69 ++++++++----------- .../tests/functional/cloud/test_groups.py | 6 +- .../tests/functional/cloud/test_project.py | 20 ++---- .../cloud/{test_identity.py => test_roles.py} | 30 +++----- .../tests/functional/cloud/test_services.py | 49 +++++-------- .../tests/functional/cloud/test_users.py | 14 ++-- .../v2/test_auto_allocated_topology.py | 7 +- 9 files changed, 81 insertions(+), 138 deletions(-) rename openstack/tests/functional/cloud/{test_identity.py => test_roles.py} (91%) diff --git a/openstack/tests/functional/base.py b/openstack/tests/functional/base.py index 24f6affdf..780176aaa 100644 --- a/openstack/tests/functional/base.py +++ b/openstack/tests/functional/base.py @@ -80,10 +80,6 @@ class BaseFunctionalTest(base.TestCase): self._set_user_cloud() self._set_operator_cloud() - self.identity_version = self.user_cloud.config.get_api_version( - 'identity' - ) - self.flavor = self._pick_flavor() self.image = self._pick_image() @@ -296,12 +292,7 @@ class KeystoneBaseFunctionalTest(BaseFunctionalTest): def setUp(self): super().setUp() - use_keystone_v2 = os.environ.get('OPENSTACKSDK_USE_KEYSTONE_V2', False) - if use_keystone_v2: - # keystone v2 has special behavior for the admin - # interface and some of the operations, so make a new cloud - # object with interface set to admin. - # We only do it for keystone tests on v2 because otherwise - # the admin interface is not a thing that wants to actually - # be used - self._set_operator_cloud(interface='admin') + # we only support v3, since v2 was deprecated in Queens (2018) + + if not self.conn.has_service('identity', '3'): + self.skipTest('identity service not supported by cloud') diff --git a/openstack/tests/functional/cloud/test_domain.py b/openstack/tests/functional/cloud/test_domain.py index 7cbd4b828..7c9a503bf 100644 --- a/openstack/tests/functional/cloud/test_domain.py +++ b/openstack/tests/functional/cloud/test_domain.py @@ -21,14 +21,13 @@ from openstack import exceptions from openstack.tests.functional import base -class TestDomain(base.BaseFunctionalTest): +class TestDomain(base.KeystoneBaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") - i_ver = self.operator_cloud.config.get_api_version('identity') - if i_ver in ('2', '2.0'): - self.skipTest('Identity service does not support domains') + self.domain_prefix = self.getUniqueString('domain') self.addCleanup(self._cleanup_domains) diff --git a/openstack/tests/functional/cloud/test_endpoints.py b/openstack/tests/functional/cloud/test_endpoints.py index 30feeb5f7..ed88057d1 100644 --- a/openstack/tests/functional/cloud/test_endpoints.py +++ b/openstack/tests/functional/cloud/test_endpoints.py @@ -22,7 +22,6 @@ Functional tests for endpoint resource. import random import string -from openstack.cloud.exc import OpenStackCloudUnavailableFeature from openstack import exceptions from openstack.tests.functional import base @@ -39,6 +38,7 @@ class TestEndpoints(base.KeystoneBaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") @@ -117,47 +117,38 @@ class TestEndpoints(base.KeystoneBaseFunctionalTest): self.assertIsNotNone(endpoints[0].get('id')) def test_update_endpoint(self): - ver = self.operator_cloud.config.get_api_version('identity') - if ver.startswith('2'): - # NOTE(SamYaple): Update endpoint only works with v3 api - self.assertRaises( - OpenStackCloudUnavailableFeature, - self.operator_cloud.update_endpoint, - 'endpoint_id1', - ) - else: - # service operations require existing region. Do not test updating - # region for now - region = list(self.operator_cloud.identity.regions())[0].id + # service operations require existing region. Do not test updating + # region for now + region = list(self.operator_cloud.identity.regions())[0].id - service = self.operator_cloud.create_service( - name='service1', type='test_type' - ) - endpoint = self.operator_cloud.create_endpoint( - service_name_or_id=service['id'], - url='http://admin.url/', - interface='admin', - region=region, - enabled=False, - )[0] + service = self.operator_cloud.create_service( + name='service1', type='test_type' + ) + endpoint = self.operator_cloud.create_endpoint( + service_name_or_id=service['id'], + url='http://admin.url/', + interface='admin', + region=region, + enabled=False, + )[0] - new_service = self.operator_cloud.create_service( - name='service2', type='test_type' - ) - new_endpoint = self.operator_cloud.update_endpoint( - endpoint.id, - service_name_or_id=new_service.id, - url='http://public.url/', - interface='public', - region=region, - enabled=True, - ) + new_service = self.operator_cloud.create_service( + name='service2', type='test_type' + ) + new_endpoint = self.operator_cloud.update_endpoint( + endpoint.id, + service_name_or_id=new_service.id, + url='http://public.url/', + interface='public', + region=region, + enabled=True, + ) - self.assertEqual(new_endpoint.url, 'http://public.url/') - self.assertEqual(new_endpoint.interface, 'public') - self.assertEqual(new_endpoint.region_id, region) - self.assertEqual(new_endpoint.service_id, new_service.id) - self.assertTrue(new_endpoint.is_enabled) + self.assertEqual(new_endpoint.url, 'http://public.url/') + self.assertEqual(new_endpoint.interface, 'public') + self.assertEqual(new_endpoint.region_id, region) + self.assertEqual(new_endpoint.service_id, new_service.id) + self.assertTrue(new_endpoint.is_enabled) def test_list_endpoints(self): service_name = self.new_item_name + '_list' diff --git a/openstack/tests/functional/cloud/test_groups.py b/openstack/tests/functional/cloud/test_groups.py index 6a9dd0ab4..fffbdfb79 100644 --- a/openstack/tests/functional/cloud/test_groups.py +++ b/openstack/tests/functional/cloud/test_groups.py @@ -21,15 +21,13 @@ from openstack import exceptions from openstack.tests.functional import base -class TestGroup(base.BaseFunctionalTest): +class TestGroup(base.KeystoneBaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") - i_ver = self.operator_cloud.config.get_api_version('identity') - if i_ver in ('2', '2.0'): - self.skipTest('Identity service does not support groups') self.group_prefix = self.getUniqueString('group') self.addCleanup(self._cleanup_groups) diff --git a/openstack/tests/functional/cloud/test_project.py b/openstack/tests/functional/cloud/test_project.py index 7924c9096..9331f9ee6 100644 --- a/openstack/tests/functional/cloud/test_project.py +++ b/openstack/tests/functional/cloud/test_project.py @@ -28,6 +28,7 @@ from openstack.tests.functional import base class TestProject(base.KeystoneBaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") @@ -52,11 +53,8 @@ class TestProject(base.KeystoneBaseFunctionalTest): params = { 'name': project_name, 'description': 'test_create_project', + 'domain_id': self.operator_cloud.get_domain('default')['id'], } - if self.identity_version == '3': - params['domain_id'] = self.operator_cloud.get_domain('default')[ - 'id' - ] project = self.operator_cloud.create_project(**params) @@ -94,11 +92,8 @@ class TestProject(base.KeystoneBaseFunctionalTest): 'name': project_name, 'description': 'test_update_project', 'enabled': True, + 'domain_id': self.operator_cloud.get_domain('default')['id'], } - if self.identity_version == '3': - params['domain_id'] = self.operator_cloud.get_domain('default')[ - 'id' - ] project = self.operator_cloud.create_project(**params) updated_project = self.operator_cloud.update_project( @@ -126,11 +121,10 @@ class TestProject(base.KeystoneBaseFunctionalTest): def test_delete_project(self): project_name = self.new_project_name + '_delete' - params = {'name': project_name} - if self.identity_version == '3': - params['domain_id'] = self.operator_cloud.get_domain('default')[ - 'id' - ] + params = { + 'name': project_name, + 'domain_id': self.operator_cloud.get_domain('default')['id'], + } project = self.operator_cloud.create_project(**params) self.assertIsNotNone(project) self.assertTrue(self.operator_cloud.delete_project(project['id'])) diff --git a/openstack/tests/functional/cloud/test_identity.py b/openstack/tests/functional/cloud/test_roles.py similarity index 91% rename from openstack/tests/functional/cloud/test_identity.py rename to openstack/tests/functional/cloud/test_roles.py index ae8b8ce3e..37ce193ff 100644 --- a/openstack/tests/functional/cloud/test_identity.py +++ b/openstack/tests/functional/cloud/test_roles.py @@ -11,10 +11,10 @@ # under the License. """ -test_identity +test_roles ---------------------------------- -Functional tests for identity methods. +Functional tests for role methods. """ import random @@ -24,11 +24,13 @@ from openstack import exceptions from openstack.tests.functional import base -class TestIdentity(base.KeystoneBaseFunctionalTest): +class TestRoles(base.KeystoneBaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") + self.role_prefix = 'test_role' + ''.join( random.choice(string.ascii_lowercase) for _ in range(5) ) @@ -36,8 +38,7 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): self.group_prefix = self.getUniqueString('group') self.addCleanup(self._cleanup_users) - if self.identity_version not in ('2', '2.0'): - self.addCleanup(self._cleanup_groups) + self.addCleanup(self._cleanup_groups) self.addCleanup(self._cleanup_roles) def _cleanup_groups(self): @@ -82,11 +83,10 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): raise exceptions.SDKException('\n'.join(exception_list)) def _create_user(self, **kwargs): - domain_id = None - if self.identity_version not in ('2', '2.0'): - domain = self.operator_cloud.get_domain('default') - domain_id = domain['id'] - return self.operator_cloud.create_user(domain_id=domain_id, **kwargs) + domain = self.operator_cloud.get_domain('default') + return self.operator_cloud.create_user( + domain_id=domain['id'], **kwargs + ) def test_list_roles(self): roles = self.operator_cloud.list_roles() @@ -124,8 +124,6 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): # need to make this test a little more specific, and add more for testing # filtering functionality. def test_list_role_assignments(self): - if self.identity_version in ('2', '2.0'): - self.skipTest("Identity service does not support role assignments") assignments = self.operator_cloud.list_role_assignments() self.assertIsInstance(assignments, list) self.assertGreater(len(assignments), 0) @@ -177,8 +175,6 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): self.assertEqual(0, len(assignments)) def test_grant_revoke_role_group_project(self): - if self.identity_version in ('2', '2.0'): - self.skipTest("Identity service does not support group") role_name = self.role_prefix + '_grant_group_project' role = self.operator_cloud.create_role(role_name) group_name = self.group_prefix + '_group_project' @@ -215,8 +211,6 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): self.assertEqual(0, len(assignments)) def test_grant_revoke_role_user_domain(self): - if self.identity_version in ('2', '2.0'): - self.skipTest("Identity service does not support domain") role_name = self.role_prefix + '_grant_user_domain' role = self.operator_cloud.create_role(role_name) user_name = self.user_prefix + '_user_domain' @@ -254,8 +248,6 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): self.assertEqual(0, len(assignments)) def test_grant_revoke_role_group_domain(self): - if self.identity_version in ('2', '2.0'): - self.skipTest("Identity service does not support domain or group") role_name = self.role_prefix + '_grant_group_domain' role = self.operator_cloud.create_role(role_name) group_name = self.group_prefix + '_group_domain' @@ -321,8 +313,6 @@ class TestIdentity(base.KeystoneBaseFunctionalTest): self.assertEqual(0, len(assignments)) def test_grant_revoke_role_group_system(self): - if self.identity_version in ('2', '2.0'): - self.skipTest("Identity service does not support system or group") role_name = self.role_prefix + '_grant_group_system' role = self.operator_cloud.create_role(role_name) group_name = self.group_prefix + '_group_system' diff --git a/openstack/tests/functional/cloud/test_services.py b/openstack/tests/functional/cloud/test_services.py index df311d3dc..48df7cd9f 100644 --- a/openstack/tests/functional/cloud/test_services.py +++ b/openstack/tests/functional/cloud/test_services.py @@ -22,7 +22,6 @@ Functional tests for service resource. import random import string -from openstack.cloud import exc from openstack import exceptions from openstack.tests.functional import base @@ -68,36 +67,24 @@ class TestServices(base.KeystoneBaseFunctionalTest): self.assertIsNotNone(service.get('id')) def test_update_service(self): - ver = self.operator_cloud.config.get_api_version('identity') - if ver.startswith('2'): - # NOTE(SamYaple): Update service only works with v3 api - self.assertRaises( - exc.OpenStackCloudUnavailableFeature, - self.operator_cloud.update_service, - 'service_id', - name='new name', - ) - else: - service = self.operator_cloud.create_service( - name=self.new_service_name + '_create', - type='test_type', - description='this is a test description', - enabled=True, - ) - new_service = self.operator_cloud.update_service( - service.id, - name=self.new_service_name + '_update', - description='this is an updated description', - enabled=False, - ) - self.assertEqual( - new_service.name, self.new_service_name + '_update' - ) - self.assertEqual( - new_service.description, 'this is an updated description' - ) - self.assertFalse(new_service.is_enabled) - self.assertEqual(service.id, new_service.id) + service = self.operator_cloud.create_service( + name=self.new_service_name + '_create', + type='test_type', + description='this is a test description', + enabled=True, + ) + new_service = self.operator_cloud.update_service( + service.id, + name=self.new_service_name + '_update', + description='this is an updated description', + enabled=False, + ) + self.assertEqual(new_service.name, self.new_service_name + '_update') + self.assertEqual( + new_service.description, 'this is an updated description' + ) + self.assertFalse(new_service.is_enabled) + self.assertEqual(service.id, new_service.id) def test_list_services(self): service = self.operator_cloud.create_service( diff --git a/openstack/tests/functional/cloud/test_users.py b/openstack/tests/functional/cloud/test_users.py index cdce80bdb..9c21bee0e 100644 --- a/openstack/tests/functional/cloud/test_users.py +++ b/openstack/tests/functional/cloud/test_users.py @@ -44,12 +44,10 @@ class TestUsers(base.KeystoneBaseFunctionalTest): raise exceptions.SDKException('\n'.join(exception_list)) def _create_user(self, **kwargs): - domain_id = None - i_ver = self.operator_cloud.config.get_api_version('identity') - if i_ver not in ('2', '2.0'): - domain = self.operator_cloud.get_domain('default') - domain_id = domain['id'] - return self.operator_cloud.create_user(domain_id=domain_id, **kwargs) + domain = self.operator_cloud.get_domain('default') + return self.operator_cloud.create_user( + domain_id=domain['id'], **kwargs + ) def test_list_users(self): users = self.operator_cloud.list_users() @@ -154,10 +152,6 @@ class TestUsers(base.KeystoneBaseFunctionalTest): self.assertIsNotNone(new_cloud.service_catalog) def test_users_and_groups(self): - i_ver = self.operator_cloud.config.get_api_version('identity') - if i_ver in ('2', '2.0'): - self.skipTest('Identity service does not support groups') - group_name = self.getUniqueString('group') self.addCleanup(self.operator_cloud.delete_group, group_name) diff --git a/openstack/tests/functional/network/v2/test_auto_allocated_topology.py b/openstack/tests/functional/network/v2/test_auto_allocated_topology.py index 3716eeca6..ff1bdd636 100644 --- a/openstack/tests/functional/network/v2/test_auto_allocated_topology.py +++ b/openstack/tests/functional/network/v2/test_auto_allocated_topology.py @@ -20,8 +20,10 @@ class TestAutoAllocatedTopology(base.BaseFunctionalTest): def setUp(self): super().setUp() + if not self.operator_cloud: self.skipTest("Operator cloud is required for this test") + if not self.operator_cloud._has_neutron_extension( "auto-allocated-topology" ): @@ -55,11 +57,8 @@ class TestAutoAllocatedTopology(base.BaseFunctionalTest): 'test project used only for the ' 'TestAutoAllocatedTopology tests class' ), + 'domain_id': self.operator_cloud.get_domain('default')['id'], } - if self.identity_version == '3': - params['domain_id'] = self.operator_cloud.get_domain( - 'default' - )['id'] project = self.operator_cloud.create_project(**params)