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

How to use connect_api_gateway_to_http_with_lambda_proxy method in localstack

Best Python code snippet using localstack_python

test_api_gateway.py

Source:test_api_gateway.py

...73            }]74        })75    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,76        stage_name=TEST_STAGE_NAME)77def connect_api_gateway_to_http_with_lambda_proxy(gateway_name, target_uri, methods=[], path=None):78    if not methods:79        methods = ['GET', 'POST']80    if not path:81        path = '/'82    resources = {}83    resource_path = path.lstrip('/')84    resources[resource_path] = []85    for method in methods:86        resources[resource_path].append({87            'httpMethod': method,88            'integrations': [{89                'type': 'AWS_PROXY',90                'uri': target_uri91            }]92        })93    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,94        stage_name=TEST_STAGE_NAME)95def test_api_gateway_kinesis_integration():96    # create target Kinesis stream97    aws_stack.create_kinesis_stream(TEST_STREAM_KINESIS_API_GW)98    # create API Gateway and connect it to the target stream99    result = connect_api_gateway_to_kinesis('test_gateway1', TEST_STREAM_KINESIS_API_GW)100    # generate test data101    test_data = {'records': [102        {'data': '{"foo": "bar1"}'},103        {'data': '{"foo": "bar2"}'},104        {'data': '{"foo": "bar3"}'}105    ]}106    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],107        stage_name=TEST_STAGE_NAME, path=API_PATH_DATA_INBOUND)108    result = requests.post(url, data=json.dumps(test_data))109    result = json.loads(to_str(result.content))110    assert result['FailedRecordCount'] == 0111    assert len(result['Records']) == len(test_data['records'])112def test_api_gateway_http_integration():113    test_port = 12123114    backend_url = 'http://localhost:%s%s' % (test_port, API_PATH_HTTP_BACKEND)115    # create target HTTP backend116    class TestListener(ProxyListener):117        def forward_request(self, **kwargs):118            response = Response()119            response.status_code = 200120            response._content = kwargs.get('data') or '{}'121            return response122    proxy = GenericProxy(test_port, update_listener=TestListener())123    proxy.start()124    # create API Gateway and connect it to the HTTP backend125    result = connect_api_gateway_to_http('test_gateway2', backend_url, path=API_PATH_HTTP_BACKEND)126    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],127        stage_name=TEST_STAGE_NAME, path=API_PATH_HTTP_BACKEND)128    # make sure CORS headers are present129    origin = 'localhost'130    result = requests.options(url, headers={'origin': origin})131    assert result.status_code == 200132    assert re.match(result.headers['Access-Control-Allow-Origin'].replace('*', '.*'), origin)133    assert 'POST' in result.headers['Access-Control-Allow-Methods']134    # make test request to gateway135    result = requests.get(url)136    assert result.status_code == 200137    assert to_str(result.content) == '{}'138    data = {'data': 123}139    result = requests.post(url, data=json.dumps(data))140    assert result.status_code == 200141    assert json.loads(to_str(result.content)) == data142    # clean up143    proxy.stop()144def test_api_gateway_lambda_proxy_integration():145    # create lambda function146    zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON), get_content=True,147        libs=TEST_LAMBDA_LIBS, runtime=LAMBDA_RUNTIME_PYTHON27)148    testutil.create_lambda_function(func_name=TEST_LAMBDA_PROXY_BACKEND,149        zip_file=zip_file, runtime=LAMBDA_RUNTIME_PYTHON27)150    # create API Gateway and connect it to the Lambda proxy backend151    lambda_uri = aws_stack.lambda_function_arn(TEST_LAMBDA_PROXY_BACKEND)152    target_uri = 'arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations' % (DEFAULT_REGION, lambda_uri)153    result = connect_api_gateway_to_http_with_lambda_proxy('test_gateway2', target_uri,154        path=API_PATH_LAMBDA_PROXY_BACKEND)155    # make test request to gateway and check response156    path = API_PATH_LAMBDA_PROXY_BACKEND.replace('{test_param1}', 'foo1')157    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'], stage_name=TEST_STAGE_NAME, path=path)158    data = {'return_status_code': 203, 'return_headers': {'foo': 'bar123'}}159    result = requests.post(url, data=json.dumps(data))160    assert result.status_code == 203161    assert result.headers.get('foo') == 'bar123'162    parsed_body = json.loads(to_str(result.content))163    assert parsed_body.get('return_status_code') == 203164    assert parsed_body.get('return_headers') == {'foo': 'bar123'}165    assert parsed_body.get('pathParameters') == {'test_param1': 'foo1'}166    result = requests.delete(url, data=json.dumps(data))167    assert result.status_code == 404168def test_api_gateway_lambda_proxy_integration_any_method():169    # create lambda function170    zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON), get_content=True,171        libs=TEST_LAMBDA_LIBS, runtime=LAMBDA_RUNTIME_PYTHON27)172    testutil.create_lambda_function(func_name=TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD,173        zip_file=zip_file, runtime=LAMBDA_RUNTIME_PYTHON27)174    # create API Gateway and connect it to the Lambda proxy backend175    lambda_uri = aws_stack.lambda_function_arn(TEST_LAMBDA_PROXY_BACKEND_ANY_METHOD)176    target_uri = aws_stack.apigateway_invocations_arn(lambda_uri)177    result = connect_api_gateway_to_http_with_lambda_proxy('test_gateway3', target_uri,178        methods=['ANY'],179        path=API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD)180    # make test request to gateway and check response181    path = API_PATH_LAMBDA_PROXY_BACKEND_ANY_METHOD.replace('{test_param1}', 'foo1')182    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'], stage_name=TEST_STAGE_NAME, path=path)183    data = {}184    for method in ('GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS'):185        body = json.dumps(data) if method in ('POST', 'PUT', 'PATCH') else None186        result = getattr(requests, method.lower())(url, data=body)187        assert result.status_code == 200188        parsed_body = json.loads(to_str(result.content))...

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.