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

How to use create_request_token method in tempest

Best Python code snippet using tempest_python

dropbox_storage.py

Source:dropbox_storage.py

...118        """ Return Request Token. If not available, a new one will be created, saved119            and a RequestUrl object will be returned.120        """121        if not self._request_token:122            return self.create_request_token(sess)123        return self._request_token124    def create_request_token(self, sess):125        """ Return Request Token. If not available, a new one will be created, saved126            and a RequestUrl object will be returned.127        """128        self._request_token = sess.obtain_request_token()129        self.save_token_file()130        return self._request_token131    def prompt_for_authorization(self, sess, request_token):132        """ Generate the authorization url, show it to the user and exit """133        message = "Dropbox not authorized, visit the following URL to authorize:\n"134        message += sess.build_authorize_url(request_token)135        raise StorageError(message)136    def get_access_token(self, sess):137        """ Return Access Token. If not available, a new one will be created and saved. """138        if not self._access_token:139            return self.create_access_token(sess)140        return self._access_token141    def create_access_token(self, sess):142        """ Create and save a new access token to self.TOKENFILEPATH. """143        request_token = self.get_request_token(sess)144        try:145            self._access_token = sess.obtain_access_token(request_token)146        except ErrorResponse:147            # If we get an error, it means the request token has expired or is not authorize, generate a new request148            # token and prompt the user to complete the authorization process149            request_token = self.create_request_token(sess)150            self.prompt_for_authorization(sess, request_token)151        # We've got a good access token, save it.152        self.save_token_file()153        return self._access_token154    def save_token_file(self):155        """ Save the request and access tokens to disk. """156        tokendata = dict(request_token=self._request_token, access_token=self._access_token)157        with open(self.TOKENS_FILEPATH, 'wb') as tokenhandle:158            pickle.dump(tokendata, tokenhandle, -1)159    def read_token_file(self):160        """ Reload the request and/or access tokens from disk. """161        if os.path.exists(self.TOKENS_FILEPATH):162            with open(self.TOKENS_FILEPATH, 'rb') as tokenhandle:163                tokendata = pickle.load(tokenhandle)...

test_token.py

Source:test_token.py

