21697093787

Committed 04 Feb 2026 09:56PM UTC coverage: 86.962% (-0.004%) from 86.966%

Build # 21697093787

Build Type

push

github

Committed by

web-flow

Commit Message

improve system information sent in session and container_info (#13680)

Run Details

10 of 17 new or added lines in 2 files covered. (58.82%)

222 existing lines in 17 files now uncovered.

70560 of 81139 relevant lines covered (86.96%)

0.87 hits per line

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

58.12

/localstack-core/localstack/testing/aws/util.py

import functools
import os
from collections.abc import Callable

import boto3
import botocore
from botocore.awsrequest import AWSPreparedRequest, AWSResponse
from botocore.client import BaseClient
from botocore.compat import HTTPHeaders
from botocore.config import Config
from botocore.exceptions import ClientError

from localstack import config
from localstack.aws.api import RequestContext
from localstack.aws.connect import (
    ClientFactory,
    ExternalAwsClientFactory,
    ExternalClientFactory,
    ServiceLevelClientFactory,
)
from localstack.aws.forwarder import create_http_request
from localstack.aws.protocol.parser import create_parser
from localstack.aws.spec import LOCALSTACK_BUILTIN_DATA_PATH, load_service
from localstack.config import is_env_true
from localstack.testing.config import (
    SECONDARY_TEST_AWS_ACCESS_KEY_ID,
    SECONDARY_TEST_AWS_PROFILE,
    SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
    SECONDARY_TEST_AWS_SESSION_TOKEN,
    TEST_AWS_ACCESS_KEY_ID,
    TEST_AWS_ENDPOINT_URL,
    TEST_AWS_REGION_NAME,
    TEST_AWS_SECRET_ACCESS_KEY,
)
from localstack.utils.aws.arns import get_partition
from localstack.utils.aws.request_context import get_account_id_from_request
from localstack.utils.sync import poll_condition


def is_aws_cloud() -> bool:
    return os.environ.get("TEST_TARGET", "") == "AWS_CLOUD"


def in_default_partition() -> bool:
    return is_aws_cloud() or get_partition(TEST_AWS_REGION_NAME) == "aws"


def get_lambda_logs(func_name, logs_client):
    log_group_name = f"/aws/lambda/{func_name}"
    streams = logs_client.describe_log_streams(logGroupName=log_group_name)["logStreams"]
    streams = sorted(streams, key=lambda x: x["creationTime"], reverse=True)
    log_events = logs_client.get_log_events(
        logGroupName=log_group_name, logStreamName=streams[0]["logStreamName"]
    )["events"]
    return log_events


def bucket_exists(client, bucket_name: str) -> bool:
    buckets = client.list_buckets()
    for bucket in buckets["Buckets"]:
        if bucket["Name"] == bucket_name:
            return True
    return False


def wait_for_user(keys, region_name: str):
    sts_client = create_client_with_keys(service="sts", keys=keys, region_name=region_name)

    def is_user_ready():
        try:
            sts_client.get_caller_identity()
            return True
        except ClientError as e:
            if e.response["Error"]["Code"] == "InvalidClientTokenId":
                return False
            return True

    # wait until the given user is ready, takes AWS IAM a while...
    poll_condition(is_user_ready, interval=5, timeout=20)


def create_client_with_keys(
    service: str,
    keys: dict[str, str],
    region_name: str,
    client_config: Config = None,
):
    """
    Create a boto client with the given access key, targeted against LS per default, but to AWS if TEST_TARGET is set
    accordingly.

    :param service: Service to create the Client for
    :param keys: Access Keys
    :param region_name: Region for the client
    :param client_config:
    :return:
    """
    return boto3.client(
        service,
        region_name=region_name,
        aws_access_key_id=keys["AccessKeyId"],
        aws_secret_access_key=keys["SecretAccessKey"],
        aws_session_token=keys.get("SessionToken"),
        config=client_config,
        endpoint_url=config.internal_service_url() if not is_aws_cloud() else None,
    )


def create_request_context(
    service_name: str, operation_name: str, region: str, aws_request: AWSPreparedRequest
) -> RequestContext:
    if hasattr(aws_request.body, "read"):
        aws_request.body = aws_request.body.read()
    request = create_http_request(aws_request)

    context = RequestContext(request=request)
    context.service = load_service(service_name)
    context.operation = context.service.operation_model(operation_name=operation_name)
    context.region = region
    parser = create_parser(context.service)
    _, instance = parser.parse(context.request)
    context.service_request = instance
    context.account_id = get_account_id_from_request(context.request)
    return context


class _RequestContextClient:
    _client: BaseClient

    def __init__(self, client: BaseClient):
        self._client = client

    def __getattr__(self, item):
        target = getattr(self._client, item)
        if not isinstance(target, Callable):
            return target

        @functools.wraps(target)
        def wrapper_method(*args, **kwargs):
            service_name = self._client.meta.service_model.service_name
            operation_name = self._client.meta.method_to_api_mapping[item]
            region = self._client.meta.region_name
            prepared_request = None

            def event_handler(request: AWSPreparedRequest, **_):
                nonlocal prepared_request
                prepared_request = request
                # we need to return an AWS Response here
                aws_response = AWSResponse(
                    url=request.url, status_code=200, headers=HTTPHeaders(), raw=None
                )
                aws_response._content = b""
                return aws_response

            self._client.meta.events.register(
                f"before-send.{service_name}.{operation_name}", handler=event_handler
            )
            try:
                target(*args, **kwargs)
            except Exception:
                pass
            self._client.meta.events.unregister(
                f"before-send.{service_name}.{operation_name}", handler=event_handler
            )

            return create_request_context(
                service_name=service_name,
                operation_name=operation_name,
                region=region,
                aws_request=prepared_request,
            )

        return wrapper_method


def RequestContextClient[T: BaseClient](client: T) -> T:
    return _RequestContextClient(client)  # noqa


# Used for the aws_session, aws_client_factory and aws_client pytest fixtures
# Supports test executions against both LocalStack and production AWS

# TODO: Add the ability to use config profiles for primary and secondary clients
# See https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-a-configuration-file


def base_aws_session() -> boto3.Session:
    # When running against AWS, initial credentials must be read from environment or config file
    if is_aws_cloud():
        return boto3.Session()

    # Otherwise, when running against LS, use primary test credentials to start with
    # This set here in the session so that both `aws_client` and `aws_client_factory` can work without explicit creds.
    session = boto3.Session(
        aws_access_key_id=TEST_AWS_ACCESS_KEY_ID,
        aws_secret_access_key=TEST_AWS_SECRET_ACCESS_KEY,
    )
    # make sure we consider our custom data paths for legacy specs (like SQS query protocol)
    session._loader.search_paths.insert(0, LOCALSTACK_BUILTIN_DATA_PATH)
    return session


def secondary_aws_session() -> boto3.Session:
    if is_aws_cloud() and SECONDARY_TEST_AWS_PROFILE:
        return boto3.Session(profile_name=SECONDARY_TEST_AWS_PROFILE)

    # Otherwise, when running against LS or AWS, but have no profile set for the secondary account,
    # we use secondary test credentials to initialize the session.
    # This set here in the session so that both `secondary_aws_client` and `secondary_aws_client_factory` can work
    # without explicit creds.
    session = boto3.Session(
        aws_access_key_id=SECONDARY_TEST_AWS_ACCESS_KEY_ID,
        aws_secret_access_key=SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
        aws_session_token=SECONDARY_TEST_AWS_SESSION_TOKEN,
    )
    if not is_aws_cloud():
        # make sure we consider our custom data paths for legacy specs (like SQS query protocol), only if we run against
        # LocalStack
        session._loader.search_paths.append(LOCALSTACK_BUILTIN_DATA_PATH)
    return session


def base_aws_client_factory(session: boto3.Session) -> ClientFactory:
    config = None
    if is_env_true("TEST_DISABLE_RETRIES_AND_TIMEOUTS"):
        config = botocore.config.Config(
            connect_timeout=1_000,
            read_timeout=1_000,
            retries={"total_max_attempts": 1},
        )

    if is_aws_cloud():
        return ExternalAwsClientFactory(session=session, config=config)
    else:
        if not config:
            config = botocore.config.Config()

        # Prevent this fixture from using the region configured in system config
        config = config.merge(botocore.config.Config(region_name=TEST_AWS_REGION_NAME))
        return ExternalClientFactory(session=session, config=config, endpoint=TEST_AWS_ENDPOINT_URL)


def base_testing_aws_client(client_factory: ClientFactory) -> ServiceLevelClientFactory:
    # Primary test credentials are already set in the boto3 session, so they're not set here again
    return client_factory()

1	import functools	1✔
2	import os	1✔
3	from collections.abc import Callable	1✔
4
5	import boto3	1✔
6	import botocore	1✔
7	from botocore.awsrequest import AWSPreparedRequest, AWSResponse	1✔
8	from botocore.client import BaseClient	1✔
9	from botocore.compat import HTTPHeaders	1✔
10	from botocore.config import Config	1✔
11	from botocore.exceptions import ClientError	1✔
12
13	from localstack import config	1✔
14	from localstack.aws.api import RequestContext	1✔
15	from localstack.aws.connect import (	1✔
16	ClientFactory,
17	ExternalAwsClientFactory,
18	ExternalClientFactory,
19	ServiceLevelClientFactory,
20	)
21	from localstack.aws.forwarder import create_http_request	1✔
22	from localstack.aws.protocol.parser import create_parser	1✔
23	from localstack.aws.spec import LOCALSTACK_BUILTIN_DATA_PATH, load_service	1✔
24	from localstack.config import is_env_true	1✔
25	from localstack.testing.config import (	1✔
26	SECONDARY_TEST_AWS_ACCESS_KEY_ID,
27	SECONDARY_TEST_AWS_PROFILE,
28	SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
29	SECONDARY_TEST_AWS_SESSION_TOKEN,
30	TEST_AWS_ACCESS_KEY_ID,
31	TEST_AWS_ENDPOINT_URL,
32	TEST_AWS_REGION_NAME,
33	TEST_AWS_SECRET_ACCESS_KEY,
34	)
35	from localstack.utils.aws.arns import get_partition	1✔
36	from localstack.utils.aws.request_context import get_account_id_from_request	1✔
37	from localstack.utils.sync import poll_condition	1✔
38
39
40	def is_aws_cloud() -> bool:	1✔
41	return os.environ.get("TEST_TARGET", "") == "AWS_CLOUD"	1✔
42
43
44	def in_default_partition() -> bool:	1✔
45	return is_aws_cloud() or get_partition(TEST_AWS_REGION_NAME) == "aws"	1✔
46
47
48	def get_lambda_logs(func_name, logs_client):	1✔
49	log_group_name = f"/aws/lambda/{func_name}"	1✔
50	streams = logs_client.describe_log_streams(logGroupName=log_group_name)["logStreams"]	1✔
51	streams = sorted(streams, key=lambda x: x["creationTime"], reverse=True)	1✔
52	log_events = logs_client.get_log_events(	1✔
53	logGroupName=log_group_name, logStreamName=streams[0]["logStreamName"]
54	)["events"]
55	return log_events	1✔
56
57
58	def bucket_exists(client, bucket_name: str) -> bool:	1✔
UNCOV 59	buckets = client.list_buckets()	×
60	for bucket in buckets["Buckets"]:	×
61	if bucket["Name"] == bucket_name:	×
62	return True	×
63	return False	×
64
65
66	def wait_for_user(keys, region_name: str):	1✔
67	sts_client = create_client_with_keys(service="sts", keys=keys, region_name=region_name)	1✔
68
69	def is_user_ready():	1✔
70	try:	1✔
71	sts_client.get_caller_identity()	1✔
72	return True	1✔
UNCOV 73	except ClientError as e:	×
74	if e.response["Error"]["Code"] == "InvalidClientTokenId":	×
75	return False	×
76	return True	×
77
78	# wait until the given user is ready, takes AWS IAM a while...
79	poll_condition(is_user_ready, interval=5, timeout=20)	1✔
80
81
82	def create_client_with_keys(	1✔
83	service: str,
84	keys: dict[str, str],
85	region_name: str,
86	client_config: Config = None,
87	):
88	"""
89	Create a boto client with the given access key, targeted against LS per default, but to AWS if TEST_TARGET is set
90	accordingly.
91
92	:param service: Service to create the Client for
93	:param keys: Access Keys
94	:param region_name: Region for the client
95	:param client_config:
96	:return:
97	"""
98	return boto3.client(	1✔
99	service,
100	region_name=region_name,
101	aws_access_key_id=keys["AccessKeyId"],
102	aws_secret_access_key=keys["SecretAccessKey"],
103	aws_session_token=keys.get("SessionToken"),
104	config=client_config,
105	endpoint_url=config.internal_service_url() if not is_aws_cloud() else None,
106	)
107
108
109	def create_request_context(	1✔
110	service_name: str, operation_name: str, region: str, aws_request: AWSPreparedRequest
111	) -> RequestContext:
UNCOV 112	if hasattr(aws_request.body, "read"):	×
113	aws_request.body = aws_request.body.read()	×
114	request = create_http_request(aws_request)	×
115
UNCOV 116	context = RequestContext(request=request)	×
117	context.service = load_service(service_name)	×
118	context.operation = context.service.operation_model(operation_name=operation_name)	×
119	context.region = region	×
120	parser = create_parser(context.service)	×
121	_, instance = parser.parse(context.request)	×
122	context.service_request = instance	×
123	context.account_id = get_account_id_from_request(context.request)	×
124	return context	×
125
126
127	class _RequestContextClient:	1✔
128	_client: BaseClient	1✔
129
130	def __init__(self, client: BaseClient):	1✔
UNCOV 131	self._client = client	×
132
133	def __getattr__(self, item):	1✔
UNCOV 134	target = getattr(self._client, item)	×
135	if not isinstance(target, Callable):	×
136	return target	×
137
UNCOV 138	@functools.wraps(target)	×
139	def wrapper_method(args, *kwargs):	×
140	service_name = self._client.meta.service_model.service_name	×
141	operation_name = self._client.meta.method_to_api_mapping[item]	×
142	region = self._client.meta.region_name	×
143	prepared_request = None	×
144
UNCOV 145	def event_handler(request: AWSPreparedRequest, **_):	×
146	nonlocal prepared_request
UNCOV 147	prepared_request = request	×
148	# we need to return an AWS Response here
UNCOV 149	aws_response = AWSResponse(	×
150	url=request.url, status_code=200, headers=HTTPHeaders(), raw=None
151	)
UNCOV 152	aws_response._content = b""	×
153	return aws_response	×
154
UNCOV 155	self._client.meta.events.register(	×
156	f"before-send.{service_name}.{operation_name}", handler=event_handler
157	)
UNCOV 158	try:	×
159	target(args, *kwargs)	×
160	except Exception:	×
161	pass	×
162	self._client.meta.events.unregister(	×
163	f"before-send.{service_name}.{operation_name}", handler=event_handler
164	)
165
UNCOV 166	return create_request_context(	×
167	service_name=service_name,
168	operation_name=operation_name,
169	region=region,
170	aws_request=prepared_request,
171	)
172
UNCOV 173	return wrapper_method	×
174
175
176	def RequestContextClient[T: BaseClient](client: T) -> T:	1✔
UNCOV 177	return _RequestContextClient(client) # noqa	×
178
179
180	# Used for the aws_session, aws_client_factory and aws_client pytest fixtures
181	# Supports test executions against both LocalStack and production AWS
182
183	# TODO: Add the ability to use config profiles for primary and secondary clients
184	# See https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-a-configuration-file
185
186
187	def base_aws_session() -> boto3.Session:	1✔
188	# When running against AWS, initial credentials must be read from environment or config file
189	if is_aws_cloud():	1✔
UNCOV 190	return boto3.Session()	×
191
192	# Otherwise, when running against LS, use primary test credentials to start with
193	# This set here in the session so that both `aws_client` and `aws_client_factory` can work without explicit creds.
194	session = boto3.Session(	1✔
195	aws_access_key_id=TEST_AWS_ACCESS_KEY_ID,
196	aws_secret_access_key=TEST_AWS_SECRET_ACCESS_KEY,
197	)
198	# make sure we consider our custom data paths for legacy specs (like SQS query protocol)
199	session._loader.search_paths.insert(0, LOCALSTACK_BUILTIN_DATA_PATH)	1✔
200	return session	1✔
201
202
203	def secondary_aws_session() -> boto3.Session:	1✔
204	if is_aws_cloud() and SECONDARY_TEST_AWS_PROFILE:	1✔
UNCOV 205	return boto3.Session(profile_name=SECONDARY_TEST_AWS_PROFILE)	×
206
207	# Otherwise, when running against LS or AWS, but have no profile set for the secondary account,
208	# we use secondary test credentials to initialize the session.
209	# This set here in the session so that both `secondary_aws_client` and `secondary_aws_client_factory` can work
210	# without explicit creds.
211	session = boto3.Session(	1✔
212	aws_access_key_id=SECONDARY_TEST_AWS_ACCESS_KEY_ID,
213	aws_secret_access_key=SECONDARY_TEST_AWS_SECRET_ACCESS_KEY,
214	aws_session_token=SECONDARY_TEST_AWS_SESSION_TOKEN,
215	)
216	if not is_aws_cloud():	1✔
217	# make sure we consider our custom data paths for legacy specs (like SQS query protocol), only if we run against
218	# LocalStack
219	session._loader.search_paths.append(LOCALSTACK_BUILTIN_DATA_PATH)	1✔
220	return session	1✔
221
222
223	def base_aws_client_factory(session: boto3.Session) -> ClientFactory:	1✔
224	config = None	1✔
225	if is_env_true("TEST_DISABLE_RETRIES_AND_TIMEOUTS"):	1✔
UNCOV 226	config = botocore.config.Config(	×
227	connect_timeout=1_000,
228	read_timeout=1_000,
229	retries={"total_max_attempts": 1},
230	)
231
232	if is_aws_cloud():	1✔
UNCOV 233	return ExternalAwsClientFactory(session=session, config=config)	×
234	else:
235	if not config:	1✔
236	config = botocore.config.Config()	1✔
237
238	# Prevent this fixture from using the region configured in system config
239	config = config.merge(botocore.config.Config(region_name=TEST_AWS_REGION_NAME))	1✔
240	return ExternalClientFactory(session=session, config=config, endpoint=TEST_AWS_ENDPOINT_URL)	1✔
241
242
243	def base_testing_aws_client(client_factory: ClientFactory) -> ServiceLevelClientFactory:	1✔
244	# Primary test credentials are already set in the boto3 session, so they're not set here again
245	return client_factory()	1✔

localstack / localstack / 21697093787

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