20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

80.95

/localstack-core/localstack/services/apigateway/next_gen/execute_api/helpers.py

import copy
import logging
import random
import re
import time
from secrets import token_hex
from typing import TypedDict

from moto.apigateway.models import RestAPI as MotoRestAPI

from localstack.services.apigateway.models import MergedRestApi, RestApiContainer, RestApiDeployment
from localstack.utils.aws.arns import get_partition

from .context import RestApiInvocationContext
from .moto_helpers import get_resources_from_moto_rest_api

LOG = logging.getLogger(__name__)

_stage_variable_pattern = re.compile(r"\${stageVariables\.(?P<varName>.*?)}")


def freeze_rest_api(
    account_id: str, region: str, moto_rest_api: MotoRestAPI, localstack_rest_api: RestApiContainer
) -> RestApiDeployment:
    """
    Snapshot a REST API in time to create a deployment
    This will merge the Moto and LocalStack data into one `MergedRestApi`
    """
    moto_resources = get_resources_from_moto_rest_api(moto_rest_api)

    rest_api = MergedRestApi.from_rest_api_container(
        rest_api_container=localstack_rest_api,
        resources=moto_resources,
    )

    return RestApiDeployment(
        account_id=account_id,
        region=region,
        rest_api=copy.deepcopy(rest_api),
    )


def render_uri_with_stage_variables(
    uri: str | None, stage_variables: dict[str, str] | None
) -> str | None:
    """
    https://docs.aws.amazon.com/apigateway/latest/developerguide/aws-api-gateway-stage-variables-reference.html#stage-variables-in-integration-HTTP-uris
    URI=https://${stageVariables.<variable_name>}
    This format is the same as VTL, but we're using a simplified version to only replace `${stageVariables.<param>}`
    values, as AWS will ignore `${path}` for example
    """
    if not uri:
        return uri
    stage_vars = stage_variables or {}

    def replace_match(match_obj: re.Match) -> str:
        return stage_vars.get(match_obj.group("varName"), "")

    return _stage_variable_pattern.sub(replace_match, uri)


def render_uri_with_path_parameters(uri: str | None, path_parameters: dict[str, str]) -> str | None:
    if not uri:
        return uri

    for key, value in path_parameters.items():
        uri = uri.replace(f"{{{key}}}", value)

    return uri


def render_integration_uri(
    uri: str | None, path_parameters: dict[str, str], stage_variables: dict[str, str]
) -> str:
    """
    A URI can contain different value to interpolate / render
    It will have path parameters substitutions with this shape (can also add a querystring).
    URI=http://myhost.test/rootpath/{path}

    It can also have another format, for stage variables, documented here:
    https://docs.aws.amazon.com/apigateway/latest/developerguide/aws-api-gateway-stage-variables-reference.html#stage-variables-in-integration-HTTP-uris
    URI=https://${stageVariables.<variable_name>}
    This format is the same as VTL.

    :param uri: the integration URI
    :param path_parameters: the list of path parameters, coming from the parameters mapping and override
    :param stage_variables: -
    :return: the rendered URI
    """
    if not uri:
        return ""

    uri_with_path = render_uri_with_path_parameters(uri, path_parameters)
    return render_uri_with_stage_variables(uri_with_path, stage_variables)


def get_source_arn(context: RestApiInvocationContext):
    method = context.resource_method["httpMethod"]
    path = context.resource["path"]
    return (
        f"arn:{get_partition(context.region)}:execute-api"
        f":{context.region}"
        f":{context.account_id}"
        f":{context.api_id}"
        f"/{context.stage}/{method}{path}"
    )


def get_lambda_function_arn_from_invocation_uri(uri: str) -> str:
    """
    "arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/arn:aws:lambda:us-east-1:123456789012:function:SimpleLambda4ProxyResource/invocations",
    :param uri: the integration URI value for a lambda function
    :return: the lambda function ARN
    """
    return uri.split("functions/")[1].removesuffix("/invocations")


def validate_sub_dict_of_typed_dict(typed_dict: type[TypedDict], obj: dict) -> bool:
    """
    Validate that the object is a subset off the keys of a given `TypedDict`.
    :param typed_dict: the `TypedDict` blueprint
    :param obj: the object to validate
    :return: True if it is a subset, False otherwise
    """
    typed_dict_keys = {*typed_dict.__required_keys__, *typed_dict.__optional_keys__}

    return not bool(set(obj) - typed_dict_keys)


