18767774552

Committed 23 Oct 2025 02:23PM UTC coverage: 86.899% (+0.006%) from 86.893%

Build # 18767774552

Build Type

push

github

Committed by

web-flow

Commit Message

CFn: Remove deprecated and non-functional deploy GUI tool (#13293)

Run Details

68342 of 78645 relevant lines covered (86.9%)

0.87 hits per line

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

59.02

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

import logging
import re
import socket
from functools import lru_cache

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

localstack / localstack / 18767774552

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