Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _create_consumer method in tempest

Best Python code snippet using tempest_python

test_oauth1_tokens.py

Source:test_oauth1_tokens.py

...16from tempest.lib import decorators17from tempest.lib import exceptions as lib_exc18from keystone_tempest_plugin.tests import base19class OAUTH1TokensTest(base.BaseIdentityTest):20    def _create_consumer(self):21        """Creates a consumer with a random description."""22        description = data_utils.rand_name('test_create_consumer')23        consumer = self.consumers_client.create_consumer(24            description)['consumer']25        # cleans up created consumers after tests26        self.addCleanup(test_utils.call_and_ignore_notfound_exc,27                        self.consumers_client.delete_consumer,28                        consumer['id'])29        return consumer30    def _create_request_token(self, consumer):31        """Create request token to authorize access for a consumer."""32        project_id = (33            self.oauth_token_client.auth_provider.credentials.project_id)34        request_token = self.oauth_token_client.create_request_token(35            consumer['id'], consumer['secret'], project_id)36        return request_token37    def _authorize_request_token(self, request_token):38        """Authorize request token to provide access to specific roles."""39        user_id = self.oauth_token_client.auth_provider.credentials.user_id40        project_id = (41            self.oauth_token_client.auth_provider.credentials.project_id)42        roles = self.roles_client.list_user_roles_on_project(43            project_id, user_id)44        role_ids = [role['id'] for role in roles['roles']]45        oauth_verifier = self.oauth_token_client.authorize_request_token(46            request_token['oauth_token'], role_ids)47        return oauth_verifier['token']48    def _create_access_token(self, consumer):49        """Create access token for a consumer."""50        request_token = self._create_request_token(consumer)51        oauth_verifier = self._authorize_request_token(request_token)52        access_token = self.oauth_token_client.create_access_token(53            consumer['id'], consumer['secret'],54            request_token['oauth_token'],55            request_token['oauth_token_secret'],56            oauth_verifier['oauth_verifier'])57        # cleans up access tokens after tests58        user_id = self.oauth_token_client.auth_provider.credentials.user_id59        self.addCleanup(test_utils.call_and_ignore_notfound_exc,60                        self.oauth_token_client.revoke_access_token,61                        user_id, access_token['oauth_token'])62        return access_token63    @decorators.idempotent_id('23d2fe8d-fc8d-4bef-8aaa-289400732c3f')64    def test_create_and_show_consumer(self):65        """Tests to make sure that a consumer with parameters is made."""66        consumer = self._create_consumer()67        # fetch created consumer from client68        fetched_consumer = self.consumers_client.show_consumer(69            consumer['id'])['consumer']70        # assert that the fetched consumer matches the created one and71        # has all parameters72        for key in ['description', 'id', 'links']:73            self.assertEqual(consumer[key], fetched_consumer[key])74    @decorators.idempotent_id('3820f3d0-9b06-4d15-8f01-c7dd4eea30a2')75    def test_delete_consumer(self):76        """Tests the delete functionality for resource consumer."""77        consumer = self._create_consumer()78        # fetch consumer from client to confirm it exists79        fetched_consumer = self.consumers_client.show_consumer(80            consumer['id'])['consumer']81        self.assertEqual(consumer['id'], fetched_consumer['id'])82        # delete existing consumer83        self.consumers_client.delete_consumer(consumer['id'])84        # check that consumer no longer exists85        self.assertRaises(lib_exc.NotFound,86                          self.consumers_client.show_consumer,87                          consumer['id'])88    @decorators.idempotent_id('5a03fa78-3a03-449b-a04c-ef9de80eb6c4')89    def test_update_consumer(self):90        """Tests the update functionality for resource consumer."""91        # create a new consumer to update92        consumer = self._create_consumer()93        # create new description94        new_description = data_utils.rand_name('test_update_consumer')95        # update consumer96        self.consumers_client.update_consumer(consumer['id'],97                                              new_description)98        # check that the same consumer now has the new description99        updated_consumer = self.consumers_client.show_consumer(100            consumer['id'])['consumer']101        self.assertEqual(new_description, updated_consumer['description'])102    @decorators.idempotent_id('6da689b1-39a0-44ee-9624-445159119c57')103    def test_list_consumers(self):104        """Test for listing consumers."""105        # create two consumers to populate list106        new_consumer_one = self._create_consumer()107        new_consumer_two = self._create_consumer()108        # fetch the list of consumers109        consumer_list = self.consumers_client.list_consumers()['consumers']110        # add fetched consumer ids to a list111        id_list = [consumer['id'] for consumer in consumer_list]112        # check if created consumers are in the list113        self.assertIn(new_consumer_one['id'], id_list)114        self.assertIn(new_consumer_two['id'], id_list)115    @decorators.idempotent_id('a17d60e4-7cb5-4e06-9e16-b044f3ee6d94')116    def test_create_request_token(self):117        """Test to create request token for consumer."""118        consumer = self._create_consumer()119        request_token = self._create_request_token(consumer)120        # check that oauth token id and secret exists121        self.assertIsNotNone(request_token['oauth_token'])122        self.assertIsNotNone(request_token['oauth_token_secret'])123    @decorators.idempotent_id('607aecc4-a623-4566-a3a5-bb0e2a6fc9c5')124    def test_authorize_request_token(self):125        """Test to authorize a request token for specific role."""126        consumer = self._create_consumer()127        request_token = self._create_request_token(consumer)128        oauth_verifier = self._authorize_request_token(request_token)129        # check that oauth verifier exists130        self.assertIsNotNone(oauth_verifier['oauth_verifier'])131    @decorators.idempotent_id('7d488fc9-342c-4c12-b6b8-b158e2183925')132    def test_create_access_token(self):133        """Test to create access token for consumer."""134        consumer = self._create_consumer()135        access_token = self._create_access_token(consumer)136        user_id = self.oauth_token_client.auth_provider.credentials.user_id137        project_id = (138            self.oauth_token_client.auth_provider.credentials.project_id)139        fetched_access_token = self.oauth_token_client.get_access_token(140            user_id, access_token['oauth_token'])['access_token']141        # check that access token details matches142        self.assertEqual(access_token['oauth_token'],143                         fetched_access_token['id'])144        self.assertEqual(consumer['id'], fetched_access_token['consumer_id'])145        self.assertEqual(access_token['oauth_expires_at'],146                         fetched_access_token['expires_at'])147        self.assertEqual(project_id, fetched_access_token['project_id'])148        self.assertEqual(user_id, fetched_access_token['authorizing_user_id'])149    @decorators.idempotent_id('1b802896-91a0-4cbb-a8b9-860c7087fad8')150    def test_revoke_access_token(self):151        """Test to delete the access token provided for consumer."""152        consumer = self._create_consumer()153        access_token = self._create_access_token(consumer)154        user_id = self.oauth_token_client.auth_provider.credentials.user_id155        # delete existing access token156        self.oauth_token_client.revoke_access_token(157            user_id, access_token['oauth_token'])158        # check that access token no longer exist159        self.assertRaises(lib_exc.NotFound,160                          self.oauth_token_client.get_access_token,161                          user_id,162                          access_token['oauth_token'])163    @decorators.idempotent_id('5929055d-7c0f-4661-a9a5-15c4b95082dc')164    def test_list_access_tokens(self):165        """Test to list access tokens provided to consumer."""166        # create two consumers and access token for each167        new_consumer_one = self._create_consumer()168        new_consumer_two = self._create_consumer()169        access_token_one = self._create_access_token(new_consumer_one)170        access_token_two = self._create_access_token(new_consumer_two)171        user_id = self.oauth_token_client.auth_provider.credentials.user_id172        # fetch the list of access tokens173        access_token_list = self.oauth_token_client.list_access_tokens(174            user_id)['access_tokens']175        # add fetch access token ids to a list176        id_list = [access_token['id'] for access_token in access_token_list]177        # check if created access tokens are in the list178        self.assertIn(access_token_one['oauth_token'], id_list)179        self.assertIn(access_token_two['oauth_token'], id_list)180    @decorators.idempotent_id('0075f413-e249-42e5-9bc9-d6e3aecf6cbc')181    def test_list_roles_for_access_token(self):182        """Test to list roles for an access token."""183        consumer = self._create_consumer()184        access_token = self._create_access_token(consumer)185        user_id = self.oauth_token_client.auth_provider.credentials.user_id186        project_id = (187            self.oauth_token_client.auth_provider.credentials.project_id)188        fetched_roles = self.oauth_token_client.list_access_token_roles(189            user_id, access_token['oauth_token'])['roles']190        fetched_role_ids = [role['id'] for role in fetched_roles]191        roles = self.roles_client.list_user_roles_on_project(192            project_id, user_id)193        role_ids = [role['id'] for role in roles['roles']]194        # check that role ids matches195        self.assertCountEqual(fetched_role_ids, role_ids)196    @decorators.idempotent_id('28aee994-86b1-4596-a652-572f558045e7')197    def test_show_role_for_access_token(self):198        """Test to show role details for an access token."""199        consumer = self._create_consumer()200        access_token = self._create_access_token(consumer)201        user_id = self.oauth_token_client.auth_provider.credentials.user_id202        project_id = (203            self.oauth_token_client.auth_provider.credentials.project_id)204        roles = self.roles_client.list_user_roles_on_project(205            project_id, user_id)206        fetched_role = self.oauth_token_client.get_access_token_role(207            user_id,208            access_token['oauth_token'],209            roles['roles'][0]['id'])210        # check that role id matches...

