18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Coverage Stats

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

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

84.54

/localstack-core/localstack/utils/diagnose.py

"""Diagnostic tool for a localstack instance running in a container."""

import inspect
import os
import socket

from localstack import config
from localstack.constants import DEFAULT_VOLUME_DIR
from localstack.services.lambda_.invocation.docker_runtime_executor import IMAGE_PREFIX
from localstack.services.lambda_.runtimes import IMAGE_MAPPING
from localstack.utils import bootstrap
from localstack.utils.analytics.metrics import MetricRegistry
from localstack.utils.container_networking import get_main_container_name
from localstack.utils.container_utils.container_client import ContainerException, NoSuchImage
from localstack.utils.docker_utils import DOCKER_CLIENT
from localstack.utils.files import load_file

LAMBDA_IMAGES = (f"{IMAGE_PREFIX}{postfix}" for postfix in IMAGE_MAPPING.values())


DIAGNOSE_IMAGES = [
    "localstack/bigdata",
    "mongo",
    *LAMBDA_IMAGES,
]

EXCLUDE_CONFIG_KEYS = {
    "CONFIG_ENV_VARS",
    "copyright",
    "__builtins__",
    "__cached__",
    "__doc__",
    "__file__",
    "__loader__",
    "__name__",
    "__package__",
    "__spec__",
}
ENDPOINT_RESOLVE_LIST = ["localhost.localstack.cloud", "api.localstack.cloud"]
INSPECT_DIRECTORIES = [DEFAULT_VOLUME_DIR, "/tmp"]


def get_localstack_logs() -> dict:
    try:
        result = DOCKER_CLIENT.get_container_logs(get_main_container_name())
    except Exception as e:
        result = f"error getting docker logs for container: {e}"

    return {"docker": result}


def get_localstack_config() -> dict:
    result = {}
    for k, v in inspect.getmembers(config):
        if k in EXCLUDE_CONFIG_KEYS:
            continue
        if inspect.isbuiltin(v):
            continue
        if inspect.isfunction(v):
            continue
        if inspect.ismodule(v):
            continue
        if inspect.isclass(v):
            continue
        if "typing." in str(type(v)):
            continue
        if k == "GATEWAY_LISTEN":
            result[k] = config.GATEWAY_LISTEN
            continue

        if hasattr(v, "__dict__"):
            result[k] = v.__dict__
        else:
            result[k] = v

    return result


def inspect_main_container() -> str | dict:
    try:
        return DOCKER_CLIENT.inspect_container(get_main_container_name())
    except Exception as e:
        return f"inspect failed: {e}"


def get_localstack_version() -> dict[str, str | None]:
    return {
        "build-date": os.environ.get("LOCALSTACK_BUILD_DATE"),
        "build-git-hash": os.environ.get("LOCALSTACK_BUILD_GIT_HASH"),
        "build-version": os.environ.get("LOCALSTACK_BUILD_VERSION"),
    }


def resolve_endpoints() -> dict[str, str]:
    result = {}
    for endpoint in ENDPOINT_RESOLVE_LIST:
        try:
            resolved_endpoint = socket.gethostbyname(endpoint)
        except Exception as e:
            resolved_endpoint = f"unable_to_resolve {e}"
        result[endpoint] = resolved_endpoint
    return result


def get_important_image_hashes() -> dict[str, str]:
    result = {}
    for image in DIAGNOSE_IMAGES:
        try:
            image_version = DOCKER_CLIENT.inspect_image(image, pull=False)["RepoDigests"]
        except NoSuchImage:
            image_version = "not_present"
        except Exception as e:
            image_version = f"error: {e}"
        result[image] = image_version
    return result


def get_service_stats() -> dict[str, str]:
    from localstack.services.plugins import SERVICE_PLUGINS

    return {service: state.value for service, state in SERVICE_PLUGINS.get_states().items()}


def get_file_tree() -> dict[str, list[str]]:
    return {d: traverse_file_tree(d) for d in INSPECT_DIRECTORIES}


def traverse_file_tree(root: str) -> list[str]:
    try:
        result = []
        if config.in_docker():
            for dirpath, _, _ in os.walk(root):
                result.append(dirpath)
        return result
    except Exception as e:
        return [f"traversing files failed {e}"]


def get_docker_image_details() -> dict[str, str]:
    try:
        image = DOCKER_CLIENT.inspect_container(get_main_container_name())["Config"]["Image"]
    except ContainerException:
        return {}
    # The default bootstrap image detection does not take custom images into account.
    # Also, the patches to correctly detect a `-pro` image are only applied on the host, so the detection fails
    # at runtime. The bootstrap detection is mostly used for the CLI, so having a different logic here makes sense.
    return bootstrap.get_docker_image_details(image_name=image)


def get_host_kernel_version() -> str:
    return load_file("/proc/version", "failed").strip()


def get_usage():
    return MetricRegistry().collect()

