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

How to use list_of_parameters_to_object method in localstack

Best Python code snippet using localstack_python

message_forwarding.py

Source:message_forwarding.py

...145    if auth_type == AUTH_OAUTH:146        oauth_parameters = auth_parameters.get("OAuthParameters", {})147        oauth_method = oauth_parameters.get("HttpMethod")148        oauth_endpoint = oauth_parameters.get("AuthorizationEndpoint", "")149        query_object = list_of_parameters_to_object(150            oauth_parameters.get("QueryStringParameters", [])151        )152        oauth_endpoint = add_query_params_to_url(oauth_endpoint, query_object)153        client_parameters = oauth_parameters.get("ClientParameters", {})154        client_id = client_parameters.get("ClientID", "")155        client_secret = client_parameters.get("ClientSecret", "")156        oauth_body = list_of_parameters_to_object(oauth_parameters.get("BodyParameters", []))157        oauth_body.update({"client_id": client_id, "client_secret": client_secret})158        oauth_header = list_of_parameters_to_object(oauth_parameters.get("HeaderParameters", []))159        oauth_result = requests.request(160            method=oauth_method,161            url=oauth_endpoint,162            data=json.dumps(oauth_body),163            headers=oauth_header,164        )165        oauth_data = json.loads(oauth_result.text)166        token_type = oauth_data.get("token_type", "")167        access_token = oauth_data.get("access_token", "")168        auth_header = "{} {}".format(token_type, access_token)169        headers.update({"authorization": auth_header})170    return headers171def list_of_parameters_to_object(items):172    return {item.get("Key"): item.get("Value") for item in items}173def send_event_to_api_destination(target_arn, event):174    """Send an event to an EventBridge API destination175    See https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-destinations.html"""176    # ARN format: ...:api-destination/{name}/{uuid}177    region = target_arn.split(":")[3]178    api_destination_name = target_arn.split(":")[-1].split("/")[1]179    events_client = connect_to_service("events", region_name=region)180    destination = events_client.describe_api_destination(Name=api_destination_name)181    # get destination endpoint details182    method = destination.get("HttpMethod", "GET")183    endpoint = destination.get("InvocationEndpoint")184    state = destination.get("ApiDestinationState") or "ACTIVE"185    LOG.debug('Calling EventBridge API destination (state "%s"): %s %s', state, method, endpoint)186    headers = {187        # default headers AWS sends with every api destination call188        "User-Agent": "Amazon/EventBridge/ApiDestinations",189        "Content-Type": "application/json; charset=utf-8",190        "Range": "bytes=0-1048575",191        "Accept-Encoding": "gzip,deflate",192        "Connection": "close",193    }194    endpoint = add_api_destination_authorization(destination, headers, event)195    result = requests.request(196        method=method, url=endpoint, data=json.dumps(event or {}), headers=headers197    )198    if result.status_code >= 400:199        LOG.debug("Received code %s forwarding events: %s %s", result.status_code, method, endpoint)200        if result.status_code == 429 or 500 <= result.status_code <= 600:201            pass  # TODO: retry logic (only retry on 429 and 5xx response status)202def add_api_destination_authorization(destination, headers, event):203    connection_arn = destination.get("ConnectionArn", "")204    connection_name = re.search(r"connection\/([a-zA-Z0-9-_]+)\/", connection_arn).group(1)205    connection_region = extract_region_from_arn(connection_arn)206    # Using backend directly due to boto hiding passwords, keys and secret values207    event_backend = moto_events_backends.get(connection_region)208    connection = event_backend.describe_connection(name=connection_name)209    headers.update(auth_keys_from_connection(connection))210    auth_parameters = connection.get("AuthParameters", {})211    invocation_parameters = auth_parameters.get("InvocationHttpParameters")212    endpoint = destination.get("InvocationEndpoint")213    if invocation_parameters:214        header_parameters = list_of_parameters_to_object(215            invocation_parameters.get("HeaderParameters", [])216        )217        headers.update(header_parameters)218        body_parameters = list_of_parameters_to_object(219            invocation_parameters.get("BodyParameters", [])220        )221        event.update(body_parameters)222        query_parameters = invocation_parameters.get("QueryStringParameters", [])223        query_object = list_of_parameters_to_object(query_parameters)224        endpoint = add_query_params_to_url(endpoint, query_object)...

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.