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

How to use set_close_connection_header method in localstack

Best Python code snippet using localstack_python

app.py

Source:app.py

1from localstack.aws import handlers2from localstack.aws.handlers.metric_handler import MetricHandler3from localstack.aws.handlers.service_plugin import ServiceLoader4from localstack.services.plugins import SERVICE_PLUGINS, ServiceManager, ServicePluginManager5from .gateway import Gateway6from .handlers.fallback import EmptyResponseHandler7from .handlers.service import ServiceRequestRouter8class LocalstackAwsGateway(Gateway):9    def __init__(self, service_manager: ServiceManager = None) -> None:10        super().__init__()11        # basic server components12        self.service_manager = service_manager or ServicePluginManager()13        self.service_request_router = ServiceRequestRouter()14        # lazy-loads services into the router15        load_service = ServiceLoader(self.service_manager, self.service_request_router)16        metric_collector = MetricHandler()17        # the main request handler chain18        self.request_handlers.extend(19            [20                handlers.push_request_context,21                metric_collector.create_metric_handler_item,22                handlers.parse_service_name,  # enforce_cors and content_decoder depend on the service name23                handlers.enforce_cors,24                handlers.content_decoder,25                handlers.serve_localstack_resources,  # try to serve internal resources in /_localstack first26                handlers.serve_default_listeners,  # legacy proxy default listeners27                handlers.serve_edge_router_rules,28                # start aws handler chain29                handlers.inject_auth_header_if_missing,30                handlers.add_region_from_header,31                handlers.add_account_id,32                handlers.parse_service_request,33                metric_collector.record_parsed_request,34                handlers.serve_custom_service_request_handlers,35                load_service,  # once we have the service request we can make sure we load the service36                self.service_request_router,  # once we know the service is loaded we can route the request37                # if the chain is still running, set an empty response38                EmptyResponseHandler(404, b'{"message": "Not Found"}'),39            ]40        )41        # exception handlers in the chain42        self.exception_handlers.extend(43            [44                handlers.log_exception,45                handlers.handle_service_exception,46                handlers.handle_internal_failure,47            ]48        )49        # response post-processing50        self.response_handlers.extend(51            [52                handlers.parse_service_response,53                handlers.set_close_connection_header,54                handlers.run_custom_response_handlers,55                handlers.add_cors_response_headers,56                handlers.log_response,57                handlers.count_service_request,58                handlers.pop_request_context,59                metric_collector.update_metric_collection,60            ]61        )62def main():63    """64    Serve the LocalstackGateway with the default configuration directly through hypercorn. This is mostly for65    development purposes and documentation on how to serve the Gateway.66    """67    from .serving.hypercorn import serve68    use_ssl = True69    port = 456670    # serve the LocalStackAwsGateway in a dev app71    from localstack.utils.bootstrap import setup_logging72    setup_logging()73    if use_ssl:74        from localstack.services.generic_proxy import (75            GenericProxy,76            install_predefined_cert_if_available,77        )78        install_predefined_cert_if_available()79        _, cert_file_name, key_file_name = GenericProxy.create_ssl_cert(serial_number=port)80        ssl_creds = (cert_file_name, key_file_name)81    else:82        ssl_creds = None83    gw = LocalstackAwsGateway(SERVICE_PLUGINS)84    serve(gw, use_reloader=True, port=port, ssl_creds=ssl_creds)85if __name__ == "__main__":...

__init__.py

Source:__init__.py

1""" A set of common handlers to build an AWS server application."""2from .. import chain3from . import analytics, auth, codec, cors, fallback, internal, legacy, logging, region, service4enforce_cors = cors.CorsEnforcer()5add_cors_response_headers = cors.CorsResponseEnricher()6content_decoder = codec.ContentDecoder()7parse_service_name = service.ServiceNameParser()8parse_service_request = service.ServiceRequestParser()9add_account_id = auth.AccountIdEnricher()10inject_auth_header_if_missing = auth.MissingAuthHeaderInjector()11add_region_from_header = region.RegionContextEnricher()12log_exception = logging.ExceptionLogger()13log_response = logging.ResponseLogger()14count_service_request = analytics.ServiceRequestCounter()15handle_service_exception = service.ServiceExceptionSerializer()16handle_internal_failure = fallback.InternalFailureHandler()17serve_custom_service_request_handlers = chain.CompositeHandler()18serve_localstack_resources = internal.LocalstackResourceHandler()19run_custom_response_handlers = chain.CompositeResponseHandler()20parse_service_response = service.ServiceResponseParser()21# legacy compatibility handlers22serve_edge_router_rules = legacy.EdgeRouterHandler()23serve_default_listeners = legacy.DefaultListenerHandler()24set_close_connection_header = legacy.set_close_connection_header25pop_request_context = legacy.pop_request_context...

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.