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

69.35

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

import logging
import re
import socket
from functools import lru_cache
from typing import Union

import boto3

from localstack import config
from localstack.config import S3_VIRTUAL_HOSTNAME
from localstack.constants import (
    LOCALHOST,
)
from localstack.utils.strings import is_string_or_bytes, to_str

# set up logger
LOG = logging.getLogger(__name__)

# cached value used to determine the DNS status of the S3 hostname (whether it can be resolved properly)
CACHE_S3_HOSTNAME_DNS_STATUS = None


@lru_cache
def get_valid_regions():
    session = boto3.Session()
    valid_regions = set()
    for partition in set(session.get_available_partitions()):
        for region in session.get_available_regions("sns", partition):
            valid_regions.add(region)
    return valid_regions


# FIXME: AWS recommends use of SSM parameter store to determine per region availability
# https://github.com/aws/aws-sdk/issues/206#issuecomment-1471354853
@lru_cache
def get_valid_regions_for_service(service_name):
    session = boto3.Session()
    regions = list(session.get_available_regions(service_name))
    regions.extend(session.get_available_regions("cloudwatch", partition_name="aws-us-gov"))
    regions.extend(session.get_available_regions("cloudwatch", partition_name="aws-cn"))
    return regions


def get_boto3_region() -> str:
    """Return the region name, as determined from the environment when creating a new boto3 session"""
    return boto3.session.Session().region_name


def get_local_service_url(service_name_or_port: Union[str, int]) -> str:
    """Return the local service URL for the given service name or port."""
    # TODO(srw): we don't need to differentiate on service name any more, so remove the argument
    if isinstance(service_name_or_port, int):
        return f"{config.get_protocol()}://{LOCALHOST}:{service_name_or_port}"
    return config.internal_service_url()


def get_s3_hostname():
    global CACHE_S3_HOSTNAME_DNS_STATUS
    if CACHE_S3_HOSTNAME_DNS_STATUS is None:
        try:
            assert socket.gethostbyname(S3_VIRTUAL_HOSTNAME)
            CACHE_S3_HOSTNAME_DNS_STATUS = True
        except OSError:
            CACHE_S3_HOSTNAME_DNS_STATUS = False
    if CACHE_S3_HOSTNAME_DNS_STATUS:
        return S3_VIRTUAL_HOSTNAME
    return LOCALHOST


def fix_account_id_in_arns(
    response, replacement: str, colon_delimiter: str = ":", existing: Union[str, list[str]] = None
):
    """Fix the account ID in the ARNs returned in the given Flask response or string"""
    from moto.core import DEFAULT_ACCOUNT_ID

    existing = existing or ["123456789", "1234567890", DEFAULT_ACCOUNT_ID]
    existing = existing if isinstance(existing, list) else [existing]
    is_str_obj = is_string_or_bytes(response)
    content = to_str(response if is_str_obj else response._content)

    replacement = rf"arn{colon_delimiter}aws{colon_delimiter}\1{colon_delimiter}\2{colon_delimiter}{replacement}{colon_delimiter}"
    for acc_id in existing:
        regex = rf"arn{colon_delimiter}aws{colon_delimiter}([^:%]+){colon_delimiter}([^:%]*){colon_delimiter}{acc_id}{colon_delimiter}"
        content = re.sub(regex, replacement, content)

    if not is_str_obj:
        response._content = content
        response.headers["Content-Length"] = len(response._content)
        return response
    return content


def inject_test_credentials_into_env(env):
    if "AWS_ACCESS_KEY_ID" not in env and "AWS_SECRET_ACCESS_KEY" not in env:
        env["AWS_ACCESS_KEY_ID"] = "test"
        env["AWS_SECRET_ACCESS_KEY"] = "test"

