Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use ForwardingFallbackDispatcher method in localstack

Best Python code snippet using localstack_python

moto.py

Source:moto.py

...94    raises a ``NotImplementedError``.95    :param provider: the ASF provider96    :return: a modified DispatchTable97    """98    return ForwardingFallbackDispatcher(provider, proxy_moto)99def dispatch_to_moto(context: RequestContext) -> MotoResponse:100    """101    Internal method to dispatch the request to moto without changing moto's dispatcher output.102    :param context: the request context103    :return: the response from moto104    """105    service = context.service106    request = context.request107    # hack to avoid call to request.form (see moto's BaseResponse.dispatch)108    request.body = request.data109    # this is where we skip the HTTP roundtrip between the moto server and the boto client110    dispatch = get_dispatcher(service.service_name, request.path)111    return dispatch(request, request.url, request.headers)112def get_dispatcher(service: str, path: str) -> MotoDispatcher:...

forwarder.py

Source:forwarder.py

...20from localstack.http import Response21from localstack.http.proxy import forward22from localstack.utils.aws import aws_stack23from localstack.utils.strings import to_str24def ForwardingFallbackDispatcher(25    provider: object, request_forwarder: ServiceRequestHandler26) -> DispatchTable:27    """28    Wraps a provider with a request forwarder. It does by creating a new DispatchTable from the original29    provider, and wrapping each method with a fallthrough method that calls ``request_forwarder`` if the30    original provider raises a ``NotImplementedError``.31    :param provider: the ASF provider32    :param request_forwarder: callable that forwards the request (e.g., to a backend server)33    :return: a modified DispatchTable34    """35    table = create_dispatch_table(provider)36    for op, fn in table.items():37        table[op] = _wrap_with_fallthrough(fn, request_forwarder)38    return table39def _wrap_with_fallthrough(40    handler: ServiceRequestHandler, fallthrough_handler: ServiceRequestHandler41) -> ServiceRequestHandler:42    def _call(context, req) -> ServiceResponse:43        try:44            # handler will typically be an ASF provider method, and in case it hasn't been45            # implemented, we try to fall back to forwarding the request to the backend46            return handler(context, req)47        except NotImplementedError:48            pass49        return fallthrough_handler(context, req)50    return _call51def HttpFallbackDispatcher(provider: object, forward_url_getter: Callable[[], str]):52    return ForwardingFallbackDispatcher(provider, get_request_forwarder_http(forward_url_getter))53def get_request_forwarder_http(forward_url_getter: Callable[[], str]) -> ServiceRequestHandler:54    def _forward_request(context, service_request: ServiceRequest = None) -> ServiceResponse:55        if service_request is not None:56            local_context = create_aws_request_context(57                service_name=context.service.service_name,58                action=context.operation.name,59                parameters=service_request,60                region=context.region,61            )62            local_context.request.headers.update(context.request.headers)63            context = local_context64        return forward_request(context, forward_url_getter)65    return _forward_request66def forward_request(...

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.