16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.35

/localstack-core/localstack/utils/aws/message_forwarding.py

import base64
import json
import logging
import re
import uuid
from typing import Optional

from moto.events.models import events_backends

from localstack.aws.connect import connect_to
from localstack.services.apigateway.legacy.helpers import extract_query_string_params
from localstack.utils import collections
from localstack.utils.aws.arns import (
    extract_account_id_from_arn,
    extract_region_from_arn,
    firehose_name,
    sqs_queue_url_for_arn,
)
from localstack.utils.http import add_path_parameters_to_url, add_query_params_to_url
from localstack.utils.http import safe_requests as requests
from localstack.utils.strings import to_bytes, to_str
from localstack.utils.time import now_utc

LOG = logging.getLogger(__name__)

AUTH_BASIC = "BASIC"
AUTH_API_KEY = "API_KEY"
AUTH_OAUTH = "OAUTH_CLIENT_CREDENTIALS"


# TODO: refactor/split this. too much here is service specific
def send_event_to_target(
    target_arn: str,
    event: dict,
    target_attributes: dict = None,
    asynchronous: bool = True,
    target: dict = None,
    role: str = None,
    source_arn: str = None,
    source_service: str = None,
    events_source: str = None,  # optional data for publishing to EventBridge
    events_detail_type: str = None,  # optional data for publishing to EventBridge
):
    region = extract_region_from_arn(target_arn)
    account_id = extract_account_id_from_arn(source_arn)

    if target is None:
        target = {}
    if role:
        clients = connect_to.with_assumed_role(
            role_arn=role, service_principal=source_service, region_name=region
        )
    else:
        clients = connect_to(aws_access_key_id=account_id, region_name=region)

    if ":lambda:" in target_arn:
        lambda_client = clients.lambda_.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )
        lambda_client.invoke(
            FunctionName=target_arn,
            Payload=to_bytes(json.dumps(event)),
            InvocationType="Event" if asynchronous else "RequestResponse",
        )

    elif ":sns:" in target_arn:
        sns_client = clients.sns.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )
        sns_client.publish(TopicArn=target_arn, Message=json.dumps(event))

    elif ":sqs:" in target_arn:
        sqs_client = clients.sqs.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )
        queue_url = sqs_queue_url_for_arn(target_arn)
        msg_group_id = collections.get_safe(target_attributes, "$.SqsParameters.MessageGroupId")
        kwargs = {"MessageGroupId": msg_group_id} if msg_group_id else {}
        sqs_client.send_message(
            QueueUrl=queue_url, MessageBody=json.dumps(event, separators=(",", ":")), **kwargs
        )

    elif ":states:" in target_arn:
        account_id = extract_account_id_from_arn(target_arn)
        stepfunctions_client = connect_to(
            aws_access_key_id=account_id, region_name=region
        ).stepfunctions
        stepfunctions_client.start_execution(stateMachineArn=target_arn, input=json.dumps(event))

    elif ":firehose:" in target_arn:
        delivery_stream_name = firehose_name(target_arn)
        firehose_client = clients.firehose.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )
        firehose_client.put_record(
            DeliveryStreamName=delivery_stream_name,
            Record={"Data": to_bytes(json.dumps(event))},
        )

    elif ":events:" in target_arn:
        if ":api-destination/" in target_arn or ":destination/" in target_arn:
            send_event_to_api_destination(target_arn, event, target.get("HttpParameters"))

        else:
            events_client = clients.events.request_metadata(
                service_principal=source_service, source_arn=source_arn
            )
            eventbus_name = target_arn.split(":")[-1].split("/")[-1]
            detail = event.get("detail") or event
            resources = event.get("resources") or [source_arn] if source_arn else []
            events_client.put_events(
                Entries=[
                    {
                        "EventBusName": eventbus_name,
                        "Source": events_source or event.get("source", source_service) or "",
                        "DetailType": events_detail_type or event.get("detail-type", ""),
                        "Detail": json.dumps(detail),
                        "Resources": resources,
                    }
                ]
            )

    elif ":kinesis:" in target_arn:
        partition_key_path = collections.get_safe(
            target_attributes,
            "$.KinesisParameters.PartitionKeyPath",
            default_value="$.id",
        )

        stream_name = target_arn.split("/")[-1]
        partition_key = collections.get_safe(event, partition_key_path, event["id"])
        kinesis_client = clients.kinesis.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )

        kinesis_client.put_record(
            StreamName=stream_name,
            Data=to_bytes(json.dumps(event)),
            PartitionKey=partition_key,
        )

    elif ":logs:" in target_arn:
        log_group_name = target_arn.split(":")[6]
        logs_client = clients.logs.request_metadata(
            service_principal=source_service, source_arn=source_arn
        )
        log_stream_name = str(uuid.uuid4())
        logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
        logs_client.put_log_events(
            logGroupName=log_group_name,
            logStreamName=log_stream_name,
            logEvents=[{"timestamp": now_utc(millis=True), "message": json.dumps(event)}],
        )
    else:
        LOG.warning('Unsupported Events rule target ARN: "%s"', target_arn)


