fix running testcontainer inside a container

see testcontainers#475 (comment) for a summary

Author: CarliJoy

core/testcontainers/core/config.py

@@ -1,10 +1,29 @@
 from dataclasses import dataclass, field
+from enum import Enum
 from logging import warning
 from os import environ
 from os.path import exists
 from pathlib import Path
 from typing import Optional, Union
 
+
+class ConnectionMode(Enum):
+    bridge_ip = "bridge_ip"
+    gateway_ip = "gateway_ip"
+    docker_host = "docker_host"
+
+    @property
+    def use_mapped_port(self) -> bool:
+        """
+        Return true if we need to use mapped port for this connection
+
+        This is true for everything but bridge mode.
+        """
+        if self == self.bridge_ip:
+            return False
+        return True
+
+
 MAX_TRIES = int(environ.get("TC_MAX_TRIES", 120))
 SLEEP_TIME = int(environ.get("TC_POOLING_INTERVAL", 1))
 TIMEOUT = MAX_TRIES * SLEEP_TIME
@@ -20,6 +39,19 @@
 TC_GLOBAL = Path.home() / TC_FILE
 
 
+def get_user_overwritten_connection_mode() -> Optional[ConnectionMode]:
+    """
+    Return the user overwritten connection mode.
+    """
+    connection_mode: str | None = environ.get("TESTCONTAINERS_CONNECTION_MODE")
+    if connection_mode:
+        try:
+            return ConnectionMode(connection_mode)
+        except ValueError as e:
+            raise ValueError(f"Error parsing TESTCONTAINERS_CONNECTION_MODE: {e}") from e
+    return None
+
+
 def read_tc_properties() -> dict[str, str]:
     """
     Read the .testcontainers.properties for settings. (see the Java implementation for details)
@@ -54,6 +86,8 @@ class TestcontainersConfiguration:
     tc_properties: dict[str, str] = field(default_factory=read_tc_properties)
     _docker_auth_config: Optional[str] = field(default_factory=lambda: environ.get("DOCKER_AUTH_CONFIG"))
     tc_host_override: Optional[str] = TC_HOST_OVERRIDE
+    connection_mode_override: Optional[ConnectionMode] = None
+
     """
     https://github.com/testcontainers/testcontainers-go/blob/dd76d1e39c654433a3d80429690d07abcec04424/docker.go#L644
     if os env TC_HOST is set, use it

core/testcontainers/core/container.py

@@ -5,14 +5,15 @@
 import docker.errors
 from docker import version
 from docker.types import EndpointConfig
-from typing_extensions import Self
+from typing_extensions import Self, assert_never
 
+from testcontainers.core.config import ConnectionMode
 from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.docker_client import DockerClient
 from testcontainers.core.exceptions import ContainerStartException
 from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
 from testcontainers.core.network import Network
-from testcontainers.core.utils import inside_container, is_arm, setup_logger
+from testcontainers.core.utils import is_arm, setup_logger
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
 
 if TYPE_CHECKING:
@@ -128,33 +129,23 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         self.stop()
 
     def get_container_host_ip(self) -> str:
-        # infer from docker host
-        host = self.get_docker_client().host()
-
-        # # check testcontainers itself runs inside docker container
-        # if inside_container() and not os.getenv("DOCKER_HOST") and not host.startswith("http://"):
-        #     # If newly spawned container's gateway IP address from the docker
-        #     # "bridge" network is equal to detected host address, we should use
-        #     # container IP address, otherwise fall back to detected host
-        #     # address. Even it's inside container, we need to double check,
-        #     # because docker host might be set to docker:dind, usually in CI/CD environment
-        #     gateway_ip = self.get_docker_client().gateway_ip(self._container.id)
-
-        #     if gateway_ip == host:
-        #         return self.get_docker_client().bridge_ip(self._container.id)
-        #     return gateway_ip
-        return host
+        connection_mode: ConnectionMode
+        connection_mode = self.get_docker_client().get_connection_mode()
+        if connection_mode == ConnectionMode.docker_host:
+            return self.get_docker_client().host()
+        elif connection_mode == ConnectionMode.gateway_ip:
+            return self.get_docker_client().gateway_ip(self._container.id)
+        elif connection_mode == ConnectionMode.bridge_ip:
+            return self.get_docker_client().bridge_ip(self._container.id)
+        else:
+            # ensure that we covered all possible connection_modes
+            assert_never(connection_mode)
 
     @wait_container_is_ready()
-    def get_exposed_port(self, port: int) -> str:
-        mapped_port = self.get_docker_client().port(self._container.id, port)
-        if inside_container():
-            gateway_ip = self.get_docker_client().gateway_ip(self._container.id)
-            host = self.get_docker_client().host()
-
-            if gateway_ip == host:
-                return port
-        return mapped_port
+    def get_exposed_port(self, port: int) -> int:
+        if self.get_docker_client().get_connection_mode().use_mapped_port:
+            return self.get_docker_client().port(self._container.id, port)
+        return port
 
     def with_command(self, command: str) -> Self:
         self._command = command

core/testcontainers/core/docker_client.py

@@ -14,6 +14,7 @@
 import importlib.metadata
 import ipaddress
 import os
+import socket
 import urllib
 import urllib.parse
 from collections.abc import Iterable
@@ -25,6 +26,7 @@
 from typing_extensions import ParamSpec
 
 from testcontainers.core.auth import DockerAuthInfo, parse_docker_auth_config
+from testcontainers.core.config import ConnectionMode
 from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.labels import SESSION_ID, create_labels
 from testcontainers.core.utils import default_gateway_ip, inside_container, is_windows, setup_logger
@@ -128,7 +130,8 @@ def find_host_network(self) -> Optional[str]:
         # If we're docker in docker running on a custom network, we need to inherit the
         # network settings, so we can access the resulting container.
         try:
-            docker_host = ipaddress.IPv4Address(self.host())
+            host_ip = socket.gethostbyname(self.host())
+            docker_host = ipaddress.IPv4Address(host_ip)
             # See if we can find the host on our networks
             for network in self.client.networks.list(filters={"type": "custom"}):
                 if "IPAM" in network.attrs:
@@ -139,7 +142,7 @@ def find_host_network(self) -> Optional[str]:
                             continue
                         if docker_host in subnet:
                             return network.name
-        except ipaddress.AddressValueError:
+        except (ipaddress.AddressValueError, OSError):
             pass
         return None
 
@@ -187,6 +190,28 @@ def gateway_ip(self, container_id: str) -> str:
         network_name = self.network_name(container_id)
         return container["NetworkSettings"]["Networks"][network_name]["Gateway"]
 
+    def get_connection_mode(self) -> ConnectionMode:
+        """
+        Determine the connection mode.
+
+        See https://github.com/testcontainers/testcontainers-python/issues/475#issuecomment-2407250970
+        """
+        if c.connection_mode_override:
+            return c.connection_mode_override
+        localhosts = {"localhost", "127.0.0.1", "::1"}
+        if not inside_container() or self.host() not in localhosts:
+            # if running not inside a container or with a non-local docker client,
+            # connect ot the docker host per default
+            return ConnectionMode.docker_host
+        elif self.find_host_network():
+            # a host network could be determined, indicator for DooD,
+            # so we should connect to the bridge_ip as the container we run in
+            # and the one we started are connected to the same network
+            # that might have no access to either docker_host or the gateway
+            return ConnectionMode.bridge_ip
+        # default for DinD
+        return ConnectionMode.gateway_ip
+
     def host(self) -> str:
         """
         Get the hostname or ip address of the docker host.