20357994067

Committed 18 Dec 2025 10:01PM UTC coverage: 86.913% (-0.02%) from 86.929%

Build # 20357994067

Build Type

push

github

Committed by

web-flow

Commit Message

Fix CloudWatch model annotation (#13545)

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1391 existing lines in 33 files now uncovered.

70000 of 80540 relevant lines covered (86.91%)

1.72 hits per line

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

61.64

/localstack-core/localstack/services/apigateway/next_gen/execute_api/integrations/http.py

import logging
from http import HTTPMethod
from typing import TypedDict

import requests
from werkzeug.datastructures import Headers

from localstack.aws.api.apigateway import Integration

from ..context import EndpointResponse, IntegrationRequest, RestApiInvocationContext
from ..gateway_response import ApiConfigurationError, IntegrationFailureError, InternalServerError
from ..header_utils import build_multi_value_headers
from .core import RestApiIntegration

LOG = logging.getLogger(__name__)

NO_BODY_METHODS = {HTTPMethod.OPTIONS, HTTPMethod.GET, HTTPMethod.HEAD}


class SimpleHttpRequest(TypedDict, total=False):
    method: HTTPMethod | str
    url: str
    params: dict[str, str | list[str]] | None
    data: bytes
    headers: dict[str, str] | None
    cookies: dict[str, str] | None
    timeout: int | None
    allow_redirects: bool | None
    stream: bool | None
    verify: bool | None
    # TODO: check if there was a situation where we'd pass certs?
    cert: str | tuple[str, str] | None


class BaseRestApiHttpIntegration(RestApiIntegration):
    @staticmethod
    def _get_integration_timeout(integration: Integration) -> float:
        return int(integration.get("timeoutInMillis", 29000)) / 1000


class RestApiHttpIntegration(BaseRestApiHttpIntegration):
    """
    This is a REST API integration responsible to send a request to another HTTP API.
    https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
    """

    name = "HTTP"

    def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:
        integration_req: IntegrationRequest = context.integration_request
        method = integration_req["http_method"]
        uri = integration_req["uri"]

        request_parameters: SimpleHttpRequest = {
            "method": method,
            "url": uri,
            "params": integration_req["query_string_parameters"],
            "headers": integration_req["headers"],
        }

        if method not in NO_BODY_METHODS:
            request_parameters["data"] = integration_req["body"]

        # TODO: configurable timeout (29 by default) (check type and default value in provider)
        # integration: Integration = context.resource_method["methodIntegration"]
        # request_parameters["timeout"] = self._get_integration_timeout(integration)
        # TODO: check for redirects
        # request_parameters["allow_redirects"] = False
        try:
            request_response = requests.request(**request_parameters)

        except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:
            LOG.warning("Execution failed due to configuration error: Invalid endpoint address")
            LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)
            raise InternalServerError("Internal server error") from e

        except (requests.exceptions.Timeout, requests.exceptions.SSLError) as e:
            # TODO make the exception catching more fine grained
            # this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
            LOG.warning("Execution failed due to a network error communicating with endpoint")
            raise IntegrationFailureError("Network error communicating with endpoint") from e

        except requests.exceptions.ConnectionError as e:
            raise ApiConfigurationError("Internal server error") from e

        return EndpointResponse(
            body=request_response.content,
            status_code=request_response.status_code,
            headers=Headers(dict(request_response.headers)),
        )