def auth_keys_from_connection(connection: dict):
    headers = {}

    auth_type = connection.get("AuthorizationType").upper()
    auth_parameters = connection.get("AuthParameters")
    if auth_type == AUTH_BASIC:
        basic_auth_parameters = auth_parameters.get("BasicAuthParameters", {})
        username = basic_auth_parameters.get("Username", "")
        password = basic_auth_parameters.get("Password", "")
        auth = "Basic " + to_str(base64.b64encode(f"{username}:{password}".encode("ascii")))
        headers.update({"authorization": auth})

    if auth_type == AUTH_API_KEY:
        api_key_parameters = auth_parameters.get("ApiKeyAuthParameters", {})
        api_key_name = api_key_parameters.get("ApiKeyName", "")
        api_key_value = api_key_parameters.get("ApiKeyValue", "")
        headers.update({api_key_name: api_key_value})

    if auth_type == AUTH_OAUTH:
        oauth_parameters = auth_parameters.get("OAuthParameters", {})
        oauth_method = oauth_parameters.get("HttpMethod")

        oauth_http_parameters = oauth_parameters.get("OAuthHttpParameters", {})
        oauth_endpoint = oauth_parameters.get("AuthorizationEndpoint", "")
        query_object = list_of_parameters_to_object(
            oauth_http_parameters.get("QueryStringParameters", [])
        )
        oauth_endpoint = add_query_params_to_url(oauth_endpoint, query_object)

        client_parameters = oauth_parameters.get("ClientParameters", {})
        client_id = client_parameters.get("ClientID", "")
        client_secret = client_parameters.get("ClientSecret", "")

        oauth_body = list_of_parameters_to_object(oauth_http_parameters.get("BodyParameters", []))
        oauth_body.update({"client_id": client_id, "client_secret": client_secret})

        oauth_header = list_of_parameters_to_object(
            oauth_http_parameters.get("HeaderParameters", [])
        )
        oauth_result = requests.request(
            method=oauth_method,
            url=oauth_endpoint,
            data=json.dumps(oauth_body),
            headers=oauth_header,
        )
        oauth_data = json.loads(oauth_result.text)

        token_type = oauth_data.get("token_type", "")
        access_token = oauth_data.get("access_token", "")
        auth_header = f"{token_type} {access_token}"
        headers.update({"authorization": auth_header})

    return headers


def list_of_parameters_to_object(items):
    return {item.get("Key"): item.get("Value") for item in items}


def send_event_to_api_destination(target_arn, event, http_parameters: Optional[dict] = None):
    """Send an event to an EventBridge API destination
    See https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-destinations.html"""

    # ARN format: ...:api-destination/{name}/{uuid}
    account_id = extract_account_id_from_arn(target_arn)
    region = extract_region_from_arn(target_arn)

    api_destination_name = target_arn.split(":")[-1].split("/")[1]
    events_client = connect_to(aws_access_key_id=account_id, region_name=region).events
    destination = events_client.describe_api_destination(Name=api_destination_name)

    # get destination endpoint details
    method = destination.get("HttpMethod", "GET")
    endpoint = destination.get("InvocationEndpoint")
    state = destination.get("ApiDestinationState") or "ACTIVE"

    LOG.debug('Calling EventBridge API destination (state "%s"): %s %s', state, method, endpoint)
    headers = {
        # default headers AWS sends with every api destination call
        "User-Agent": "Amazon/EventBridge/ApiDestinations",
        "Content-Type": "application/json; charset=utf-8",
        "Range": "bytes=0-1048575",
        "Accept-Encoding": "gzip,deflate",
        "Connection": "close",
    }

    endpoint = add_api_destination_authorization(destination, headers, event)
    if http_parameters:
        endpoint = add_target_http_parameters(http_parameters, endpoint, headers, event)

    result = requests.request(
        method=method, url=endpoint, data=json.dumps(event or {}), headers=headers
    )
    if result.status_code >= 400:
        LOG.debug("Received code %s forwarding events: %s %s", result.status_code, method, endpoint)
        if result.status_code == 429 or 500 <= result.status_code <= 600:
            pass  # TODO: retry logic (only retry on 429 and 5xx response status)


