diff --git a/conf.py b/conf.py index b310e939..35c2ae9c 100644 --- a/conf.py +++ b/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"), +] diff --git a/core/README.rst b/core/README.rst index 8cc9a278..1461ba7d 100644 --- a/core/README.rst +++ b/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 diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py index f677182f..da486fb5 100644 --- a/core/testcontainers/core/container.py +++ b/core/testcontainers/core/container.py @@ -65,11 +65,38 @@ def with_env_file(self, env_file: Union[str, PathLike]) -> Self: self.with_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("nginx") + >>> container = container.with_bind_ports("8080/tcp", 8080) + >>> container = container.with_bind_ports("8081/tcp", 8081) + + """ 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("nginx") + >>> container = container.with_exposed_ports("8080/tcp", "8081/tcp") + + """ + for port in ports: self.ports[port] = None return self diff --git a/core/tests/conftest.py b/core/tests/conftest.py index a86faa10..a6d8ecb8 100644 --- a/core/tests/conftest.py +++ b/core/tests/conftest.py @@ -3,6 +3,7 @@ import pytest from typing import Callable from testcontainers.core.container import DockerClient +from pprint import pprint import sys PROJECT_DIR = Path(__file__).parent.parent.parent.resolve() @@ -50,3 +51,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 diff --git a/core/tests/test_core_ports.py b/core/tests/test_core_ports.py new file mode 100644 index 00000000..148ddf08 --- /dev/null +++ b/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() + + # prepare to inspect container + container_id = container._container.id + client = container._container.client + + # assemble expected output to compare to container API + container_port = str(container_port) + host_port = str(host_port or "") + + # if the port protocol is not specified, it will default to tcp + if "/" not in container_port: + container_port += "/tcp" + + expected = {container_port: [{"HostIp": "", "HostPort": host_port}]} + + # compare PortBindings to expected output + assert client.containers.get(container_id).attrs["HostConfig"]["PortBindings"] == expected + 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()