diff --git a/core/testcontainers/core/docker_client.py b/core/testcontainers/core/docker_client.py index 9b7fe747..286e1ef9 100644 --- a/core/testcontainers/core/docker_client.py +++ b/core/testcontainers/core/docker_client.py @@ -211,6 +211,10 @@ def login(self, docker_auth_config: str) -> None: login_info = self.client.login(**auth_config._asdict()) LOGGER.debug(f"logged in using {login_info}") + def client_networks_create(self, name: str, param: dict): + labels = create_labels("", param.get("labels")) + return self.client.networks.create(name, **{**param, "labels": labels}) + def get_docker_host() -> Optional[str]: return c.tc_properties_get_tc_host() or os.getenv("DOCKER_HOST") diff --git a/core/testcontainers/core/network.py b/core/testcontainers/core/network.py index 9903d071..d149d5e4 100644 --- a/core/testcontainers/core/network.py +++ b/core/testcontainers/core/network.py @@ -32,10 +32,13 @@ def connect(self, container_id: str, network_aliases: Optional[list] = None): def remove(self) -> None: self._network.remove() - def __enter__(self) -> "Network": - self._network = self._docker.client.networks.create(self.name, **self._docker_network_kw) + def create(self) -> "Network": + self._network = self._docker.client_networks_create(self.name, self._docker_network_kw) self.id = self._network.id return self + def __enter__(self) -> "Network": + return self.create() + def __exit__(self, exc_type, exc_val, exc_tb) -> None: self.remove() diff --git a/core/tests/test_network.py b/core/tests/test_network.py index 4b0764d4..7191153b 100644 --- a/core/tests/test_network.py +++ b/core/tests/test_network.py @@ -1,7 +1,12 @@ +from http import HTTPStatus from testcontainers.core.container import DockerContainer from testcontainers.core.docker_client import DockerClient +from testcontainers.core.labels import LABEL_SESSION_ID from testcontainers.core.network import Network +import docker.errors +import pytest + NGINX_ALPINE_SLIM_IMAGE = "nginx:1.25.4-alpine-slim" @@ -14,6 +19,31 @@ def test_network_gets_created_and_cleaned_up(): assert not docker.client.networks.list(network.name) +def test_network_create_wo_cm(): + network = Network() + network.create() + docker = DockerClient() + networks_list = docker.client.networks.list(network.name) + assert networks_list[0].name == network.name + assert networks_list[0].id == network.id + + network.remove() + assert not docker.client.networks.list(network.name) + + +def test_network_create_errors(): + network = Network() + network.create() + + # calling create the second time should raise an error + with pytest.raises(docker.errors.APIError) as excinfo: + network.create() + + assert excinfo.value.response.status_code == HTTPStatus.CONFLICT + excinfo.match(f"network with name {network.name} already exists") + network.remove() + + def test_containers_can_communicate_over_network(): with Network() as network: with ( @@ -41,3 +71,13 @@ def assert_can_ping(container: DockerContainer, remote_name: str): status, output = container.exec("ping -c 1 %s" % remote_name) assert status == 0 assert "64 bytes" in str(output) + + +def test_network_has_labels(): + network = Network() + try: + network.create() + network = network._docker.client.networks.get(network_id=network.id) + assert LABEL_SESSION_ID in network.attrs.get("Labels") + finally: + network.remove()