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

How to use connect_api_gateway_to_http method in localstack

Best Python code snippet using localstack_python

34116_test_api_gateway.py

Source:34116_test_api_gateway.py

...52        }]53    }]54    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,55        stage_name=TEST_STAGE_NAME)56def connect_api_gateway_to_http(gateway_name, target_url, methods=[], path=None):57    if not methods:58        methods = ['GET', 'POST']59    if not path:60        path = '/'61    resources = {}62    resource_path = path.replace('/', '')63    resources[resource_path] = []64    for method in methods:65        resources[resource_path].append({66            'httpMethod': method,67            'integrations': [{68                'type': 'HTTP',69                'uri': target_url70            }]71        })72    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,73        stage_name=TEST_STAGE_NAME)74def connect_api_gateway_to_http_with_lambda_proxy(gateway_name, target_uri, methods=[], path=None):75    if not methods:76        methods = ['GET', 'POST']77    if not path:78        path = '/'79    resources = {}80    resource_path = path.lstrip('/')81    resources[resource_path] = []82    for method in methods:83        resources[resource_path].append({84            'httpMethod': method,85            'integrations': [{86                'type': 'AWS_PROXY',87                'uri': target_uri88            }]89        })90    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,91        stage_name=TEST_STAGE_NAME)92def test_api_gateway_kinesis_integration():93    # create target Kinesis stream94    aws_stack.create_kinesis_stream(TEST_STREAM_KINESIS_API_GW)95    # create API Gateway and connect it to the target stream96    result = connect_api_gateway_to_kinesis('test_gateway1', TEST_STREAM_KINESIS_API_GW)97    # generate test data98    test_data = {'records': [99        {'data': '{"foo": "bar1"}'},100        {'data': '{"foo": "bar2"}'},101        {'data': '{"foo": "bar3"}'}102    ]}103    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],104        stage_name=TEST_STAGE_NAME, path=API_PATH_DATA_INBOUND)105    result = requests.post(url, data=json.dumps(test_data))106    result = json.loads(to_str(result.content))107    assert result['FailedRecordCount'] == 0108    assert len(result['Records']) == len(test_data['records'])109def test_api_gateway_http_integration():110    test_port = 12123111    backend_url = 'http://localhost:%s%s' % (test_port, API_PATH_HTTP_BACKEND)112    # create target HTTP backend113    class TestListener(ProxyListener):114        def forward_request(self, **kwargs):115            response = Response()116            response.status_code = 200117            response._content = json.dumps(kwargs['data']) if kwargs['data'] else '{}'118            return response119    proxy = GenericProxy(test_port, update_listener=TestListener())120    proxy.start()121    # create API Gateway and connect it to the HTTP backend122    result = connect_api_gateway_to_http('test_gateway2', backend_url, path=API_PATH_HTTP_BACKEND)123    # make test request to gateway124    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],125        stage_name=TEST_STAGE_NAME, path=API_PATH_HTTP_BACKEND)126    result = requests.get(url)127    assert result.status_code == 200128    assert to_str(result.content) == '{}'129    data = {'data': 123}130    result = requests.post(url, data=json.dumps(data))131    assert result.status_code == 200132    assert json.loads(to_str(result.content)) == data133    # clean up134    proxy.stop()135def test_api_gateway_lambda_proxy_integration():136    # create lambda function...

test_api_gateway.py

Source:test_api_gateway.py

...46        }]47    }]48    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,49        stage_name=TEST_STAGE_NAME)50def connect_api_gateway_to_http(gateway_name, target_url, methods=[], path=None):51    if not methods:52        methods = ['GET', 'POST']53    if not path:54        path = '/'55    resources = {}56    resource_data_inbound = path.replace('/', '')57    resources[resource_data_inbound] = []58    for method in methods:59        resources[resource_data_inbound].append({60            'httpMethod': method,61            'integrations': [{62                'type': 'HTTP',63                'uri': target_url64            }]65        })66    return aws_stack.create_api_gateway(name=gateway_name, resources=resources,67        stage_name=TEST_STAGE_NAME)68def test_api_gateway_kinesis_integration():69    # create target Kinesis stream70    aws_stack.create_kinesis_stream(TEST_STREAM_KINESIS_API_GW)71    # create API Gateway and connect it to the target stream72    result = connect_api_gateway_to_kinesis('test_gateway1', TEST_STREAM_KINESIS_API_GW)73    # generate test data74    test_data = {'records': [75        {'data': '{"foo": "bar1"}'},76        {'data': '{"foo": "bar2"}'},77        {'data': '{"foo": "bar3"}'}78    ]}79    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],80        stage_name=TEST_STAGE_NAME, path=API_PATH_DATA_INBOUND)81    result = requests.post(url, data=json.dumps(test_data))82    result = json.loads(to_str(result.content))83    assert result['FailedRecordCount'] == 084    assert len(result['Records']) == len(test_data['records'])85def test_api_gateway_http_integration():86    test_port = 1212387    backend_url = 'http://localhost:%s%s' % (test_port, API_PATH_HTTP_BACKEND)88    # create target HTTP backend89    def listener(**kwargs):90        response = Response()91        response.status_code = 20092        response._content = json.dumps(kwargs['data']) if kwargs['data'] else '{}'93        return response94    proxy = GenericProxy(test_port, update_listener=listener)95    proxy.start()96    # create API Gateway and connect it to the HTTP backend97    result = connect_api_gateway_to_http('test_gateway2', backend_url, path=API_PATH_HTTP_BACKEND)98    # make test request to gateway99    url = INBOUND_GATEWAY_URL_PATTERN.format(api_id=result['id'],100        stage_name=TEST_STAGE_NAME, path=API_PATH_HTTP_BACKEND)101    result = requests.get(url)102    assert result.status_code == 200103    assert to_str(result.content) == '{}'104    data = {"data": 123}105    result = requests.post(url, data=json.dumps(data))106    assert result.status_code == 200107    assert json.loads(to_str(result.content)) == data108    # clean up...

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.