...26@pytest.mark.ok27@pytest.mark.it('Successfully request access_token')28def test_token_ok(client):29    # Get request token30    token = create_request_token(client_id=client_config['id'], satellite_id=satellite_config['id'], key=client_config['key'], crt=client_config['crt'])31    t_auth_params = auth_params32    t_auth_params['client_assertion'] = token33    # Invoke request34    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)35    36    # Status code37    assert response.status_code == 20038    # Response attributes39    assert response.json['scope'] == 'iSHARE'40    assert response.json['token_type'] == 'Bearer'41    42    # Access token exists43    assert 'access_token' in response.json44    # Get header45    token_header = decode_header(response.json['access_token'])46    # Verify token with provided x5c header and get decoded payload47    assert 'x5c' in token_header48    access_token = {}49    try:50        access_token = verify_token(response.json['access_token'], token_header['x5c'][0], alg="RS256", aud=satellite_config['id'])51    except Exception as ex:52        pytest.fail('Error when verifying and decoding returned access token --> Exception {}: {}'.format(type(ex).__name__, ex))53    54    # Access token parameters55    assert access_token['client_id'] == client_config['id']56    assert access_token['iss'] == satellite_config['id']57    assert access_token['aud'] == satellite_config['id']58    assert access_token['scope'] == ["iSHARE"]59    # Valid expiration claim60    now = int(str(time.time()).split('.')[0])61    assert access_token['nbf'] <= now62    assert access_token['exp'] > now63    64@pytest.mark.failure65@pytest.mark.it('Failure: Request access token with invalid client')66def test_token_invalid_client(client):67    # Get config68    cnf = load_config("tests/config/client_invalid_fiware.yml", app)69    70    # Get request token71    token = create_request_token(client_id=cnf['id'], satellite_id=satellite_config['id'], key=cnf['key'], crt=cnf['crt'])72    t_auth_params = auth_params73    t_auth_params['client_assertion'] = token74    t_auth_params['client_id'] = cnf['id']75    # Invoke request76    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)77    78    # Status code79    assert response.status_code == 40080@pytest.mark.failure81@pytest.mark.it('Failure: Request access token with client_id not matching signed request JWT issuer')82def test_token_client_id_unequals_iss(client):83    # Get config84    cnf = load_config("tests/config/client_invalid_fiware.yml", app)85    86    # Get request token87    token = create_request_token(client_id=cnf['id'], satellite_id=satellite_config['id'], key=cnf['key'], crt=cnf['crt'])88    t_auth_params = auth_params89    t_auth_params['client_assertion'] = token90    91    # Invoke request92    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)93    94    # Status code95    assert response.status_code == 40096@pytest.mark.failure97@pytest.mark.it('Failure: Request access token with invalid client and certificate from invalid root CA')98def test_token_invalid_root_ca(client):99    # Get config100    cnf = load_config("tests/config/client_invalid_ca_fiware.yml", app)101    102    # Get request token103    token = create_request_token(client_id=cnf['id'], satellite_id=satellite_config['id'], key=cnf['key'], crt=cnf['crt'])104    t_auth_params = auth_params105    t_auth_params['client_assertion'] = token106    t_auth_params['client_id'] = cnf['id']107    # Invoke request108    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)109    110    # Status code111    assert response.status_code == 400112@pytest.mark.failure113@pytest.mark.it('Failure: Request access token with valid client but cert chain replaced with invalid intermediate+root CA')114def test_token_valid_client_invalid_cert_chain(client):115    # Get configs116    cnf = load_config("tests/config/client_fiware.yml", app)117    cnf_invalid = load_config("tests/config/client_invalid_ca_fiware.yml", app)118    # Build new cert chain119    crt = cnf['client_crt']120    crt = '{}\n{}\n{}'.format(crt, cnf_invalid['intermediate'], cnf_invalid['rootca'])121        122    # Get request token123    token = create_request_token(client_id=cnf['id'], satellite_id=satellite_config['id'], key=cnf['key'], crt=crt)124    t_auth_params = auth_params125    t_auth_params['client_assertion'] = token126    t_auth_params['client_id'] = cnf['id']127    # Invoke request128    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)129    130    # Status code131    assert response.status_code == 400132@pytest.mark.failure133@pytest.mark.it('Failure: Request access token with valid client but cert chain replaced with invalid intermediate')134def test_token_valid_client_invalid_intermediate(client):135    # Get configs136    cnf = load_config("tests/config/client_fiware.yml", app)137    cnf_invalid = load_config("tests/config/client_invalid_ca_fiware.yml", app)138    # Build new cert chain139    crt = cnf['client_crt']140    crt = '{}\n{}\n{}'.format(crt, cnf_invalid['intermediate'], cnf['rootca'])141        142    # Get request token143    token = create_request_token(client_id=cnf['id'], satellite_id=satellite_config['id'], key=cnf['key'], crt=crt)144    t_auth_params = auth_params145    t_auth_params['client_assertion'] = token146    t_auth_params['client_id'] = cnf['id']147    # Invoke request148    response = client.post(TOKEN_ENDPOINT, data=t_auth_params)149    150    # Status code...

auth.py

Source:auth.py

...11        super().__init__()12        self.username = username13        self.password = password14        self.expires_at = None15        self.request_token = self._create_request_token()16        self._authorise_request_token_with_login()17        self._create_session()18    def _create_request_token(self):19        """20        Create a temporary request token that can be used to validate a TMDb user login.21        """22        resp = self._call(self._urls["create_request_token"], "")23        self.expires_at = resp["expires_at"]24        return resp["request_token"]25    def _create_session(self):26        """27        You can use this method to create a fully valid session ID once a user has validated the request token.28        """29        data = {"request_token": self.request_token}30        req = self._call(31            action=self._urls["create_session"],32            append_to_response="",...

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.