def generate_trace_id():
    """https://docs.aws.amazon.com/xray/latest/devguide/xray-api-sendingdata.html#xray-api-traceids"""
    original_request_epoch = int(time.time())
    timestamp_hex = hex(original_request_epoch)[2:]
    version_number = "1"
    unique_id = token_hex(12)
    return f"{version_number}-{timestamp_hex}-{unique_id}"


def generate_trace_parent():
    return token_hex(8)


def parse_trace_id(trace_id: str) -> dict[str, str]:
    split_trace = trace_id.split(";")
    trace_values = {}
    for trace_part in split_trace:
        key_value = trace_part.split("=")
        if len(key_value) == 2:
            trace_values[key_value[0].capitalize()] = key_value[1]

    return trace_values


def mime_type_matches_binary_media_types(mime_type: str | None, binary_media_types: list[str]):
    if not mime_type or not binary_media_types:
        return False

    mime_type_and_subtype = mime_type.split(",")[0].split(";")[0].split("/")
    if len(mime_type_and_subtype) != 2:
        return False
    mime_type, mime_subtype = mime_type_and_subtype

    for bmt in binary_media_types:
        type_and_subtype = bmt.split(";")[0].split("/")
        if len(type_and_subtype) != 2:
            continue
        _type, subtype = type_and_subtype
        if _type == "*":
            continue

        if subtype == "*" and mime_type == _type:
            return True

        if mime_type == _type and mime_subtype == subtype:
            return True

    return False


def should_divert_to_canary(percent_traffic: float) -> bool:
    if int(percent_traffic) == 100:
        return True
    return percent_traffic > random.random() * 100