1	import logging	1✔
2	import re	1✔
3	import socket	1✔
4	from functools import lru_cache	1✔
5	from typing import Union	1✔
6
7	import boto3	1✔
8
9	from localstack import config	1✔
10	from localstack.config import S3_VIRTUAL_HOSTNAME	1✔
11	from localstack.constants import (	1✔
12	LOCALHOST,
13	)
14	from localstack.utils.strings import is_string_or_bytes, to_str	1✔
15
16	# set up logger
17	LOG = logging.getLogger(__name__)	1✔
18
19	# cached value used to determine the DNS status of the S3 hostname (whether it can be resolved properly)
20	CACHE_S3_HOSTNAME_DNS_STATUS = None	1✔
21
22
23	@lru_cache	1✔
24	def get_valid_regions():	1✔
25	session = boto3.Session()	1✔
26	valid_regions = set()	1✔
27	for partition in set(session.get_available_partitions()):	1✔
28	for region in session.get_available_regions("sns", partition):	1✔
29	valid_regions.add(region)	1✔
30	return valid_regions	1✔
31
32
33	# FIXME: AWS recommends use of SSM parameter store to determine per region availability
34	# https://github.com/aws/aws-sdk/issues/206#issuecomment-1471354853
35	@lru_cache	1✔
36	def get_valid_regions_for_service(service_name):	1✔
37	session = boto3.Session()	1✔
38	regions = list(session.get_available_regions(service_name))	1✔
39	regions.extend(session.get_available_regions("cloudwatch", partition_name="aws-us-gov"))	1✔
40	regions.extend(session.get_available_regions("cloudwatch", partition_name="aws-cn"))	1✔
41	return regions	1✔
42
43
44	def get_boto3_region() -> str:	1✔
45	"""Return the region name, as determined from the environment when creating a new boto3 session"""
46	return boto3.session.Session().region_name	×
47
48
49	def get_local_service_url(service_name_or_port: Union[str, int]) -> str:	1✔
50	"""Return the local service URL for the given service name or port."""
51	# TODO(srw): we don't need to differentiate on service name any more, so remove the argument
52	if isinstance(service_name_or_port, int):	1✔
53	return f"{config.get_protocol()}://{LOCALHOST}:{service_name_or_port}"	×
54	return config.internal_service_url()	1✔
55
56
57	def get_s3_hostname():	1✔
58	global CACHE_S3_HOSTNAME_DNS_STATUS
59	if CACHE_S3_HOSTNAME_DNS_STATUS is None:	1✔
60	try:	1✔
61	assert socket.gethostbyname(S3_VIRTUAL_HOSTNAME)	1✔
62	CACHE_S3_HOSTNAME_DNS_STATUS = True	1✔
63	except OSError:	×
64	CACHE_S3_HOSTNAME_DNS_STATUS = False	×
65	if CACHE_S3_HOSTNAME_DNS_STATUS:	1✔
66	return S3_VIRTUAL_HOSTNAME	1✔
67	return LOCALHOST	×
68
69
70	def fix_account_id_in_arns(	1✔
71	response, replacement: str, colon_delimiter: str = ":", existing: Union[str, list[str]] = None
72	):
73	"""Fix the account ID in the ARNs returned in the given Flask response or string"""
74	from moto.core import DEFAULT_ACCOUNT_ID	×
75
76	existing = existing or ["123456789", "1234567890", DEFAULT_ACCOUNT_ID]	×
77	existing = existing if isinstance(existing, list) else [existing]	×
78	is_str_obj = is_string_or_bytes(response)	×
79	content = to_str(response if is_str_obj else response._content)	×
80
81	replacement = rf"arn{colon_delimiter}aws{colon_delimiter}\1{colon_delimiter}\2{colon_delimiter}{replacement}{colon_delimiter}"	×
UNCOV 82	for acc_id in existing:	×
UNCOV 83	regex = rf"arn{colon_delimiter}aws{colon_delimiter}([^:%]+){colon_delimiter}([^:%]*){colon_delimiter}{acc_id}{colon_delimiter}"	×
84	content = re.sub(regex, replacement, content)	×
85
UNCOV 86	if not is_str_obj:	×
UNCOV 87	response._content = content	×
88	response.headers["Content-Length"] = len(response._content)	×
UNCOV 89	return response	×
90	return content	×
91
92
93	def inject_test_credentials_into_env(env):	1✔
94	if "AWS_ACCESS_KEY_ID" not in env and "AWS_SECRET_ACCESS_KEY" not in env:	1✔
95	env["AWS_ACCESS_KEY_ID"] = "test"	1✔
96	env["AWS_SECRET_ACCESS_KEY"] = "test"	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