def add_api_destination_authorization(destination, headers, event):
    connection_arn = destination.get("ConnectionArn", "")
    connection_name = re.search(r"connection\/([a-zA-Z0-9-_]+)\/", connection_arn).group(1)

    account_id = extract_account_id_from_arn(connection_arn)
    region = extract_region_from_arn(connection_arn)

    # Using backend directly due to boto hiding passwords, keys and secret values
    event_backend = events_backends[account_id][region]
    connection = event_backend.describe_connection(name=connection_name)

    headers.update(auth_keys_from_connection(connection))

    auth_parameters = connection.get("AuthParameters", {})
    invocation_parameters = auth_parameters.get("InvocationHttpParameters")

    endpoint = destination.get("InvocationEndpoint")
    if invocation_parameters:
        header_parameters = list_of_parameters_to_object(
            invocation_parameters.get("HeaderParameters", [])
        )
        headers.update(header_parameters)

        body_parameters = list_of_parameters_to_object(
            invocation_parameters.get("BodyParameters", [])
        )
        event.update(body_parameters)

        query_parameters = invocation_parameters.get("QueryStringParameters", [])
        query_object = list_of_parameters_to_object(query_parameters)
        endpoint = add_query_params_to_url(endpoint, query_object)

    return endpoint


def add_target_http_parameters(http_parameters: dict, endpoint: str, headers: dict, body):
    endpoint = add_path_parameters_to_url(endpoint, http_parameters.get("PathParameterValues", []))

    # The request should prioritze connection header/query parameters over target params if there is an overlap
    query_params = http_parameters.get("QueryStringParameters", {})
    prev_query_params = extract_query_string_params(endpoint)[1]
    query_params.update(prev_query_params)
    endpoint = add_query_params_to_url(endpoint, query_params)

    target_headers = http_parameters.get("HeaderParameters", {})
    for target_header in target_headers:
        if target_header not in headers:
            headers.update({target_header: target_headers.get(target_header)})

    return endpoint

