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

How to use dispatch_to_moto method in localstack

Best Python code snippet using localstack_python

moto.py

Source:moto.py

...35    :param context: the request context36    :param include_response_metadata: whether to include botocore's "ResponseMetadata" attribute37    :return: a serialized AWS ServiceResponse (same as boto3 would return)38    """39    status, headers, content = dispatch_to_moto(context)40    operation_model = context.operation41    response_dict = {  # this is what botocore.endpoint.convert_to_response_dict normally does42        "headers": dict(headers.items()),  # boto doesn't like werkzeug headers43        "status_code": status,44        "body": to_bytes(content),45        "context": {46            "operation_name": operation_model.name,47        },48    }49    parser = create_parser(context.service.protocol)50    response = parser.parse(response_dict, operation_model.output_shape)51    if status >= 301:52        error = response["Error"]53        raise CommonServiceException(54            code=error.get("Code", "UnknownError"),55            status_code=status,56            message=error.get("Message", ""),57        )58    if not include_response_metadata:59        response.pop("ResponseMetadata", None)60    return response61def call_moto_with_request(62    context: RequestContext, service_request: ServiceRequest63) -> ServiceResponse:64    """65    Like `call_moto`, but you can pass a modified version of the service request before calling moto. The caveat is66    that a new HTTP request has to be created. The service_request is serialized into a new RequestContext object,67    and headers from the old request are merged into the new one.68    :param context: the original request context69    :param service_request: the dictionary containing the service request parameters70    :return: a serialized AWS ServiceResponse (same as boto3 would return)71    """72    local_context = create_aws_request_context(73        service_name=context.service.service_name,74        action=context.operation.name,75        parameters=service_request,76        region=context.region,77    )78    local_context.request.headers.extend(context.request.headers)79    return call_moto(local_context)80def proxy_moto(context: RequestContext, service_request: ServiceRequest = None) -> HttpResponse:81    """82    Similar to ``call``, only that ``proxy`` does not parse the HTTP response into a ServiceResponse, but instead83    returns directly the HTTP response. This can be useful to pass through moto's response directly to the client.84    :param context: the request context85    :param service_request: currently not being used, added to satisfy ServiceRequestHandler contract86    :return: the HttpResponse from moto87    """88    status, headers, content = dispatch_to_moto(context)89    return HttpResponse(response=content, status=status, headers=headers)90def MotoFallbackDispatcher(provider: object) -> DispatchTable:91    """92    Wraps a provider with a moto fallthrough mechanism. It does by creating a new DispatchTable from the original93    provider, and wrapping each method with a fallthrough method that calls ``request`` if the original provider94    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:113    url_map = get_moto_routing_table(service)...

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.