20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

56.52

/localstack-core/localstack/utils/container_networking.py

import logging
import os
import re
from functools import lru_cache

from localstack import config, constants
from localstack.utils.container_utils.container_client import ContainerException
from localstack.utils.docker_utils import DOCKER_CLIENT
from localstack.utils.net import get_docker_host_from_container

LOG = logging.getLogger(__name__)


@lru_cache
def get_main_container_network() -> str | None:
    """
    Gets the main network of the LocalStack container (if we run in one, bridge otherwise)
    If there are multiple networks connected to the LocalStack container, we choose the first as "main" network

    :return: Network name
    """
    if config.MAIN_DOCKER_NETWORK:
        if config.is_in_docker:
            networks = DOCKER_CLIENT.get_networks(get_main_container_name())
            if config.MAIN_DOCKER_NETWORK not in networks:
                LOG.warning(
                    "The specified 'MAIN_DOCKER_NETWORK' is not connected to the LocalStack container! Falling back to %s",
                    networks[0],
                )
                return networks[0]
        return config.MAIN_DOCKER_NETWORK

    # use the default bridge network in case of host mode or if we can't resolve the networks for the main container
    main_container_network = "bridge"
    if config.is_in_docker:
        try:
            networks = DOCKER_CLIENT.get_networks(get_main_container_name())
            main_container_network = networks[0]
        except Exception as e:
            container_name = get_main_container_name()
            LOG.info(
                'Unable to get network name of main container "%s", falling back to "bridge": %s',
                container_name,
                e,
            )

    LOG.info("Determined main container network: %s", main_container_network)
    return main_container_network


@lru_cache
def get_endpoint_for_network(network: str | None = None) -> str:
    """
    Get the LocalStack endpoint (= IP address) on the given network.
    If a network is given, it will return the IP address/hostname of LocalStack on that network
    If omitted, it will return the IP address/hostname of the main container network
    This is a cached call, clear cache if networks might have changed

    :param network: Network to return the endpoint for
    :return: IP address or hostname of LS on the given network
    """
    container_name = get_main_container_name()
    network = network or get_main_container_network()
    main_container_ip = None
    try:
        if config.is_in_docker:
            main_container_ip = DOCKER_CLIENT.get_container_ipv4_for_network(
                container_name_or_id=container_name,
                container_network=network,
            )
        else:
            # default gateway for the network should be the host
            # In a Linux host-mode environment, the default gateway for the network should be the IP of the host
            if config.is_in_linux:
                main_container_ip = DOCKER_CLIENT.inspect_network(network)["IPAM"]["Config"][0][
                    "Gateway"
                ]
            else:
                # In a non-Linux host-mode environment, we need to determine the IP of the host by running a container
                # (basically macOS host mode, i.e. this is a feature to improve the developer experience)
                image_name = constants.DOCKER_IMAGE_NAME
                out, _ = DOCKER_CLIENT.run_container(
                    image_name,
                    remove=True,
                    entrypoint="",
                    command=["ping", "-c", "1", "host.docker.internal"],
                )
                out = out.decode(config.DEFAULT_ENCODING) if isinstance(out, bytes) else out
                ip = re.match(r"PING[^\(]+\(([^\)]+)\).*", out, re.MULTILINE | re.DOTALL)
                ip = ip and ip.group(1)
                if ip:
                    main_container_ip = ip
        LOG.info("Determined main container target IP: %s", main_container_ip)
    except Exception as e:
        LOG.info("Unable to get main container IP address: %s", e)

    if not main_container_ip:
        # fall back to returning the hostname/IP of the Docker host, if we cannot determine the main container IP
        return get_docker_host_from_container()

    return main_container_ip


def get_main_container_ip():
    """
    Get the container IP address of the LocalStack container.
    Use get_endpoint_for network where possible, as it allows better control about which address to return

    :return: IP address of LocalStack container
    """
    container_name = get_main_container_name()
    return DOCKER_CLIENT.get_container_ip(container_name)


def get_main_container_id():
    """
    Return the container ID of the LocalStack container

    :return: container ID
    """
    container_name = get_main_container_name()
    try:
        return DOCKER_CLIENT.get_container_id(container_name)
    except ContainerException:
        return None


@lru_cache
def get_main_container_name():
    """
    Returns the container name of the LocalStack container

    :return: LocalStack container name
    """
    hostname = os.environ.get("HOSTNAME")
    if hostname:
        try:
            return DOCKER_CLIENT.get_container_name(hostname)
        except ContainerException:
            return config.MAIN_CONTAINER_NAME
    else:
        return config.MAIN_CONTAINER_NAME