test_repositories_secret_consumers.py

Source:test_repositories_secret_consumers.py

...26        self.secret = self._create_secret()27        self.session.commit()28    def _create_secret(self):29        return utils.create_secret(self.project, session=self.session)30    def _create_consumer(self, secret=None, resource_id=0):31        if secret is None:32            secret = self.secret33        return utils.create_secret_consumer(34            secret,35            resource_id="resource_id_{}".format(resource_id),36            session=self.session,37        )38    def _count_consumers(self):39        return self.consumer_repo.get_count(40            self.project.id, session=self.session41        )42    def test_should_raise_no_result_found_get_by_secret_id(self):43        self.assertRaises(44            exception.NotFound,45            self.consumer_repo.get_by_secret_id,46            self.secret.id,47            session=self.session,48        )49    def test_get_by_secret_id(self):50        for resource_id in [1, 2, 3]:51            self._create_consumer(resource_id=resource_id)52        self.assertEqual(53            3, self.consumer_repo.get_by_secret_id(self.secret.id)[3]54        )55    def test_should_raise_no_result_found_get_by_resource_id(self):56        self.assertRaises(57            exception.NotFound,58            self.consumer_repo.get_by_resource_id,59            "my resource id",60            session=self.session,61        )62    def test_get_by_resource_id(self):63        secret1 = self._create_secret()64        secret2 = self._create_secret()65        secret3 = self._create_secret()66        for secret in [secret1, secret2, secret3]:67            self._create_consumer(secret=secret)68        self.assertEqual(69            3, self.consumer_repo.get_by_resource_id("resource_id_0")[3]70        )71    def test_should_update_with_duplicate_consumer(self):72        consumer1 = self._create_consumer()73        self.assertEqual(1, len(self.secret.consumers))74        # Commit things so far, because the 'create_or_update_from' call below75        # will handle consumer metadata with same composite key items already76        # existing, and then rollback this session's transaction, which would77        # remove the items added above and result in a not-found error below.78        self.session.commit()79        consumer2 = models.SecretConsumerMetadatum(80            secret_id=consumer1.secret_id,81            project_id=consumer1.project_id,82            service=consumer1.service,83            resource_type=consumer1.resource_type,84            resource_id=consumer1.resource_id,85        )86        self.consumer_repo.create_or_update_from(87            consumer2, self.secret, self.session88        )89        secret = self.secret_repo.get_secret_by_id(90            self.secret.id, session=self.session91        )92        self.assertEqual(1, len(secret.consumers))93    def test_should_raise_constraint_create_same_composite_key_no_id(self):94        self._create_consumer()95        exception_result = self.assertRaises(96            exception.ConstraintCheck, self._create_consumer97        )98        self.assertIn(99            "SQL constraint check failed", six.text_type(exception_result)100        )101    def test_should_get_count_zero(self):102        self.assertEqual(0, self._count_consumers())103    def test_should_get_count_one(self):104        self._create_consumer()105        self.assertEqual(1, self._count_consumers())106    def test_should_get_count_one_after_delete(self):107        consumer1 = self._create_consumer(resource_id=1)108        self._create_consumer(resource_id=2)109        self.assertEqual(2, self._count_consumers())110        self.consumer_repo.delete_entity_by_id(111            consumer1.id, consumer1.project_id, session=self.session112        )...

test_oauth_consumers.py

Source:test_oauth_consumers.py

...21    # NOTE: force_tenant_isolation is true in the base class by default but22    # overridden to false here to allow test execution for clouds using the23    # pre-provisioned credentials provider.24    force_tenant_isolation = False25    def _create_consumer(self):26        """Creates a consumer with a random description."""27        description = data_utils.rand_name('test_create_consumer')28        consumer = self.oauth_consumers_client.create_consumer(29            description)['consumer']30        # cleans up created consumers after tests31        self.addCleanup(test_utils.call_and_ignore_notfound_exc,32                        self.oauth_consumers_client.delete_consumer,33                        consumer['id'])34        return consumer35    @decorators.idempotent_id('c8307ea6-a86c-47fd-ae7b-5b3b2caca76d')36    def test_create_and_show_consumer(self):37        """Tests to make sure that a consumer with parameters is made"""38        consumer = self._create_consumer()39        # fetch created consumer from client40        fetched_consumer = self.oauth_consumers_client.show_consumer(41            consumer['id'])['consumer']42        # assert that the fetched consumer matches the created one and43        # has all parameters44        for key in ['description', 'id', 'links']:45            self.assertEqual(consumer[key], fetched_consumer[key])46    @decorators.idempotent_id('fdfa1b7f-2a31-4354-b2c7-f6ae20554f93')47    def test_delete_consumer(self):48        """Tests the delete function."""49        consumer = self._create_consumer()50        # fetch consumer from client to confirm it exists51        fetched_consumer = self.oauth_consumers_client.show_consumer(52            consumer['id'])['consumer']53        self.assertEqual(consumer['id'], fetched_consumer['id'])54        # delete existing consumer55        self.oauth_consumers_client.delete_consumer(consumer['id'])56        # check that consumer no longer exists57        self.assertRaises(exceptions.NotFound,58                          self.oauth_consumers_client.show_consumer,59                          consumer['id'])60    @decorators.idempotent_id('080a9b1a-c009-47c0-9979-5305bf72e3dc')61    def test_update_consumer(self):62        """Tests the update functionality"""63        # create a new consumer to update64        consumer = self._create_consumer()65        # create new description66        new_description = data_utils.rand_name('test_update_consumer')67        # update consumer68        self.oauth_consumers_client.update_consumer(consumer['id'],69                                                    new_description)70        # check that the same consumer now has the new description71        updated_consumer = self.oauth_consumers_client.show_consumer(72            consumer['id'])['consumer']73        self.assertEqual(new_description, updated_consumer['description'])74    @decorators.idempotent_id('09ca50de-78f2-4ffb-ac71-f2254036b2b8')75    def test_list_consumers(self):76        """Test for listing consumers"""77        # create two consumers to populate list78        new_consumer_one = self._create_consumer()79        new_consumer_two = self._create_consumer()80        # fetch the list of consumers81        consumer_list = self.oauth_consumers_client \82                            .list_consumers()['consumers']83        # add fetched consumer ids to a list84        id_list = [consumer['id'] for consumer in consumer_list]85        # check if created consumers are in the list86        self.assertIn(new_consumer_one['id'], id_list)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.