allow determining host network by running container id

core/testcontainers/core/docker_client.py

@@ -10,6 +10,7 @@
 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 #    License for the specific language governing permissions and limitations
 #    under the License.
+import contextlib
 import functools as ft
 import importlib.metadata
 import ipaddress
@@ -129,6 +130,15 @@ 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.
+
+        # first to try to find the network the container runs in, if we can determine
+        container_id = utils.get_running_in_container_id()
+        if container_id:
+            with contextlib.suppress(Exception):
+                return self.network_name(container_id)
+
+        # if this results nothing, try to determine the network based on the
+        # docker_host
         try:
             host_ip = socket.gethostbyname(self.host())
             docker_host = ipaddress.IPv4Address(host_ip)

core/testcontainers/core/utils.py

@@ -3,7 +3,8 @@
 import platform
 import subprocess
 import sys
-from typing import Any, Optional
+from pathlib import Path
+from typing import Any, Final, Optional
 
 LINUX = "linux"
 MAC = "mac"
@@ -80,3 +81,20 @@ def raise_for_deprecated_parameter(kwargs: dict[Any, Any], name: str, replacemen
     if kwargs.pop(name, None):
         raise ValueError(f"Use `{replacement}` instead of `{name}`")
     return kwargs
+
+
+CGROUP_FILE: Final[Path] = Path("/proc/self/cgroup")
+
+
+def get_running_in_container_id() -> str | None:
+    """
+    Get the id of the currently running container
+    """
+    if not CGROUP_FILE.is_file():
+        return None
+    cgroup = CGROUP_FILE.read_text()
+    for line in cgroup.splitlines(keepends=False):
+        path = line.rpartition(":")[2]
+        if path.startswith("/docker"):
+            return path.removeprefix("/docker/")
+    return None

core/tests/test_docker_client.py

@@ -1,6 +1,7 @@
 import os
 import json
 from collections import namedtuple
+from typing import Any
 from unittest import mock
 from unittest.mock import MagicMock, patch
 
@@ -249,3 +250,46 @@ def networks(self):
     monkeypatch.setattr(client, "client", FakeClient())
 
     assert client.find_host_network() == "runner-346da30e-2641-1-8365005"
+
+
+def test_find_host_network_found_by_running_id(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = DockerClient()
+    fake_id = "abcde1234"
+
+    def network_name(container_id: str) -> str:
+        assert container_id == fake_id
+        return "FAKE_NETWORK"
+
+    monkeypatch.setattr(utils, "get_running_in_container_id", lambda: fake_id)
+    monkeypatch.setattr(client, "network_name", network_name)
+
+    assert client.find_host_network() == "FAKE_NETWORK"
+
+
+def test_run_uses_found_network(monkeypatch: pytest.MonkeyPatch) -> None:
+    """
+    If a host network is found, use it
+    """
+
+    client = DockerClient()
+
+    class ContainerRunFake:
+        def __init__(self) -> None:
+            self.calls: list[dict[str, Any]] = []
+
+        def run(self, image: str, **kwargs: Any) -> str:
+            self.calls.append(kwargs)
+            return "CONTAINER"
+
+    class FakeClient:
+        def __init__(self) -> None:
+            self.containers = ContainerRunFake()
+
+    fake_client = FakeClient()
+
+    monkeypatch.setattr(client, "find_host_network", lambda: "new_bridge_network")
+    monkeypatch.setattr(client, "client", fake_client)
+
+    assert client.run("test") == "CONTAINER"
+
+    assert fake_client.containers.calls[0]["network"] == "new_bridge_network"

core/tests/test_utils.py

@@ -1,3 +1,6 @@
+from pathlib import Path
+
+import pytest
 from pytest import MonkeyPatch, raises, mark
 
 from testcontainers.core import utils
@@ -51,3 +54,24 @@ def test_raise_for_deprecated_parameters() -> None:
         result = utils.raise_for_deprecated_parameter(kwargs, current, replacement)
         assert str(e.value) == "Parameter 'deprecated' is deprecated and should be replaced by 'replacement'."
         assert result == {}
+
+
+@pytest.fixture
+def fake_cgroup(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
+    target = tmp_path / "cgroup"
+    monkeypatch.setattr(utils, "CGROUP_FILE", target)
+    return target
+
+
+def test_get_running_container_id_empty_or_missing(fake_cgroup: Path) -> None:
+    # non existing does not fail but is only none
+    assert utils.get_running_in_container_id() is None
+    fake_cgroup.write_text("12:devices:/system.slice/sshd.service\n" "13:cpuset:\n")
+    # missing docker does also not fail
+    assert utils.get_running_in_container_id() is None
+
+
+def test_get_running_container_id(fake_cgroup: Path) -> None:
+    container_id = "b78eebb08f89158ed6e2ed2fe"
+    fake_cgroup.write_text(f"13:cpuset:/docker/{container_id}")
+    assert utils.get_running_in_container_id() == container_id