1	import logging	1✔
2	import os	1✔
3	import re	1✔
4	from functools import lru_cache	1✔
5
6	from localstack import config, constants	1✔
7	from localstack.utils.container_utils.container_client import ContainerException	1✔
8	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
9	from localstack.utils.net import get_docker_host_from_container	1✔
10
11	LOG = logging.getLogger(__name__)	1✔
12
13
14	@lru_cache	1✔
15	def get_main_container_network() -> str \| None:	1✔
16	"""
17	Gets the main network of the LocalStack container (if we run in one, bridge otherwise)
18	If there are multiple networks connected to the LocalStack container, we choose the first as "main" network
19
20	:return: Network name
21	"""
22	if config.MAIN_DOCKER_NETWORK:	1✔
23	if config.is_in_docker:	×
24	networks = DOCKER_CLIENT.get_networks(get_main_container_name())	×
25	if config.MAIN_DOCKER_NETWORK not in networks:	×
26	LOG.warning(	×
27	"The specified 'MAIN_DOCKER_NETWORK' is not connected to the LocalStack container! Falling back to %s",
28	networks[0],
29	)
30	return networks[0]	×
31	return config.MAIN_DOCKER_NETWORK	×
32
33	# use the default bridge network in case of host mode or if we can't resolve the networks for the main container
34	main_container_network = "bridge"	1✔
35	if config.is_in_docker:	1✔
36	try:	1✔
37	networks = DOCKER_CLIENT.get_networks(get_main_container_name())	1✔
38	main_container_network = networks[0]	1✔
39	except Exception as e:	×
40	container_name = get_main_container_name()	×
41	LOG.info(	×
42	'Unable to get network name of main container "%s", falling back to "bridge": %s',
43	container_name,
44	e,
45	)
46
47	LOG.info("Determined main container network: %s", main_container_network)	1✔
48	return main_container_network	1✔
49
50
51	@lru_cache	1✔
52	def get_endpoint_for_network(network: str \| None = None) -> str:	1✔
53	"""
54	Get the LocalStack endpoint (= IP address) on the given network.
55	If a network is given, it will return the IP address/hostname of LocalStack on that network
56	If omitted, it will return the IP address/hostname of the main container network
57	This is a cached call, clear cache if networks might have changed
58
59	:param network: Network to return the endpoint for
60	:return: IP address or hostname of LS on the given network
61	"""
62	container_name = get_main_container_name()	1✔
63	network = network or get_main_container_network()	1✔
64	main_container_ip = None	1✔
65	try:	1✔
66	if config.is_in_docker:	1✔
67	main_container_ip = DOCKER_CLIENT.get_container_ipv4_for_network(	1✔
68	container_name_or_id=container_name,
69	container_network=network,
70	)
71	else:
72	# default gateway for the network should be the host
73	# In a Linux host-mode environment, the default gateway for the network should be the IP of the host
74	if config.is_in_linux:	×
75	main_container_ip = DOCKER_CLIENT.inspect_network(network)["IPAM"]["Config"][0][	×
76	"Gateway"
77	]
78	else:
79	# In a non-Linux host-mode environment, we need to determine the IP of the host by running a container
80	# (basically macOS host mode, i.e. this is a feature to improve the developer experience)
81	image_name = constants.DOCKER_IMAGE_NAME	×
82	out, _ = DOCKER_CLIENT.run_container(	×
83	image_name,
84	remove=True,
85	entrypoint="",
86	command=["ping", "-c", "1", "host.docker.internal"],
87	)
88	out = out.decode(config.DEFAULT_ENCODING) if isinstance(out, bytes) else out	×
89	ip = re.match(r"PING[^\(]+\(([^\)]+)\).*", out, re.MULTILINE \| re.DOTALL)	×
90	ip = ip and ip.group(1)	×
91	if ip:	×
92	main_container_ip = ip	×
93	LOG.info("Determined main container target IP: %s", main_container_ip)	1✔
94	except Exception as e:	×
95	LOG.info("Unable to get main container IP address: %s", e)	×
96
97	if not main_container_ip:	1✔
98	# fall back to returning the hostname/IP of the Docker host, if we cannot determine the main container IP
99	return get_docker_host_from_container()	×
100
101	return main_container_ip	1✔
102
103
104	def get_main_container_ip():	1✔
105	"""
106	Get the container IP address of the LocalStack container.
107	Use get_endpoint_for network where possible, as it allows better control about which address to return
108
109	:return: IP address of LocalStack container
110	"""
111	container_name = get_main_container_name()	×
112	return DOCKER_CLIENT.get_container_ip(container_name)	×
113
114
115	def get_main_container_id():	1✔
116	"""
117	Return the container ID of the LocalStack container
118
119	:return: container ID
120	"""
121	container_name = get_main_container_name()	×
122	try:	×
123	return DOCKER_CLIENT.get_container_id(container_name)	×
124	except ContainerException:	×
125	return None	×
126
127
128	@lru_cache	1✔
129	def get_main_container_name():	1✔
130	"""
131	Returns the container name of the LocalStack container
132
133	:return: LocalStack container name
134	"""
135	hostname = os.environ.get("HOSTNAME")	1✔
136	if hostname:	1✔
137	try:	1✔
138	return DOCKER_CLIENT.get_container_name(hostname)	1✔
139	except ContainerException:	×
140	return config.MAIN_CONTAINER_NAME	×
141	else:
142	return config.MAIN_CONTAINER_NAME	1✔

localstack / localstack / 20565403496

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