KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _create_headers method in localstack

Best Python code snippet using localstack_python

mpower.py

Source:mpower.py

...6from kitiwa.settings import MPOWER_TOKEN, MPOWER_MASTER_KEY,\7    MPOWER_PRIVATE_KEY, MPOWER_ENDPOINT_OPR_TOKEN_REQUEST,\8    MPOWER_ENDPOINT_OPR_TOKEN_CHARGE, MPOWER_ENDPOINT_CHECK_INVOICE_STATUS,\9    MPOWER_RESPONSE_SUCCESS, MPOWER_STATUS_COMPLETED10def _create_headers():11    return {12        'content-type': 'application/json',13        'MP-Master-Key': MPOWER_MASTER_KEY,14        'MP-Private-Key': MPOWER_PRIVATE_KEY,15        'MP-Token': MPOWER_TOKEN16    }17def opr_token_request(amount, mpower_phone_number,18                      invoice_desc='', store_name='Kitiwa'):19    # convert to local phone number format20    mpower_phone_number = '0{}'.format(mpower_phone_number[4::])21    payload = {22        'invoice_data': {23            'invoice': {24                'total_amount': amount,25                'description': invoice_desc26            },27            'store': {28                'name': store_name29            }30        },31        'opr_data': {32            'account_alias': mpower_phone_number33        }34    }35    headers = _create_headers()36    try:37        response = requests.post(38            MPOWER_ENDPOINT_OPR_TOKEN_REQUEST,39            data=json.dumps(payload),40            headers=headers41        )42        decoded_response = response.json()43        response_code = decoded_response['response_code']44        response_text = decoded_response['response_text']45        opr_token = decoded_response['token']46        invoice_token = decoded_response['invoice_token']47    except KeyError as e:48        opr_token = invoice_token = 'N/A'49    except RequestException as e:50        message = 'ERROR - MPOWER (opr_token_request for no: {}): {}'51        log_error(message.format(mpower_phone_number, repr(e)))52        response_code = opr_token = invoice_token = 'N/A'53        response_text = repr(e)54    return response_code, response_text, opr_token, invoice_token55def opr_charge_action(opr_token, confirm_token):56    headers = _create_headers()57    payload = {58        'token': opr_token,59        'confirm_token': confirm_token60    }61    try:62        response = requests.post(63            MPOWER_ENDPOINT_OPR_TOKEN_CHARGE,64            data=json.dumps(payload),65            headers=headers66        )67        decoded_response = response.json()68        response_code = decoded_response['response_code']69        response_text = decoded_response['response_text']70    except (RequestException, KeyError) as e:71        message = 'ERROR - MPOWER (opr_charge_action for token {}): {}'72        log_error(message.format(opr_token, repr(e)))73        response_code = "N/A"74        response_text = repr(e)75    return response_code, response_text76def check_invoice_status(invoice_token):77    headers = _create_headers()78    endpoint = MPOWER_ENDPOINT_CHECK_INVOICE_STATUS.format(invoice_token)79    try:80        response = requests.get(81            endpoint,82            headers=headers83        )84        decoded_response = response.json()85        if decoded_response['response_code'] != MPOWER_RESPONSE_SUCCESS:86            message = 'ERROR - MPOWER (check_invoice_status for token {}): transaction not found'87            raise MPowerException88        if decoded_response['status'] != MPOWER_STATUS_COMPLETED:89            message = 'ERROR - MPOWER (check_invoice_status for token {}): transaction not completed'90            raise MPowerException91        return True...

twitter_api.py

Source:twitter_api.py

...14        if tweet_fields is not None:15            url += f"&tweet.fields={tweet_fields}"16        if since_id is not None:17            url += f"&since_id={since_id}"18        headers = TwitterApi._create_headers()19        response = requests.request("GET", url, headers=headers)20        if response.status_code != 200:21            raise Exception(response.status_code, response.text)22        return response.json()23    @staticmethod24    def _create_headers() -> dict:25        """26        Create authorization header with bearer token27        :return: Authorization header with bearer token28        """29        return {"Authorization": f"Bearer {twitter_config.api['bearer_token']}"}30    @staticmethod31    def get_rules() -> any:32        """33        Get rules34        :return: Rules35        """36        response = requests.get(37            "https://api.twitter.com/2/tweets/search/stream/rules", headers=TwitterApi._create_headers()38        )39        if response.status_code != 200:40            raise Exception(f"Cannot get rules (HTTP {response.status_code}): {response.text}")41        return response.json()42    @staticmethod43    def delete_rules(rules) -> any:44        """45        Delete given rules46        :param rules: Rules to be deleted47        """48        if rules is None or "data" not in rules:49            return None50        ids = list(map(lambda rule: rule["id"], rules["data"]))51        payload = {"delete": {"ids": ids}}52        response = requests.post(53            "https://api.twitter.com/2/tweets/search/stream/rules",54            headers=TwitterApi._create_headers(),55            json=payload56        )57        if response.status_code != 200:58            raise Exception(f"Cannot delete rules (HTTP {response.status_code}): {response.text}")59        return response.json()60    @staticmethod61    def set_rules(rules: list) -> any:62        """63        Set rules64        :param rules: Rules to be set65        sample_rules = [66            {"value": "dog has:images", "tag": "dog pictures"},67            {"value": "cat has:images -grumpy", "tag": "cat pictures"},68        ]69        """70        # You can adjust the rules if needed71        payload = {"add": rules}72        response = requests.post(73            "https://api.twitter.com/2/tweets/search/stream/rules",74            headers=TwitterApi._create_headers(),75            json=payload,76        )77        if response.status_code != 201:78            raise Exception(f"Cannot add rules (HTTP {response.status_code}): {response.text}")79        return response.json()80    @staticmethod81    def get_stream() -> any:82        response = requests.get(83            "https://api.twitter.com/2/tweets/search/stream", headers=TwitterApi._create_headers(), stream=True,84        )85        if response.status_code != 200:86            raise Exception(f"Cannot get stream (HTTP {response.status_code}): {response.text}")...

apiclient.py

Source:apiclient.py

...13        if query:14            params = urlencode(query)15            return self.prefix_url + endpoint + "?" + params16        return self.prefix_url + endpoint17    def _create_headers(self, extra: dict = None) -> dict:18        headers = deepcopy(self.auth_header)19        if extra:20            headers.update(extra)21        return headers22    def fetch_events(self, query: dict) -> requests.Response:23        url = self._create_url("api/v1/events/", query)24        headers = self._create_headers()25        response = requests.get(url=url, headers=headers)26        logger.info(response.text)27        return response28    def fetch_event_progresses(self, query: dict) -> requests.Response:29        url = self._create_url("api/v1/events/progresses/", query)30        headers = self._create_headers()31        response = requests.get(url=url, headers=headers)32        logger.info(response.text)33        return response34    def create_and_update_event_progresses(self, query: dict) -> requests.Response:35        url = self._create_url("api/v1/events/progresses/")36        headers = self._create_headers(37            {38                "Content-Type": "application/json",39            }40        )41        response = requests.post(url=url, headers=headers, data=json.dumps(query))42        logger.info(response.text)43        return response44    def delete_event_progresses(self, query: dict) -> requests.Response:45        url = self._create_url("api/v1/events/progresses/", query)46        headers = self._create_headers()47        response = requests.delete(url=url, headers=headers)48        logger.info(response.text)49        return response50apiClient = ApiClient(51    prefix_url=WMS_API_BASE_URL,52    auth_header={"Authorization": f"Basic {WMS_API_BASIC_AUTHORIZATION}"},...

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.