class RestApiHttpProxyIntegration(BaseRestApiHttpIntegration):
    """
    This is a simplified REST API integration responsible to send a request to another HTTP API by proxying it almost
    directly.
    https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
    """

    name = "HTTP_PROXY"

    def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:
        integration_req: IntegrationRequest = context.integration_request
        method = integration_req["http_method"]
        uri = integration_req["uri"]

        multi_value_headers = build_multi_value_headers(integration_req["headers"])
        request_headers = {key: ",".join(value) for key, value in multi_value_headers.items()}

        request_parameters: SimpleHttpRequest = {
            "method": method,
            "url": uri,
            "params": integration_req["query_string_parameters"],
            "headers": request_headers,
        }

        # TODO: validate this for HTTP_PROXY
        if method not in NO_BODY_METHODS:
            request_parameters["data"] = integration_req["body"]

        # TODO: configurable timeout (29 by default) (check type and default value in provider)
        # integration: Integration = context.resource_method["methodIntegration"]
        # request_parameters["timeout"] = self._get_integration_timeout(integration)
        try:
            request_response = requests.request(**request_parameters)

        except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:
            LOG.warning("Execution failed due to configuration error: Invalid endpoint address")
            LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)
            raise InternalServerError("Internal server error") from e

        except (requests.exceptions.Timeout, requests.exceptions.SSLError):
            # TODO make the exception catching more fine grained
            # this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
            LOG.warning("Execution failed due to a network error communicating with endpoint")
            raise IntegrationFailureError("Network error communicating with endpoint")

        except requests.exceptions.ConnectionError:
            raise ApiConfigurationError("Internal server error")

        response_headers = Headers(dict(request_response.headers))

        return EndpointResponse(
            body=request_response.content,
            status_code=request_response.status_code,
            headers=response_headers,
        )