1	"""Diagnostic tool for a localstack instance running in a container."""
2
3	import inspect	1✔
4	import os	1✔
5	import socket	1✔
6
7	from localstack import config	1✔
8	from localstack.constants import DEFAULT_VOLUME_DIR	1✔
9	from localstack.services.lambda_.invocation.docker_runtime_executor import IMAGE_PREFIX	1✔
10	from localstack.services.lambda_.runtimes import IMAGE_MAPPING	1✔
11	from localstack.utils import bootstrap	1✔
12	from localstack.utils.analytics.metrics import MetricRegistry	1✔
13	from localstack.utils.container_networking import get_main_container_name	1✔
14	from localstack.utils.container_utils.container_client import ContainerException, NoSuchImage	1✔
15	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
16	from localstack.utils.files import load_file	1✔
17
18	LAMBDA_IMAGES = (f"{IMAGE_PREFIX}{postfix}" for postfix in IMAGE_MAPPING.values())	1✔
19
20
21	DIAGNOSE_IMAGES = [	1✔
22	"localstack/bigdata",
23	"mongo",
24	*LAMBDA_IMAGES,
25	]
26
27	EXCLUDE_CONFIG_KEYS = {	1✔
28	"CONFIG_ENV_VARS",
29	"copyright",
30	"__builtins__",
31	"__cached__",
32	"__doc__",
33	"__file__",
34	"__loader__",
35	"__name__",
36	"__package__",
37	"__spec__",
38	}
39	ENDPOINT_RESOLVE_LIST = ["localhost.localstack.cloud", "api.localstack.cloud"]	1✔
40	INSPECT_DIRECTORIES = [DEFAULT_VOLUME_DIR, "/tmp"]	1✔
41
42
43	def get_localstack_logs() -> dict:	1✔
44	try:	1✔
45	result = DOCKER_CLIENT.get_container_logs(get_main_container_name())	1✔
UNCOV 46	except Exception as e:	×
47	result = f"error getting docker logs for container: {e}"	×
48
49	return {"docker": result}	1✔
50
51
52	def get_localstack_config() -> dict:	1✔
53	result = {}	1✔
54	for k, v in inspect.getmembers(config):	1✔
55	if k in EXCLUDE_CONFIG_KEYS:	1✔
56	continue	1✔
57	if inspect.isbuiltin(v):	1✔
UNCOV 58	continue	×
59	if inspect.isfunction(v):	1✔
60	continue	1✔
61	if inspect.ismodule(v):	1✔
62	continue	1✔
63	if inspect.isclass(v):	1✔
64	continue	1✔
65	if "typing." in str(type(v)):	1✔
66	continue	1✔
67	if k == "GATEWAY_LISTEN":	1✔
68	result[k] = config.GATEWAY_LISTEN	1✔
69	continue	1✔
70
71	if hasattr(v, "__dict__"):	1✔
72	result[k] = v.__dict__	1✔
73	else:
74	result[k] = v	1✔
75
76	return result	1✔
77
78
79	def inspect_main_container() -> str \| dict:	1✔
80	try:	1✔
81	return DOCKER_CLIENT.inspect_container(get_main_container_name())	1✔
UNCOV 82	except Exception as e:	×
83	return f"inspect failed: {e}"	×
84
85
86	def get_localstack_version() -> dict[str, str \| None]:	1✔
87	return {	1✔
88	"build-date": os.environ.get("LOCALSTACK_BUILD_DATE"),
89	"build-git-hash": os.environ.get("LOCALSTACK_BUILD_GIT_HASH"),
90	"build-version": os.environ.get("LOCALSTACK_BUILD_VERSION"),
91	}
92
93
94	def resolve_endpoints() -> dict[str, str]:	1✔
95	result = {}	1✔
96	for endpoint in ENDPOINT_RESOLVE_LIST:	1✔
97	try:	1✔
98	resolved_endpoint = socket.gethostbyname(endpoint)	1✔
UNCOV 99	except Exception as e:	×
100	resolved_endpoint = f"unable_to_resolve {e}"	×
101	result[endpoint] = resolved_endpoint	1✔
102	return result	1✔
103
104
105	def get_important_image_hashes() -> dict[str, str]:	1✔
106	result = {}	1✔
107	for image in DIAGNOSE_IMAGES:	1✔
108	try:	1✔
109	image_version = DOCKER_CLIENT.inspect_image(image, pull=False)["RepoDigests"]	1✔
110	except NoSuchImage:	1✔
111	image_version = "not_present"	1✔
UNCOV 112	except Exception as e:	×
113	image_version = f"error: {e}"	×
114	result[image] = image_version	1✔
115	return result	1✔
116
117
118	def get_service_stats() -> dict[str, str]:	1✔
119	from localstack.services.plugins import SERVICE_PLUGINS	1✔
120
121	return {service: state.value for service, state in SERVICE_PLUGINS.get_states().items()}	1✔
122
123
124	def get_file_tree() -> dict[str, list[str]]:	1✔
125	return {d: traverse_file_tree(d) for d in INSPECT_DIRECTORIES}	1✔
126
127
128	def traverse_file_tree(root: str) -> list[str]:	1✔
129	try:	1✔
130	result = []	1✔
131	if config.in_docker():	1✔
UNCOV 132	for dirpath, _, _ in os.walk(root):	×
133	result.append(dirpath)	×
134	return result	1✔
UNCOV 135	except Exception as e:	×
136	return [f"traversing files failed {e}"]	×
137
138
139	def get_docker_image_details() -> dict[str, str]:	1✔
140	try:	1✔
141	image = DOCKER_CLIENT.inspect_container(get_main_container_name())["Config"]["Image"]	1✔
UNCOV 142	except ContainerException:	×
143	return {}	×
144	# The default bootstrap image detection does not take custom images into account.
145	# Also, the patches to correctly detect a `-pro` image are only applied on the host, so the detection fails
146	# at runtime. The bootstrap detection is mostly used for the CLI, so having a different logic here makes sense.
147	return bootstrap.get_docker_image_details(image_name=image)	1✔
148
149
150	def get_host_kernel_version() -> str:	1✔
151	return load_file("/proc/version", "failed").strip()	1✔
152
153
154	def get_usage():	1✔
155	return MetricRegistry().collect()	1✔

localstack / localstack / 18505123992

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