1	import base64	1✔
2	import json	1✔
3	import logging	1✔
4	import re	1✔
5	import uuid	1✔
6	from typing import Optional	1✔
7
8	from moto.events.models import events_backends	1✔
9
10	from localstack.aws.connect import connect_to	1✔
11	from localstack.services.apigateway.legacy.helpers import extract_query_string_params	1✔
12	from localstack.utils import collections	1✔
13	from localstack.utils.aws.arns import (	1✔
14	extract_account_id_from_arn,
15	extract_region_from_arn,
16	firehose_name,
17	sqs_queue_url_for_arn,
18	)
19	from localstack.utils.http import add_path_parameters_to_url, add_query_params_to_url	1✔
20	from localstack.utils.http import safe_requests as requests	1✔
21	from localstack.utils.strings import to_bytes, to_str	1✔
22	from localstack.utils.time import now_utc	1✔
23
24	LOG = logging.getLogger(__name__)	1✔
25
26	AUTH_BASIC = "BASIC"	1✔
27	AUTH_API_KEY = "API_KEY"	1✔
28	AUTH_OAUTH = "OAUTH_CLIENT_CREDENTIALS"	1✔
29
30
31	# TODO: refactor/split this. too much here is service specific
32	def send_event_to_target(	1✔
33	target_arn: str,
34	event: dict,
35	target_attributes: dict = None,
36	asynchronous: bool = True,
37	target: dict = None,
38	role: str = None,
39	source_arn: str = None,
40	source_service: str = None,
41	events_source: str = None, # optional data for publishing to EventBridge
42	events_detail_type: str = None, # optional data for publishing to EventBridge
43	):
44	region = extract_region_from_arn(target_arn)	1✔
45	account_id = extract_account_id_from_arn(source_arn)	1✔
46
47	if target is None:	1✔
48	target = {}	1✔
49	if role:	1✔
50	clients = connect_to.with_assumed_role(	1✔
51	role_arn=role, service_principal=source_service, region_name=region
52	)
53	else:
54	clients = connect_to(aws_access_key_id=account_id, region_name=region)	×
55
56	if ":lambda:" in target_arn:	1✔
57	lambda_client = clients.lambda_.request_metadata(	×
58	service_principal=source_service, source_arn=source_arn
59	)
60	lambda_client.invoke(	×
61	FunctionName=target_arn,
62	Payload=to_bytes(json.dumps(event)),
63	InvocationType="Event" if asynchronous else "RequestResponse",
64	)
65
66	elif ":sns:" in target_arn:	1✔
67	sns_client = clients.sns.request_metadata(	1✔
68	service_principal=source_service, source_arn=source_arn
69	)
70	sns_client.publish(TopicArn=target_arn, Message=json.dumps(event))	1✔
71
72	elif ":sqs:" in target_arn:	1✔
73	sqs_client = clients.sqs.request_metadata(	1✔
74	service_principal=source_service, source_arn=source_arn
75	)
76	queue_url = sqs_queue_url_for_arn(target_arn)	1✔
77	msg_group_id = collections.get_safe(target_attributes, "$.SqsParameters.MessageGroupId")	1✔
78	kwargs = {"MessageGroupId": msg_group_id} if msg_group_id else {}	1✔
79	sqs_client.send_message(	1✔
80	QueueUrl=queue_url, MessageBody=json.dumps(event, separators=(",", ":")), **kwargs
81	)
82
83	elif ":states:" in target_arn:	1✔
84	account_id = extract_account_id_from_arn(target_arn)	×
85	stepfunctions_client = connect_to(	×
86	aws_access_key_id=account_id, region_name=region
87	).stepfunctions
88	stepfunctions_client.start_execution(stateMachineArn=target_arn, input=json.dumps(event))	×
89
90	elif ":firehose:" in target_arn:	1✔
91	delivery_stream_name = firehose_name(target_arn)	×
92	firehose_client = clients.firehose.request_metadata(	×
93	service_principal=source_service, source_arn=source_arn
94	)
95	firehose_client.put_record(	×
96	DeliveryStreamName=delivery_stream_name,
97	Record={"Data": to_bytes(json.dumps(event))},
98	)
99
100	elif ":events:" in target_arn:	1✔
101	if ":api-destination/" in target_arn or ":destination/" in target_arn:	1✔
102	send_event_to_api_destination(target_arn, event, target.get("HttpParameters"))	×
103
104	else:
105	events_client = clients.events.request_metadata(	1✔
106	service_principal=source_service, source_arn=source_arn
107	)
108	eventbus_name = target_arn.split(":")[-1].split("/")[-1]	1✔
109	detail = event.get("detail") or event	1✔
110	resources = event.get("resources") or [source_arn] if source_arn else []	1✔
111	events_client.put_events(	1✔
112	Entries=[
113	{
114	"EventBusName": eventbus_name,
115	"Source": events_source or event.get("source", source_service) or "",
116	"DetailType": events_detail_type or event.get("detail-type", ""),
117	"Detail": json.dumps(detail),
118	"Resources": resources,
119	}
120	]
121	)
122
123	elif ":kinesis:" in target_arn:	×
124	partition_key_path = collections.get_safe(	×
125	target_attributes,
126	"$.KinesisParameters.PartitionKeyPath",
127	default_value="$.id",
128	)
129
130	stream_name = target_arn.split("/")[-1]	×
131	partition_key = collections.get_safe(event, partition_key_path, event["id"])	×
132	kinesis_client = clients.kinesis.request_metadata(	×
133	service_principal=source_service, source_arn=source_arn
134	)
135
136	kinesis_client.put_record(	×
137	StreamName=stream_name,
138	Data=to_bytes(json.dumps(event)),
139	PartitionKey=partition_key,
140	)
141
142	elif ":logs:" in target_arn:	×
143	log_group_name = target_arn.split(":")[6]	×
144	logs_client = clients.logs.request_metadata(	×
145	service_principal=source_service, source_arn=source_arn
146	)
147	log_stream_name = str(uuid.uuid4())	×
148	logs_client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)	×
149	logs_client.put_log_events(	×
150	logGroupName=log_group_name,
151	logStreamName=log_stream_name,
152	logEvents=[{"timestamp": now_utc(millis=True), "message": json.dumps(event)}],
153	)
154	else:
155	LOG.warning('Unsupported Events rule target ARN: "%s"', target_arn)	×
156
157
158	def auth_keys_from_connection(connection: dict):	1✔
159	headers = {}	×
160
161	auth_type = connection.get("AuthorizationType").upper()	×
162	auth_parameters = connection.get("AuthParameters")	×
163	if auth_type == AUTH_BASIC:	×
164	basic_auth_parameters = auth_parameters.get("BasicAuthParameters", {})	×
165	username = basic_auth_parameters.get("Username", "")	×
166	password = basic_auth_parameters.get("Password", "")	×
167	auth = "Basic " + to_str(base64.b64encode(f"{username}:{password}".encode("ascii")))	×
UNCOV 168	headers.update({"authorization": auth})	×
169
170	if auth_type == AUTH_API_KEY:	×
UNCOV 171	api_key_parameters = auth_parameters.get("ApiKeyAuthParameters", {})	×
172	api_key_name = api_key_parameters.get("ApiKeyName", "")	×
173	api_key_value = api_key_parameters.get("ApiKeyValue", "")	×
174	headers.update({api_key_name: api_key_value})	×
175
176	if auth_type == AUTH_OAUTH:	×
UNCOV 177	oauth_parameters = auth_parameters.get("OAuthParameters", {})	×
178	oauth_method = oauth_parameters.get("HttpMethod")	×
179
180	oauth_http_parameters = oauth_parameters.get("OAuthHttpParameters", {})	×
UNCOV 181	oauth_endpoint = oauth_parameters.get("AuthorizationEndpoint", "")	×
182	query_object = list_of_parameters_to_object(	×
183	oauth_http_parameters.get("QueryStringParameters", [])
184	)
UNCOV 185	oauth_endpoint = add_query_params_to_url(oauth_endpoint, query_object)	×
186
187	client_parameters = oauth_parameters.get("ClientParameters", {})	×
UNCOV 188	client_id = client_parameters.get("ClientID", "")	×
189	client_secret = client_parameters.get("ClientSecret", "")	×
190
191	oauth_body = list_of_parameters_to_object(oauth_http_parameters.get("BodyParameters", []))	×
UNCOV 192	oauth_body.update({"client_id": client_id, "client_secret": client_secret})	×
193
194	oauth_header = list_of_parameters_to_object(	×
195	oauth_http_parameters.get("HeaderParameters", [])
196	)
UNCOV 197	oauth_result = requests.request(	×
198	method=oauth_method,
199	url=oauth_endpoint,
200	data=json.dumps(oauth_body),
201	headers=oauth_header,
202	)
UNCOV 203	oauth_data = json.loads(oauth_result.text)	×
204
205	token_type = oauth_data.get("token_type", "")	×
UNCOV 206	access_token = oauth_data.get("access_token", "")	×
207	auth_header = f"{token_type} {access_token}"	×
208	headers.update({"authorization": auth_header})	×
209
210	return headers	×
211
212
213	def list_of_parameters_to_object(items):	1✔
214	return {item.get("Key"): item.get("Value") for item in items}	1✔
215
216
217	def send_event_to_api_destination(target_arn, event, http_parameters: Optional[dict] = None):	1✔
218	"""Send an event to an EventBridge API destination
219	See https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-destinations.html"""
220
221	# ARN format: ...:api-destination/{name}/{uuid}
UNCOV 222	account_id = extract_account_id_from_arn(target_arn)	×
UNCOV 223	region = extract_region_from_arn(target_arn)	×
224
225	api_destination_name = target_arn.split(":")[-1].split("/")[1]	×
UNCOV 226	events_client = connect_to(aws_access_key_id=account_id, region_name=region).events	×
227	destination = events_client.describe_api_destination(Name=api_destination_name)	×
228
229	# get destination endpoint details
UNCOV 230	method = destination.get("HttpMethod", "GET")	×
UNCOV 231	endpoint = destination.get("InvocationEndpoint")	×
232	state = destination.get("ApiDestinationState") or "ACTIVE"	×
233
234	LOG.debug('Calling EventBridge API destination (state "%s"): %s %s', state, method, endpoint)	×
UNCOV 235	headers = {	×
236	# default headers AWS sends with every api destination call
237	"User-Agent": "Amazon/EventBridge/ApiDestinations",
238	"Content-Type": "application/json; charset=utf-8",
239	"Range": "bytes=0-1048575",
240	"Accept-Encoding": "gzip,deflate",
241	"Connection": "close",
242	}
243
UNCOV 244	endpoint = add_api_destination_authorization(destination, headers, event)	×
UNCOV 245	if http_parameters:	×
246	endpoint = add_target_http_parameters(http_parameters, endpoint, headers, event)	×
247
248	result = requests.request(	×
249	method=method, url=endpoint, data=json.dumps(event or {}), headers=headers
250	)
UNCOV 251	if result.status_code >= 400:	×
UNCOV 252	LOG.debug("Received code %s forwarding events: %s %s", result.status_code, method, endpoint)	×
253	if result.status_code == 429 or 500 <= result.status_code <= 600:	×
254	pass # TODO: retry logic (only retry on 429 and 5xx response status)	×
255
256
257	def add_api_destination_authorization(destination, headers, event):	1✔
UNCOV 258	connection_arn = destination.get("ConnectionArn", "")	×
UNCOV 259	connection_name = re.search(r"connection\/([a-zA-Z0-9-_]+)\/", connection_arn).group(1)	×
260
261	account_id = extract_account_id_from_arn(connection_arn)	×
UNCOV 262	region = extract_region_from_arn(connection_arn)	×
263
264	# Using backend directly due to boto hiding passwords, keys and secret values
UNCOV 265	event_backend = events_backends[account_id][region]	×
UNCOV 266	connection = event_backend.describe_connection(name=connection_name)	×
267
268	headers.update(auth_keys_from_connection(connection))	×
269
270	auth_parameters = connection.get("AuthParameters", {})	×
UNCOV 271	invocation_parameters = auth_parameters.get("InvocationHttpParameters")	×
272
273	endpoint = destination.get("InvocationEndpoint")	×
UNCOV 274	if invocation_parameters:	×
275	header_parameters = list_of_parameters_to_object(	×
276	invocation_parameters.get("HeaderParameters", [])
277	)
UNCOV 278	headers.update(header_parameters)	×
279
280	body_parameters = list_of_parameters_to_object(	×
281	invocation_parameters.get("BodyParameters", [])
282	)
UNCOV 283	event.update(body_parameters)	×
284
285	query_parameters = invocation_parameters.get("QueryStringParameters", [])	×
UNCOV 286	query_object = list_of_parameters_to_object(query_parameters)	×
287	endpoint = add_query_params_to_url(endpoint, query_object)	×
288
289	return endpoint	×
290
291
292	def add_target_http_parameters(http_parameters: dict, endpoint: str, headers: dict, body):	1✔
293	endpoint = add_path_parameters_to_url(endpoint, http_parameters.get("PathParameterValues", []))	1✔
294
295	# The request should prioritze connection header/query parameters over target params if there is an overlap
296	query_params = http_parameters.get("QueryStringParameters", {})	1✔
297	prev_query_params = extract_query_string_params(endpoint)[1]	1✔
298	query_params.update(prev_query_params)	1✔
299	endpoint = add_query_params_to_url(endpoint, query_params)	1✔
300
301	target_headers = http_parameters.get("HeaderParameters", {})	1✔
302	for target_header in target_headers:	1✔
303	if target_header not in headers:	1✔
304	headers.update({target_header: target_headers.get(target_header)})	1✔
305
306	return endpoint	1✔

localstack / localstack / 16820655284

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous