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

How to use _call_moto method in localstack

Best Python code snippet using localstack_python

provider.py

Source:provider.py

...549        parameters: MapOfStringToString = None,550    ) -> RestApi:551        body_data = body.read()552        openapi_spec = parse_json_or_yaml(to_str(body_data))553        response = _call_moto(554            context,555            "CreateRestApi",556            CreateRestApiRequest(name=openapi_spec.get("info").get("title")),557        )558        return _call_moto(559            context,560            "PutRestApi",561            PutRestApiRequest(562                restApiId=response.get("id"),563                failOnWarnings=str_to_bool(fail_on_warnings) or False,564                parameters=parameters or {},565                body=body_data,566            ),567        )568    def delete_integration(569        self, context: RequestContext, rest_api_id: String, resource_id: String, http_method: String570    ) -> None:571        try:572            call_moto(context)573        except Exception as e:574            raise NotFoundException("Invalid Resource identifier specified") from e575    def get_export(576        self,577        context: RequestContext,578        rest_api_id: String,579        stage_name: String,580        export_type: String,581        parameters: MapOfStringToString = None,582        accepts: String = None,583    ) -> ExportResponse:584        openapi_exporter = OpenApiExporter()585        result = openapi_exporter.export_api(586            api_id=rest_api_id, stage=stage_name, export_type=export_type, export_format=accepts587        )588        if accepts == APPLICATION_JSON:589            result = json.dumps(result, indent=2)590        return ExportResponse(contentType=accepts, body=result)591    def get_api_keys(592        self,593        context: RequestContext,594        position: String = None,595        limit: NullableInteger = None,596        name_query: String = None,597        customer_id: String = None,598        include_values: NullableBoolean = None,599    ) -> ApiKeys:600        moto_response: ApiKeys = call_moto(context=context)601        item_list = PaginatedList(moto_response["items"])602        def token_generator(item):603            return item["id"]604        def filter_function(item):605            return item["name"].startswith(name_query)606        paginated_list, next_token = item_list.get_page(607            token_generator=token_generator,608            next_token=position,609            page_size=limit,610            filter_function=filter_function if name_query else None,611        )612        return ApiKeys(613            items=paginated_list, warnings=moto_response.get("warnings"), position=next_token614        )615# ---------------616# UTIL FUNCTIONS617# ---------------618def _call_moto(context: RequestContext, operation_name: str, parameters: ServiceRequest):619    """620    Not necessarily the pattern we want to follow in the future, but this makes possible to nest621    moto call and still be interface compatible.622    Ripped :call_moto_with_request: from moto.py but applicable to any operation (operation_name).623    """624    local_context = create_aws_request_context(625        service_name=context.service.service_name,626        action=operation_name,627        parameters=parameters,628        region=context.region,629    )630    local_context.request.headers.update(context.request.headers)631    return call_moto(local_context)632def normalize_authorizer(data):...

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.