1	import logging	2✔
2	from http import HTTPMethod	2✔
3	from typing import TypedDict	2✔
4
5	import requests	2✔
6	from werkzeug.datastructures import Headers	2✔
7
8	from localstack.aws.api.apigateway import Integration	2✔
9
10	from ..context import EndpointResponse, IntegrationRequest, RestApiInvocationContext	2✔
11	from ..gateway_response import ApiConfigurationError, IntegrationFailureError, InternalServerError	2✔
12	from ..header_utils import build_multi_value_headers	2✔
13	from .core import RestApiIntegration	2✔
14
15	LOG = logging.getLogger(__name__)	2✔
16
17	NO_BODY_METHODS = {HTTPMethod.OPTIONS, HTTPMethod.GET, HTTPMethod.HEAD}	2✔
18
19
20	class SimpleHttpRequest(TypedDict, total=False):	2✔
21	method: HTTPMethod \| str	2✔
22	url: str	2✔
23	params: dict[str, str \| list[str]] \| None	2✔
24	data: bytes	2✔
25	headers: dict[str, str] \| None	2✔
26	cookies: dict[str, str] \| None	2✔
27	timeout: int \| None	2✔
28	allow_redirects: bool \| None	2✔
29	stream: bool \| None	2✔
30	verify: bool \| None	2✔
31	# TODO: check if there was a situation where we'd pass certs?
32	cert: str \| tuple[str, str] \| None	2✔
33
34
35	class BaseRestApiHttpIntegration(RestApiIntegration):	2✔
36	@staticmethod	2✔
37	def _get_integration_timeout(integration: Integration) -> float:	2✔
38	return int(integration.get("timeoutInMillis", 29000)) / 1000	×
39
40
41	class RestApiHttpIntegration(BaseRestApiHttpIntegration):	2✔
42	"""
43	This is a REST API integration responsible to send a request to another HTTP API.
44	https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
45	"""
46
47	name = "HTTP"	2✔
48
49	def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:	2✔
50	integration_req: IntegrationRequest = context.integration_request	×
51	method = integration_req["http_method"]	×
52	uri = integration_req["uri"]	×
53
54	request_parameters: SimpleHttpRequest = {	×
55	"method": method,
56	"url": uri,
57	"params": integration_req["query_string_parameters"],
58	"headers": integration_req["headers"],
59	}
60
61	if method not in NO_BODY_METHODS:	×
62	request_parameters["data"] = integration_req["body"]	×
63
64	# TODO: configurable timeout (29 by default) (check type and default value in provider)
65	# integration: Integration = context.resource_method["methodIntegration"]
66	# request_parameters["timeout"] = self._get_integration_timeout(integration)
67	# TODO: check for redirects
68	# request_parameters["allow_redirects"] = False
69	try:	×
70	request_response = requests.request(**request_parameters)	×
71
72	except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:	×
73	LOG.warning("Execution failed due to configuration error: Invalid endpoint address")	×
74	LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)	×
75	raise InternalServerError("Internal server error") from e	×
76
77	except (requests.exceptions.Timeout, requests.exceptions.SSLError) as e:	×
78	# TODO make the exception catching more fine grained
79	# this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
80	LOG.warning("Execution failed due to a network error communicating with endpoint")	×
81	raise IntegrationFailureError("Network error communicating with endpoint") from e	×
82
83	except requests.exceptions.ConnectionError as e:	×
84	raise ApiConfigurationError("Internal server error") from e	×
85
86	return EndpointResponse(	×
87	body=request_response.content,
88	status_code=request_response.status_code,
89	headers=Headers(dict(request_response.headers)),
90	)
91
92
93	class RestApiHttpProxyIntegration(BaseRestApiHttpIntegration):	2✔
94	"""
95	This is a simplified REST API integration responsible to send a request to another HTTP API by proxying it almost
96	directly.
97	https://docs.aws.amazon.com/apigateway/latest/developerguide/setup-http-integrations.html#api-gateway-set-up-http-proxy-integration-on-proxy-resource
98	"""
99
100	name = "HTTP_PROXY"	2✔
101
102	def invoke(self, context: RestApiInvocationContext) -> EndpointResponse:	2✔
UNCOV 103	integration_req: IntegrationRequest = context.integration_request	1✔
UNCOV 104	method = integration_req["http_method"]	1✔
UNCOV 105	uri = integration_req["uri"]	1✔
106
UNCOV 107	multi_value_headers = build_multi_value_headers(integration_req["headers"])	1✔
UNCOV 108	request_headers = {key: ",".join(value) for key, value in multi_value_headers.items()}	1✔
109
UNCOV 110	request_parameters: SimpleHttpRequest = {	1✔
111	"method": method,
112	"url": uri,
113	"params": integration_req["query_string_parameters"],
114	"headers": request_headers,
115	}
116
117	# TODO: validate this for HTTP_PROXY
UNCOV 118	if method not in NO_BODY_METHODS:	1✔
UNCOV 119	request_parameters["data"] = integration_req["body"]	1✔
120
121	# TODO: configurable timeout (29 by default) (check type and default value in provider)
122	# integration: Integration = context.resource_method["methodIntegration"]
123	# request_parameters["timeout"] = self._get_integration_timeout(integration)
UNCOV 124	try:	1✔
UNCOV 125	request_response = requests.request(**request_parameters)	1✔
126
127	except (requests.exceptions.InvalidURL, requests.exceptions.InvalidSchema) as e:	×
128	LOG.warning("Execution failed due to configuration error: Invalid endpoint address")	×
129	LOG.debug("The URI specified for the HTTP/HTTP_PROXY integration is invalid: %s", uri)	×
130	raise InternalServerError("Internal server error") from e	×
131
132	except (requests.exceptions.Timeout, requests.exceptions.SSLError):	×
133	# TODO make the exception catching more fine grained
134	# this can be reproduced in AWS if you try to hit an HTTP endpoint which is HTTPS only like lambda URL
135	LOG.warning("Execution failed due to a network error communicating with endpoint")	×
136	raise IntegrationFailureError("Network error communicating with endpoint")	×
137
138	except requests.exceptions.ConnectionError:	×
139	raise ApiConfigurationError("Internal server error")	×
140
UNCOV 141	response_headers = Headers(dict(request_response.headers))	1✔
142
UNCOV 143	return EndpointResponse(	1✔
144	body=request_response.content,
145	status_code=request_response.status_code,
146	headers=response_headers,
147	)

localstack / localstack / 20357994067

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