1	import copy	1✔
2	import logging	1✔
3	import random	1✔
4	import re	1✔
5	import time	1✔
6	from secrets import token_hex	1✔
7	from typing import TypedDict	1✔
8
9	from moto.apigateway.models import RestAPI as MotoRestAPI	1✔
10
11	from localstack.services.apigateway.models import MergedRestApi, RestApiContainer, RestApiDeployment	1✔
12	from localstack.utils.aws.arns import get_partition	1✔
13
14	from .context import RestApiInvocationContext	1✔
15	from .moto_helpers import get_resources_from_moto_rest_api	1✔
16
17	LOG = logging.getLogger(__name__)	1✔
18
19	_stage_variable_pattern = re.compile(r"\${stageVariables\.(?P<varName>.*?)}")	1✔
20
21
22	def freeze_rest_api(	1✔
23	account_id: str, region: str, moto_rest_api: MotoRestAPI, localstack_rest_api: RestApiContainer
24	) -> RestApiDeployment:
25	"""
26	Snapshot a REST API in time to create a deployment
27	This will merge the Moto and LocalStack data into one `MergedRestApi`
28	"""
29	moto_resources = get_resources_from_moto_rest_api(moto_rest_api)	1✔
30
31	rest_api = MergedRestApi.from_rest_api_container(	1✔
32	rest_api_container=localstack_rest_api,
33	resources=moto_resources,
34	)
35
36	return RestApiDeployment(	1✔
37	account_id=account_id,
38	region=region,
39	rest_api=copy.deepcopy(rest_api),
40	)
41
42
43	def render_uri_with_stage_variables(	1✔
44	uri: str \| None, stage_variables: dict[str, str] \| None
45	) -> str \| None:
46	"""
47	https://docs.aws.amazon.com/apigateway/latest/developerguide/aws-api-gateway-stage-variables-reference.html#stage-variables-in-integration-HTTP-uris
48	URI=https://${stageVariables.<variable_name>}
49	This format is the same as VTL, but we're using a simplified version to only replace `${stageVariables.<param>}`
50	values, as AWS will ignore `${path}` for example
51	"""
52	if not uri:	1✔
53	return uri	×
54	stage_vars = stage_variables or {}	1✔
55
56	def replace_match(match_obj: re.Match) -> str:	1✔
57	return stage_vars.get(match_obj.group("varName"), "")	1✔
58
59	return _stage_variable_pattern.sub(replace_match, uri)	1✔
60
61
62	def render_uri_with_path_parameters(uri: str \| None, path_parameters: dict[str, str]) -> str \| None:	1✔
63	if not uri:	1✔
64	return uri	×
65
66	for key, value in path_parameters.items():	1✔
67	uri = uri.replace(f"{{{key}}}", value)	1✔
68
69	return uri	1✔
70
71
72	def render_integration_uri(	1✔
73	uri: str \| None, path_parameters: dict[str, str], stage_variables: dict[str, str]
74	) -> str:
75	"""
76	A URI can contain different value to interpolate / render
77	It will have path parameters substitutions with this shape (can also add a querystring).
78	URI=http://myhost.test/rootpath/{path}
79
80	It can also have another format, for stage variables, documented here:
81	https://docs.aws.amazon.com/apigateway/latest/developerguide/aws-api-gateway-stage-variables-reference.html#stage-variables-in-integration-HTTP-uris
82	URI=https://${stageVariables.<variable_name>}
83	This format is the same as VTL.
84
85	:param uri: the integration URI
86	:param path_parameters: the list of path parameters, coming from the parameters mapping and override
87	:param stage_variables: -
88	:return: the rendered URI
89	"""
90	if not uri:	1✔
91	return ""	×
92
93	uri_with_path = render_uri_with_path_parameters(uri, path_parameters)	1✔
94	return render_uri_with_stage_variables(uri_with_path, stage_variables)	1✔
95
96
97	def get_source_arn(context: RestApiInvocationContext):	1✔
98	method = context.resource_method["httpMethod"]	×
99	path = context.resource["path"]	×
100	return (	×
101	f"arn:{get_partition(context.region)}:execute-api"
102	f":{context.region}"
103	f":{context.account_id}"
104	f":{context.api_id}"
105	f"/{context.stage}/{method}{path}"
106	)
107
108
109	def get_lambda_function_arn_from_invocation_uri(uri: str) -> str:	1✔
110	"""
111	"arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/arn:aws:lambda:us-east-1:123456789012:function:SimpleLambda4ProxyResource/invocations",
112	:param uri: the integration URI value for a lambda function
113	:return: the lambda function ARN
114	"""
115	return uri.split("functions/")[1].removesuffix("/invocations")	×
116
117
118	def validate_sub_dict_of_typed_dict(typed_dict: type[TypedDict], obj: dict) -> bool:	1✔
119	"""
120	Validate that the object is a subset off the keys of a given `TypedDict`.
121	:param typed_dict: the `TypedDict` blueprint
122	:param obj: the object to validate
123	:return: True if it is a subset, False otherwise
124	"""
125	typed_dict_keys = {typed_dict.__required_keys__, typed_dict.__optional_keys__}	×
126
127	return not bool(set(obj) - typed_dict_keys)	×
128
129
130	def generate_trace_id():	1✔
131	"""https://docs.aws.amazon.com/xray/latest/devguide/xray-api-sendingdata.html#xray-api-traceids"""
132	original_request_epoch = int(time.time())	1✔
133	timestamp_hex = hex(original_request_epoch)[2:]	1✔
134	version_number = "1"	1✔
135	unique_id = token_hex(12)	1✔
136	return f"{version_number}-{timestamp_hex}-{unique_id}"	1✔
137
138
139	def generate_trace_parent():	1✔
140	return token_hex(8)	1✔
141
142
143	def parse_trace_id(trace_id: str) -> dict[str, str]:	1✔
144	split_trace = trace_id.split(";")	1✔
145	trace_values = {}	1✔
146	for trace_part in split_trace:	1✔
147	key_value = trace_part.split("=")	1✔
148	if len(key_value) == 2:	1✔
149	trace_values[key_value[0].capitalize()] = key_value[1]	1✔
150
151	return trace_values	1✔
152
153
154	def mime_type_matches_binary_media_types(mime_type: str \| None, binary_media_types: list[str]):	1✔
155	if not mime_type or not binary_media_types:	1✔
156	return False	1✔
157
158	mime_type_and_subtype = mime_type.split(",")[0].split(";")[0].split("/")	1✔
159	if len(mime_type_and_subtype) != 2:	1✔
160	return False	×
161	mime_type, mime_subtype = mime_type_and_subtype	1✔
162
163	for bmt in binary_media_types:	1✔
164	type_and_subtype = bmt.split(";")[0].split("/")	1✔
165	if len(type_and_subtype) != 2:	1✔
166	continue	×
167	_type, subtype = type_and_subtype	1✔
168	if _type == "*":	1✔
169	continue	×
170
171	if subtype == "*" and mime_type == _type:	1✔
172	return True	×
173
174	if mime_type == _type and mime_subtype == subtype:	1✔
175	return True	1✔
176
177	return False	1✔
178
179
180	def should_divert_to_canary(percent_traffic: float) -> bool:	1✔
181	if int(percent_traffic) == 100:	×
182	return True	×
183	return percent_traffic > random.random() * 100	×

localstack / localstack / 20565403496

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