17086927072

Committed 19 Aug 2025 10:02PM UTC coverage: 86.889% (+0.01%) from 86.875%

Build # 17086927072

Build Type

push

github

Committed by

web-flow

Commit Message

APIGW: fix TestInvokeMethod path logic (#13030)

Coverage Stats

4 of 23 new or added lines in 1 file covered. (17.39%)

264 existing lines in 17 files now uncovered.

67018 of 77131 relevant lines covered (86.89%)

0.87 hits per line

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

25.74

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

import datetime
import logging
from urllib.parse import parse_qs

from rolo import Request
from rolo.gateway.chain import HandlerChain
from werkzeug.datastructures import Headers

from localstack.aws.api.apigateway import TestInvokeMethodRequest, TestInvokeMethodResponse
from localstack.constants import APPLICATION_JSON
from localstack.http import Response
from localstack.utils.strings import to_bytes, to_str

from ...models import RestApiDeployment
from . import handlers
from .context import InvocationRequest, RestApiInvocationContext
from .handlers.resource_router import RestAPIResourceRouter
from .header_utils import build_multi_value_headers
from .template_mapping import dict_to_string
from .variables import (
    ContextVariableOverrides,
    ContextVarsRequestOverride,
    ContextVarsResponseOverride,
)

LOG = logging.getLogger(__name__)


# TODO: we probably need to write and populate those logs as part of the handler chain itself
#  and store it in the InvocationContext. That way, we could also retrieve in when calling TestInvoke

TEST_INVOKE_TEMPLATE = """Execution log for request {request_id}
{formatted_date} : Starting execution for request: {request_id}
{formatted_date} : HTTP Method: {request_method}, Resource Path: {resource_path}
{formatted_date} : Method request path: {method_request_path_parameters}
{formatted_date} : Method request query string: {method_request_query_string}
{formatted_date} : Method request headers: {method_request_headers}
{formatted_date} : Method request body before transformations: {method_request_body}
{formatted_date} : Endpoint request URI: {endpoint_uri}
{formatted_date} : Endpoint request headers: {endpoint_request_headers}
{formatted_date} : Endpoint request body after transformations: {endpoint_request_body}
{formatted_date} : Sending request to {endpoint_uri}
{formatted_date} : Received response. Status: {endpoint_response_status_code}, Integration latency: {endpoint_response_latency} ms
{formatted_date} : Endpoint response headers: {endpoint_response_headers}
{formatted_date} : Endpoint response body before transformations: {endpoint_response_body}
{formatted_date} : Method response body after transformations: {method_response_body}
{formatted_date} : Method response headers: {method_response_headers}
{formatted_date} : Successfully completed execution
{formatted_date} : Method completed with status: {method_response_status}
"""

TEST_INVOKE_TEMPLATE_MOCK = """Execution log for request {request_id}
{formatted_date} : Starting execution for request: {request_id}
{formatted_date} : HTTP Method: {request_method}, Resource Path: {resource_path}
{formatted_date} : Method request path: {method_request_path_parameters}
{formatted_date} : Method request query string: {method_request_query_string}
{formatted_date} : Method request headers: {method_request_headers}
{formatted_date} : Method request body before transformations: {method_request_body}
{formatted_date} : Method response body after transformations: {method_response_body}
{formatted_date} : Method response headers: {method_response_headers}
{formatted_date} : Successfully completed execution
{formatted_date} : Method completed with status: {method_response_status}
"""


def _dump_headers(headers: Headers) -> str:
    if not headers:
        return "{}"
    multi_headers = {key: ",".join(headers.getlist(key)) for key in headers.keys()}
    string_headers = dict_to_string(multi_headers)
    if len(string_headers) > 998:
        return f"{string_headers[:998]} [TRUNCATED]"

    return string_headers


def log_template(invocation_context: RestApiInvocationContext, response_headers: Headers) -> str:
    # TODO: funny enough, in AWS for the `endpoint_response_headers` in AWS_PROXY, they log the response headers from
    #  lambda HTTP Invoke call even though we use the headers from the lambda response itself
    formatted_date = datetime.datetime.now(tz=datetime.UTC).strftime("%a %b %d %H:%M:%S %Z %Y")
    request = invocation_context.invocation_request
    context_var = invocation_context.context_variables
    integration_req = invocation_context.integration_request
    endpoint_resp = invocation_context.endpoint_response
    method_resp = invocation_context.invocation_response
    # TODO: if endpoint_uri is an ARN, it means it's an AWS_PROXY integration
    #  this should be transformed to the true URL of a lambda invoke call
    endpoint_uri = integration_req.get("uri", "")

    return TEST_INVOKE_TEMPLATE.format(
        formatted_date=formatted_date,
        request_id=context_var["requestId"],
        resource_path=request["path"],
        request_method=request["http_method"],
        method_request_path_parameters=dict_to_string(request["path_parameters"]),
        method_request_query_string=dict_to_string(request["query_string_parameters"]),
        method_request_headers=_dump_headers(request.get("headers")),
        method_request_body=to_str(request.get("body", "")),
        endpoint_uri=endpoint_uri,
        endpoint_request_headers=_dump_headers(integration_req.get("headers")),
        endpoint_request_body=to_str(integration_req.get("body", "")),
        # TODO: measure integration latency
        endpoint_response_latency=150,
        endpoint_response_status_code=endpoint_resp.get("status_code"),
        endpoint_response_body=to_str(endpoint_resp.get("body", "")),
        endpoint_response_headers=_dump_headers(endpoint_resp.get("headers")),
        method_response_status=method_resp.get("status_code"),
        method_response_body=to_str(method_resp.get("body", "")),
        method_response_headers=_dump_headers(response_headers),
    )


def log_mock_template(
    invocation_context: RestApiInvocationContext, response_headers: Headers
) -> str:
    formatted_date = datetime.datetime.now(tz=datetime.UTC).strftime("%a %b %d %H:%M:%S %Z %Y")
    request = invocation_context.invocation_request
    context_var = invocation_context.context_variables
    method_resp = invocation_context.invocation_response

    return TEST_INVOKE_TEMPLATE_MOCK.format(
        formatted_date=formatted_date,
        request_id=context_var["requestId"],
        resource_path=request["path"],
        request_method=request["http_method"],
        method_request_path_parameters=dict_to_string(request["path_parameters"]),
        method_request_query_string=dict_to_string(request["query_string_parameters"]),
        method_request_headers=_dump_headers(request.get("headers")),
        method_request_body=to_str(request.get("body", "")),
        method_response_status=method_resp.get("status_code"),
        method_response_body=to_str(method_resp.get("body", "")),
        method_response_headers=_dump_headers(response_headers),
    )


def create_test_chain() -> HandlerChain[RestApiInvocationContext]:
    return HandlerChain(
        request_handlers=[
            handlers.method_request_handler,
            handlers.integration_request_handler,
            handlers.integration_handler,
            handlers.integration_response_handler,
            handlers.method_response_handler,
        ],
        exception_handlers=[
            handlers.gateway_exception_handler,
        ],
    )


def create_test_invocation_context(
    test_request: TestInvokeMethodRequest,
    deployment: RestApiDeployment,
) -> RestApiInvocationContext:
    parse_handler = handlers.parse_request
    http_method = test_request["httpMethod"]
    resource = deployment.rest_api.resources[test_request["resourceId"]]
    resource_path = resource["path"]

    # we do not need a true HTTP request for the context, as we are skipping all the parsing steps and using the
    # provider data
    invocation_context = RestApiInvocationContext(
        request=Request(method=http_method),
    )
    test_request_path = test_request.get("pathWithQueryString") or resource_path
    path_query = test_request_path.split("?")
    path = path_query[0]
    multi_query_args: dict[str, list[str]] = {}

    if len(path_query) > 1:
        multi_query_args = parse_qs(path_query[1])

    # for the single value parameters, AWS only keeps the last value of the list
    single_query_args = {k: v[-1] for k, v in multi_query_args.items()}

    invocation_request = InvocationRequest(
        http_method=http_method,
        path=path,
        raw_path=path,
        query_string_parameters=single_query_args,
        multi_value_query_string_parameters=multi_query_args,
        headers=Headers(test_request.get("headers")),
        # TODO: handle multiValueHeaders
        body=to_bytes(test_request.get("body") or ""),
    )

    invocation_context.invocation_request = invocation_request
    try:
        # this is AWS behavior, it will accept any value for the `pathWithQueryString`, even if it doesn't match
        # the expected format. It will just fall back to no value if it cannot parse the path parameters out of it
        _, path_parameters = RestAPIResourceRouter(deployment).match(invocation_context)
    except Exception as e:
        LOG.warning(
            "Error while trying to extract path parameters from user-provided 'pathWithQueryString=%s' "
            "for the following resource path: '%s'. Error: '%s'",
            path,
            resource_path,
            e,
        )
        path_parameters = {}

    invocation_request["path_parameters"] = path_parameters

    invocation_context.deployment = deployment
    invocation_context.api_id = test_request["restApiId"]
    invocation_context.stage = None
    invocation_context.deployment_id = ""
    invocation_context.account_id = deployment.account_id
    invocation_context.region = deployment.region
    invocation_context.stage_variables = test_request.get("stageVariables", {})
    invocation_context.context_variables = parse_handler.create_context_variables(
        invocation_context
    )
    invocation_context.context_variable_overrides = ContextVariableOverrides(
        requestOverride=ContextVarsRequestOverride(header={}, path={}, querystring={}),
        responseOverride=ContextVarsResponseOverride(header={}, status=0),
    )
    invocation_context.trace_id = parse_handler.populate_trace_id({})
    resource_method = resource["resourceMethods"][http_method]
    invocation_context.resource = resource
    invocation_context.resource_method = resource_method
    invocation_context.integration = resource_method["methodIntegration"]
    handlers.route_request.update_context_variables_with_resource(
        invocation_context.context_variables, resource
    )

    return invocation_context


def run_test_invocation(
    test_request: TestInvokeMethodRequest, deployment: RestApiDeployment
) -> TestInvokeMethodResponse:
    # validate resource exists in deployment
    invocation_context = create_test_invocation_context(test_request, deployment)

    test_chain = create_test_chain()
    is_mock_integration = invocation_context.integration["type"] == "MOCK"

    # header order is important
    if is_mock_integration:
        base_headers = {"Content-Type": APPLICATION_JSON}
    else:
        # we manually add the trace-id, as it is normally added by handlers.response_enricher which adds to much data
        # for the TestInvoke. It needs to be first
        base_headers = {
            "X-Amzn-Trace-Id": invocation_context.trace_id,
            "Content-Type": APPLICATION_JSON,
        }

    test_response = Response(headers=base_headers)
    start_time = datetime.datetime.now()
    test_chain.handle(context=invocation_context, response=test_response)
    end_time = datetime.datetime.now()

    response_headers = test_response.headers.copy()
    # AWS does not return the Content-Length for TestInvokeMethod
    response_headers.remove("Content-Length")

    if is_mock_integration:
        # TODO: revisit how we're building the logs
        log = log_mock_template(invocation_context, response_headers)
    else:
        log = log_template(invocation_context, response_headers)

    headers = dict(response_headers)
    multi_value_headers = build_multi_value_headers(response_headers)

    return TestInvokeMethodResponse(
        log=log,
        status=test_response.status_code,
        body=test_response.get_data(as_text=True),
        headers=headers,
        multiValueHeaders=multi_value_headers,
        latency=int((end_time - start_time).total_seconds()),
    )

1	import datetime	1✔
2	import logging	1✔
3	from urllib.parse import parse_qs	1✔
4
5	from rolo import Request	1✔
6	from rolo.gateway.chain import HandlerChain	1✔
7	from werkzeug.datastructures import Headers	1✔
8
9	from localstack.aws.api.apigateway import TestInvokeMethodRequest, TestInvokeMethodResponse	1✔
10	from localstack.constants import APPLICATION_JSON	1✔
11	from localstack.http import Response	1✔
12	from localstack.utils.strings import to_bytes, to_str	1✔
13
14	from ...models import RestApiDeployment	1✔
15	from . import handlers	1✔
16	from .context import InvocationRequest, RestApiInvocationContext	1✔
17	from .handlers.resource_router import RestAPIResourceRouter	1✔
18	from .header_utils import build_multi_value_headers	1✔
19	from .template_mapping import dict_to_string	1✔
20	from .variables import (	1✔
21	ContextVariableOverrides,
22	ContextVarsRequestOverride,
23	ContextVarsResponseOverride,
24	)
25
26	LOG = logging.getLogger(__name__)	1✔
27
28
29	# TODO: we probably need to write and populate those logs as part of the handler chain itself
30	# and store it in the InvocationContext. That way, we could also retrieve in when calling TestInvoke
31
32	TEST_INVOKE_TEMPLATE = """Execution log for request {request_id}	1✔
33	{formatted_date} : Starting execution for request: {request_id}
34	{formatted_date} : HTTP Method: {request_method}, Resource Path: {resource_path}
35	{formatted_date} : Method request path: {method_request_path_parameters}
36	{formatted_date} : Method request query string: {method_request_query_string}
37	{formatted_date} : Method request headers: {method_request_headers}
38	{formatted_date} : Method request body before transformations: {method_request_body}
39	{formatted_date} : Endpoint request URI: {endpoint_uri}
40	{formatted_date} : Endpoint request headers: {endpoint_request_headers}
41	{formatted_date} : Endpoint request body after transformations: {endpoint_request_body}
42	{formatted_date} : Sending request to {endpoint_uri}
43	{formatted_date} : Received response. Status: {endpoint_response_status_code}, Integration latency: {endpoint_response_latency} ms
44	{formatted_date} : Endpoint response headers: {endpoint_response_headers}
45	{formatted_date} : Endpoint response body before transformations: {endpoint_response_body}
46	{formatted_date} : Method response body after transformations: {method_response_body}
47	{formatted_date} : Method response headers: {method_response_headers}
48	{formatted_date} : Successfully completed execution
49	{formatted_date} : Method completed with status: {method_response_status}
50	"""
51
52	TEST_INVOKE_TEMPLATE_MOCK = """Execution log for request {request_id}	1✔
53	{formatted_date} : Starting execution for request: {request_id}
54	{formatted_date} : HTTP Method: {request_method}, Resource Path: {resource_path}
55	{formatted_date} : Method request path: {method_request_path_parameters}
56	{formatted_date} : Method request query string: {method_request_query_string}
57	{formatted_date} : Method request headers: {method_request_headers}
58	{formatted_date} : Method request body before transformations: {method_request_body}
59	{formatted_date} : Method response body after transformations: {method_response_body}
60	{formatted_date} : Method response headers: {method_response_headers}
61	{formatted_date} : Successfully completed execution
62	{formatted_date} : Method completed with status: {method_response_status}
63	"""
64
65
66	def _dump_headers(headers: Headers) -> str:	1✔
67	if not headers:	×
68	return "{}"	×
69	multi_headers = {key: ",".join(headers.getlist(key)) for key in headers.keys()}	×
70	string_headers = dict_to_string(multi_headers)	×
71	if len(string_headers) > 998:	×
72	return f"{string_headers[:998]} [TRUNCATED]"	×
73
74	return string_headers	×
75
76
77	def log_template(invocation_context: RestApiInvocationContext, response_headers: Headers) -> str:	1✔
78	# TODO: funny enough, in AWS for the `endpoint_response_headers` in AWS_PROXY, they log the response headers from
79	# lambda HTTP Invoke call even though we use the headers from the lambda response itself
80	formatted_date = datetime.datetime.now(tz=datetime.UTC).strftime("%a %b %d %H:%M:%S %Z %Y")	×
81	request = invocation_context.invocation_request	×
82	context_var = invocation_context.context_variables	×
83	integration_req = invocation_context.integration_request	×
84	endpoint_resp = invocation_context.endpoint_response	×
85	method_resp = invocation_context.invocation_response	×
86	# TODO: if endpoint_uri is an ARN, it means it's an AWS_PROXY integration
87	# this should be transformed to the true URL of a lambda invoke call
88	endpoint_uri = integration_req.get("uri", "")	×
89
90	return TEST_INVOKE_TEMPLATE.format(	×
91	formatted_date=formatted_date,
92	request_id=context_var["requestId"],
93	resource_path=request["path"],
94	request_method=request["http_method"],
95	method_request_path_parameters=dict_to_string(request["path_parameters"]),
96	method_request_query_string=dict_to_string(request["query_string_parameters"]),
97	method_request_headers=_dump_headers(request.get("headers")),
98	method_request_body=to_str(request.get("body", "")),
99	endpoint_uri=endpoint_uri,
100	endpoint_request_headers=_dump_headers(integration_req.get("headers")),
101	endpoint_request_body=to_str(integration_req.get("body", "")),
102	# TODO: measure integration latency
103	endpoint_response_latency=150,
104	endpoint_response_status_code=endpoint_resp.get("status_code"),
105	endpoint_response_body=to_str(endpoint_resp.get("body", "")),
106	endpoint_response_headers=_dump_headers(endpoint_resp.get("headers")),
107	method_response_status=method_resp.get("status_code"),
108	method_response_body=to_str(method_resp.get("body", "")),
109	method_response_headers=_dump_headers(response_headers),
110	)
111
112
113	def log_mock_template(	1✔
114	invocation_context: RestApiInvocationContext, response_headers: Headers
115	) -> str:
NEW 116	formatted_date = datetime.datetime.now(tz=datetime.UTC).strftime("%a %b %d %H:%M:%S %Z %Y")	×
NEW 117	request = invocation_context.invocation_request	×
NEW 118	context_var = invocation_context.context_variables	×
NEW 119	method_resp = invocation_context.invocation_response	×
120
NEW 121	return TEST_INVOKE_TEMPLATE_MOCK.format(	×
122	formatted_date=formatted_date,
123	request_id=context_var["requestId"],
124	resource_path=request["path"],
125	request_method=request["http_method"],
126	method_request_path_parameters=dict_to_string(request["path_parameters"]),
127	method_request_query_string=dict_to_string(request["query_string_parameters"]),
128	method_request_headers=_dump_headers(request.get("headers")),
129	method_request_body=to_str(request.get("body", "")),
130	method_response_status=method_resp.get("status_code"),
131	method_response_body=to_str(method_resp.get("body", "")),
132	method_response_headers=_dump_headers(response_headers),
133	)
134
135
136	def create_test_chain() -> HandlerChain[RestApiInvocationContext]:	1✔
137	return HandlerChain(	×
138	request_handlers=[
139	handlers.method_request_handler,
140	handlers.integration_request_handler,
141	handlers.integration_handler,
142	handlers.integration_response_handler,
143	handlers.method_response_handler,
144	],
145	exception_handlers=[
146	handlers.gateway_exception_handler,
147	],
148	)
149
150
151	def create_test_invocation_context(	1✔
152	test_request: TestInvokeMethodRequest,
153	deployment: RestApiDeployment,
154	) -> RestApiInvocationContext:
155	parse_handler = handlers.parse_request	×
156	http_method = test_request["httpMethod"]	×
NEW 157	resource = deployment.rest_api.resources[test_request["resourceId"]]	×
NEW 158	resource_path = resource["path"]	×
159
160	# we do not need a true HTTP request for the context, as we are skipping all the parsing steps and using the
161	# provider data
162	invocation_context = RestApiInvocationContext(	×
163	request=Request(method=http_method),
164	)
NEW 165	test_request_path = test_request.get("pathWithQueryString") or resource_path	×
NEW 166	path_query = test_request_path.split("?")	×
167	path = path_query[0]	×
168	multi_query_args: dict[str, list[str]] = {}	×
169
170	if len(path_query) > 1:	×
171	multi_query_args = parse_qs(path_query[1])	×
172
173	# for the single value parameters, AWS only keeps the last value of the list
174	single_query_args = {k: v[-1] for k, v in multi_query_args.items()}	×
175
176	invocation_request = InvocationRequest(	×
177	http_method=http_method,
178	path=path,
179	raw_path=path,
180	query_string_parameters=single_query_args,
181	multi_value_query_string_parameters=multi_query_args,
182	headers=Headers(test_request.get("headers")),
183	# TODO: handle multiValueHeaders
184	body=to_bytes(test_request.get("body") or ""),
185	)
186
UNCOV 187	invocation_context.invocation_request = invocation_request	×
NEW 188	try:	×
189	# this is AWS behavior, it will accept any value for the `pathWithQueryString`, even if it doesn't match
190	# the expected format. It will just fall back to no value if it cannot parse the path parameters out of it
NEW 191	_, path_parameters = RestAPIResourceRouter(deployment).match(invocation_context)	×
NEW 192	except Exception as e:	×
NEW 193	LOG.warning(	×
194	"Error while trying to extract path parameters from user-provided 'pathWithQueryString=%s' "
195	"for the following resource path: '%s'. Error: '%s'",
196	path,
197	resource_path,
198	e,
199	)
NEW 200	path_parameters = {}	×
201
202	invocation_request["path_parameters"] = path_parameters	×
203
204	invocation_context.deployment = deployment	×
205	invocation_context.api_id = test_request["restApiId"]	×
206	invocation_context.stage = None	×
207	invocation_context.deployment_id = ""	×
208	invocation_context.account_id = deployment.account_id	×
209	invocation_context.region = deployment.region	×
210	invocation_context.stage_variables = test_request.get("stageVariables", {})	×
211	invocation_context.context_variables = parse_handler.create_context_variables(	×
212	invocation_context
213	)
214	invocation_context.context_variable_overrides = ContextVariableOverrides(	×
215	requestOverride=ContextVarsRequestOverride(header={}, path={}, querystring={}),
216	responseOverride=ContextVarsResponseOverride(header={}, status=0),
217	)
218	invocation_context.trace_id = parse_handler.populate_trace_id({})	×
219	resource_method = resource["resourceMethods"][http_method]	×
220	invocation_context.resource = resource	×
221	invocation_context.resource_method = resource_method	×
222	invocation_context.integration = resource_method["methodIntegration"]	×
223	handlers.route_request.update_context_variables_with_resource(	×
224	invocation_context.context_variables, resource
225	)
226
227	return invocation_context	×
228
229
230	def run_test_invocation(	1✔
231	test_request: TestInvokeMethodRequest, deployment: RestApiDeployment
232	) -> TestInvokeMethodResponse:
233	# validate resource exists in deployment
234	invocation_context = create_test_invocation_context(test_request, deployment)	×
235
236	test_chain = create_test_chain()	×
NEW 237	is_mock_integration = invocation_context.integration["type"] == "MOCK"	×
238
239	# header order is important
NEW 240	if is_mock_integration:	×
241	base_headers = {"Content-Type": APPLICATION_JSON}	×
242	else:
243	# we manually add the trace-id, as it is normally added by handlers.response_enricher which adds to much data
244	# for the TestInvoke. It needs to be first
245	base_headers = {	×
246	"X-Amzn-Trace-Id": invocation_context.trace_id,
247	"Content-Type": APPLICATION_JSON,
248	}
249
250	test_response = Response(headers=base_headers)	×
251	start_time = datetime.datetime.now()	×
252	test_chain.handle(context=invocation_context, response=test_response)	×
253	end_time = datetime.datetime.now()	×
254
255	response_headers = test_response.headers.copy()	×
256	# AWS does not return the Content-Length for TestInvokeMethod
257	response_headers.remove("Content-Length")	×
258
NEW 259	if is_mock_integration:	×
260	# TODO: revisit how we're building the logs
NEW 261	log = log_mock_template(invocation_context, response_headers)	×
262	else:
NEW 263	log = log_template(invocation_context, response_headers)	×
264
265	headers = dict(response_headers)	×
266	multi_value_headers = build_multi_value_headers(response_headers)	×
267
268	return TestInvokeMethodResponse(	×
269	log=log,
270	status=test_response.status_code,
271	body=test_response.get_data(as_text=True),
272	headers=headers,
273	multiValueHeaders=multi_value_headers,
274	latency=int((end_time - start_time).total_seconds()),
275	)

localstack / localstack / 17086927072

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