feat(core): Protocol support for container port bind and expose

Author: Tranquility2

conf.py

@@ -161,4 +161,9 @@
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
     "selenium": ("https://seleniumhq.github.io/selenium/docs/api/py/", None),
+    "typing_extensions": ("https://typing-extensions.readthedocs.io/en/latest/", None),
 }
+
+nitpick_ignore = [
+    ("py:class", "typing_extensions.Self"),
+]

core/README.rst

@@ -4,6 +4,15 @@ Testcontainers Core
 :code:`testcontainers-core` is the core functionality for spinning up Docker containers in test environments.
 
 .. autoclass:: testcontainers.core.container.DockerContainer
+    :members: with_bind_ports, with_exposed_ports
+
+.. note::
+    When using `with_bind_ports` or `with_exposed_ports`
+    you can specify the port in the following formats: :code:`{private_port}/{protocol}`
+
+    e.g. `8080/tcp` or `8125/udp` or just `8080` (default protocol is tcp)
+
+    For legacy reasons, the port can be an *integer*
 
 .. autoclass:: testcontainers.core.image.DockerImage
 

core/testcontainers/core/container.py

@@ -1,7 +1,7 @@
 import contextlib
 from platform import system
 from socket import socket
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Optional, Union
 
 import docker.errors
 from docker import version
@@ -57,11 +57,38 @@ def with_env(self, key: str, value: str) -> Self:
         self.env[key] = value
         return self
 
-    def with_bind_ports(self, container: int, host: Optional[int] = None) -> Self:
+    def with_bind_ports(self, container: Union[str, int], host: Optional[Union[str, int]] = None) -> Self:
+        """
+        Bind container port to host port
+
+        :param container: container port
+        :param host: host port
+
+        :doctest:
+
+        >>> from testcontainers.core.container import DockerContainer
+        >>> container = DockerContainer("alpine:latest")
+        >>> container.with_bind_ports("8080/tcp", 8080)
+
+        """
+
         self.ports[container] = host
         return self
 
-    def with_exposed_ports(self, *ports: int) -> Self:
+    def with_exposed_ports(self, *ports: tuple[Union[str, int], ...]) -> Self:
+        """
+        Expose ports from the container without binding them to the host.
+
+        :param ports: ports to expose
+
+        :doctest:
+
+        >>> from testcontainers.core.container import DockerContainer
+        >>> container = DockerContainer("alpine:latest")
+        >>> container.with_exposed_ports(8080/tcp, 8081)
+
+        """
+
         for port in ports:
             self.ports[port] = None
         return self

core/tests/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 from typing import Callable
 from testcontainers.core.container import DockerClient
+from pprint import pprint
 
 
 @pytest.fixture
@@ -20,3 +21,16 @@ def _check_for_image(image_short_id: str, cleaned: bool) -> None:
         assert found is not cleaned, f'Image {image_short_id} was {"found" if cleaned else "not found"}'
 
     return _check_for_image
+
+
+@pytest.fixture
+def show_container_attributes() -> None:
+    """Wrap the show_container_attributes function in a fixture"""
+
+    def _show_container_attributes(container_id: str) -> None:
+        """Print the attributes of a container"""
+        client = DockerClient().client
+        data = client.containers.get(container_id).attrs
+        pprint(data)
+
+    return _show_container_attributes

core/tests/test_core_ports.py

@@ -0,0 +1,99 @@
+import pytest
+from typing import Union, Optional
+from testcontainers.core.container import DockerContainer
+
+from docker.errors import APIError
+
+
+@pytest.mark.parametrize(
+    "container_port, host_port",
+    [
+        ("8080", "8080"),
+        ("8125/udp", "8125/udp"),
+        ("8092/udp", "8092/udp"),
+        ("9000/tcp", "9000/tcp"),
+        ("8080", "8080/udp"),
+        (8080, 8080),
+        (9000, None),
+        ("9009", None),
+        ("9000", ""),
+        ("9000/udp", ""),
+    ],
+)
+def test_docker_container_with_bind_ports(container_port: Union[str, int], host_port: Optional[Union[str, int]]):
+    container = DockerContainer("alpine:latest")
+    container.with_bind_ports(container_port, host_port)
+    container.start()
+
+    container_id = container._container.id
+    client = container._container.client
+
+    if isinstance(container_port, int):
+        container_port = str(container_port)
+    if isinstance(host_port, int):
+        host_port = str(host_port)
+    if not host_port:
+        host_port = ""
+
+    # if the port protocol is not specified, it will default to tcp
+    if "/" not in container_port:
+        container_port += "/tcp"
+
+    excepted = {container_port: [{"HostIp": "", "HostPort": host_port}]}
+    assert client.containers.get(container_id).attrs["HostConfig"]["PortBindings"] == excepted
+    container.stop()
+
+
+@pytest.mark.parametrize(
+    "container_port, host_port",
+    [
+        ("0", "8080"),
+        ("8080", "abc"),
+        (0, 0),
+        (-1, 8080),
+        (None, 8080),
+    ],
+)
+def test_error_docker_container_with_bind_ports(container_port: Union[str, int], host_port: Optional[Union[str, int]]):
+    with pytest.raises(APIError):
+        container = DockerContainer("alpine:latest")
+        container.with_bind_ports(container_port, host_port)
+        container.start()
+
+
+@pytest.mark.parametrize(
+    "ports, expected",
+    [
+        (("8125/udp",), {"8125/udp": {}}),
+        (("8092/udp", "9000/tcp"), {"8092/udp": {}, "9000/tcp": {}}),
+        (("8080", "8080/udp"), {"8080/tcp": {}, "8080/udp": {}}),
+        ((9000,), {"9000/tcp": {}}),
+        ((8080, 8080), {"8080/tcp": {}}),
+        (("9001", 9002), {"9001/tcp": {}, "9002/tcp": {}}),
+        (("9001", 9002, "9003/udp", 9004), {"9001/tcp": {}, "9002/tcp": {}, "9003/udp": {}, "9004/tcp": {}}),
+    ],
+)
+def test_docker_container_with_exposed_ports(ports: tuple[Union[str, int], ...], expected: dict):
+    container = DockerContainer("alpine:latest")
+    container.with_exposed_ports(*ports)
+    container.start()
+
+    container_id = container._container.id
+    client = container._container.client
+    assert client.containers.get(container_id).attrs["Config"]["ExposedPorts"] == expected
+    container.stop()
+
+
+@pytest.mark.parametrize(
+    "ports",
+    [
+        ((9000, None)),
+        (("", 9000)),
+        ("tcp", ""),
+    ],
+)
+def test_error_docker_container_with_exposed_ports(ports: tuple[Union[str, int], ...]):
+    with pytest.raises(APIError):
+        container = DockerContainer("alpine:latest")
+        container.with_exposed_ports(*ports)
+        container.start()