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

How to use mock_aws_request_headers method in localstack

Best Python code snippet using localstack_python

apigateway_listener.py

Source:apigateway_listener.py

...53                if uri.endswith('kinesis:action/PutRecords'):54                    template = integration['requestTemplates'][APPLICATION_JSON]55                    new_request = aws_stack.render_velocity_template(template, data)56                    # forward records to target kinesis stream57                    headers = aws_stack.mock_aws_request_headers(service='kinesis')58                    headers['X-Amz-Target'] = kinesis_listener.ACTION_PUT_RECORDS59                    result = common.make_http_request(url=TEST_KINESIS_URL,60                        method='POST', data=new_request, headers=headers)61                    return result62                elif uri.startswith('arn:aws:apigateway:') and ':sqs:path' in uri:63                    template = integration['requestTemplates'][APPLICATION_JSON]64                    account_id, queue = uri.split('/')[-2:]65                    region_name = uri.split(':')[3]66                    new_request = aws_stack.render_velocity_template(template, data) + '&QueueName=%s' % queue67                    headers = aws_stack.mock_aws_request_headers(service='sqs', region_name=region_name)68                    url = urljoin(TEST_SQS_URL, '%s/%s?%s' % (account_id, queue, new_request))69                    result = common.make_http_request(url, method='GET', headers=headers)70                    return result71                else:72                    msg = 'API Gateway action uri "%s" not yet implemented' % uri73                    LOGGER.warning(msg)74                    return make_error(msg, 404)75            elif integration['type'] == 'AWS_PROXY':76                if uri.startswith('arn:aws:apigateway:') and ':lambda:path' in uri:77                    func_arn = uri.split(':lambda:path')[1].split('functions/')[1].split('/invocations')[0]78                    data_str = json.dumps(data) if isinstance(data, (dict, list)) else data79                    account_id = uri.split(':lambda:path')[1].split(':function:')[0].split(':')[-1]80                    source_ip = headers['X-Forwarded-For'].split(',')[-2]81                    # Sample request context:...

test_dynamodb.py

Source:test_dynamodb.py

...3def test_fix_region_in_headers():4    # the NoSQL Workbench sends "localhost" or "local" as the region name5    # TODO: this may need to be updated once we migrate DynamoDB to ASF6    for region_name in ["local", "localhost"]:7        headers = aws_stack.mock_aws_request_headers("dynamodb", region_name=region_name)8        assert aws_stack.get_region() not in headers.get("Authorization")9        ProxyListenerDynamoDB.prepare_request_headers(headers)...

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.