fix(network): Now able to use Network without context, and has labels to be automatically cleaned up (testcontainers#627) (testcontainers#630)

This PR adds `create` method to the `Network` class to enable a
non-context-manager usage.

Fixes testcontainers#627

---------

Co-authored-by: Dave Ankin <[email protected]>

Author: dogukancagatay

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")

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()

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()