feat(core): Added SrvContainer

Added doctest to CustomContainer and update readme

Fix issue with CustomContainer image handle

Add test for CustomContainer

Update image var on CustomContainer

Image build based on id

Image clean is now optional

Fixed doctest

Update log about image removal

Improve image related test flow

Refactor CustomContainer into SrvContainer

Refactor image related test flow

Improve SrvContainer

Fix test_srv_container logs check

Improve SrvContainer and update tests

Updates for SrvContainer

Author: Tranquility2

core/README.rst

@@ -7,13 +7,16 @@ testcontainers-core
 
 .. autoclass:: testcontainers.core.image.DockerImage
 
-Conjoint usage:
+.. autoclass:: testcontainers.core.generic.SrvContainer
+
+Using `DockerContainer` and `DockerImage` directly:
 
 .. doctest::
+
     >>> from testcontainers.core.container import DockerContainer
     >>> from testcontainers.core.waiting_utils import wait_for_logs
     >>> from testcontainers.core.image import DockerImage
 
-    >>> with DockerImage(tag="test-image:latest", path=".") as image:
-    ...     with DockerContainer(tag=image.tag) as container:
-    ...         delay = wait_for_log(container, "Server ready", 60)
+    >>> with DockerImage(path="./core/tests/image_fixtures/sample/", tag="test-sample:latest") as image:
+    ...     with DockerContainer(str(image)) as container:
+    ...         delay = wait_for_logs(container, "Test Sample Image")

core/testcontainers/core/generic.py

@@ -11,10 +11,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 from typing import Optional
+from urllib.error import HTTPError
 from urllib.parse import quote
+from urllib.request import urlopen
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.exceptions import ContainerStartException
+from testcontainers.core.image import DockerImage
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
@@ -79,3 +82,67 @@ def _configure(self) -> None:
 
     def _transfer_seed(self) -> None:
         pass
+
+
+class SrvContainer(DockerContainer):
+    """
+    Container for a generic server that is based on a custom image.
+
+    Example:
+
+    .. doctest::
+
+        >>> import httpx
+        >>> from testcontainers.core.generic import SrvContainer
+        >>> from testcontainers.core.waiting_utils import wait_for_logs
+
+        >>> with SrvContainer(path="./core/tests/image_fixtures/python_server", port=9000, tag="test-srv:latest") as srv:
+        ...     url = srv._create_connection_url()
+        ...     response = httpx.get(f"{url}", timeout=5)
+        ...     assert response.status_code == 200, "Response status code is not 200"
+        ...     delay = wait_for_logs(srv, "GET / HTTP/1.1")
+
+
+    :param path: Path to the Dockerfile to build the image
+    :param tag: Tag for the image to be built (default: None)
+    """
+
+    def __init__(self, path: str, port: int, tag: Optional[str], image_cleanup: bool = True) -> None:
+        self.docker_image = DockerImage(path=path, tag=tag, clean_up=image_cleanup).build()
+        super().__init__(str(self.docker_image))
+        self.internal_port = port
+        self.with_exposed_ports(self.internal_port)
+
+    @wait_container_is_ready(HTTPError)
+    def _connect(self) -> None:
+        # noinspection HttpUrlsUsage
+        url = self._create_connection_url()
+        try:
+            with urlopen(url) as r:
+                assert b"" in r.read()
+        except HTTPError as e:
+            # 404 is expected, as the server may not have the specific endpoint we are looking for
+            if e.code == 404:
+                pass
+            else:
+                raise
+
+    def get_api_url(self) -> str:
+        raise NotImplementedError
+
+    def _create_connection_url(self) -> str:
+        if self._container is None:
+            raise ContainerStartException("container has not been started")
+        host = self.get_container_host_ip()
+        exposed_port = self.get_exposed_port(self.internal_port)
+        url = f"http://{host}:{exposed_port}"
+        return url
+
+    def start(self) -> "SrvContainer":
+        super().start()
+        self._connect()
+        return self
+
+    def stop(self, force=True, delete_volume=True) -> None:
+        super().stop(force, delete_volume)
+        self.docker_image.remove()

core/testcontainers/core/image.py

@@ -16,44 +16,61 @@ class DockerImage:
     Basic image object to build Docker images.
 
     .. doctest::
+
             >>> from testcontainers.core.image import DockerImage
-            >>> with DockerImage(tag="new_image", path=".") as image:
+            >>> with DockerImage(path="./core/tests/image_fixtures/sample/", tag="test-image") as image:
             ...    logs = image.get_logs()
 
-    :param tag: Tag of the image
-    :param path: Path to the Dockerfile
+    :param tag: Tag for the image to be built (default: None)
+    :param path: Path to the Dockerfile to build the image
     """
 
     def __init__(
         self,
-        tag: str,
         path: str,
         docker_client_kw: Optional[dict] = None,
+        tag: Optional[str] = None,
+        clean_up: bool = True,
         **kwargs,
     ) -> None:
         self.tag = tag
         self.path = path
+        self.id = None
         self._docker = DockerClient(**(docker_client_kw or {}))
+        self.clean_up = clean_up
         self._kwargs = kwargs
 
     def build(self, **kwargs) -> Self:
-        logger.info(f"Building image {self.tag} from {self.path}")
+        logger.info(f"Building image from {self.path}")
         docker_client = self.get_docker_client()
         self._image, self._logs = docker_client.build(path=self.path, tag=self.tag, **kwargs)
-        logger.info(f"Built image {self.tag}")
+        logger.info(f"Built image {self.short_id} with tag {self.tag}")
         return self
 
+    @property
+    def short_id(self) -> str:
+        """
+        The ID of the image truncated to 12 characters, without the ``sha256:`` prefix.
+        """
+        if self._image.id.startswith("sha256:"):
+            return self._image.id.split(":")[1][:12]
+        return self._image.id[:12]
+
     def remove(self, force=True, noprune=False) -> None:
         """
         Remove the image.
 
         :param force: Remove the image even if it is in use
         :param noprune: Do not delete untagged parent images
         """
-        if self._image:
+        if self._image and self.clean_up:
+            logger.info(f"Removing image {self.short_id}")
             self._image.remove(force=force, noprune=noprune)
         self.get_docker_client().client.close()
 
+    def __str__(self) -> str:
+        return f"{self.tag if self.tag else self.short_id}"
+
     def __enter__(self) -> Self:
         return self.build()
 

core/tests/conftest.py

@@ -0,0 +1,22 @@
+import pytest
+from typing import Callable
+from testcontainers.core.container import DockerClient
+
+
+@pytest.fixture
+def check_for_image() -> Callable[[str, bool], None]:
+    """Warp the check_for_image function in a fixture"""
+
+    def _check_for_image(image_short_id: str, cleaned: bool) -> None:
+        """
+        Validates if the image is present or not.
+
+        :param image_short_id: The short id of the image
+        :param cleaned: True if the image should not be present, False otherwise
+        """
+        client = DockerClient()
+        images = client.client.images.list()
+        found = any(image.short_id.endswith(image_short_id) for image in images)
+        assert found is not cleaned, f'Image {image_short_id} was {"found" if cleaned else "not found"}'
+
+    return _check_for_image

core/tests/Dockerfile renamed to core/tests/image_fixtures/busybox/Dockerfile

@@ -0,0 +1,3 @@
+FROM python:3
+EXPOSE 9000
+CMD ["python", "-m", "http.server", "9000"]

core/tests/image_fixtures/python_server/Dockerfile

@@ -0,0 +1,2 @@
+FROM alpine:latest
+CMD echo "Test Sample Image"

core/tests/image_fixtures/sample/Dockerfile

@@ -1,5 +1,8 @@
 import pytest
 import tempfile
+import random
+
+from typing import Optional
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.image import DockerImage
@@ -35,22 +38,29 @@ def test_can_get_logs():
         assert stdout, "There should be something on stdout"
 
 
-def test_from_image():
-    test_image_tag = "test-image:latest"
+@pytest.mark.parametrize("test_cleanup", [True, False])
+@pytest.mark.parametrize("test_image_tag", [None, "test-image:latest"])
+def test_docker_image(test_image_tag: Optional[str], test_cleanup: bool, check_for_image):
     with tempfile.TemporaryDirectory() as temp_directory:
+        # It's important to use a random string to avoid image caching
+        random_string = "Hello from Docker Image! " + str(random.randint(0, 1000))
         with open(f"{temp_directory}/Dockerfile", "w") as f:
             f.write(
-                """
+                f"""
                 FROM alpine:latest
-                CMD echo "Hello from Docker Image!"
+                CMD echo "{random_string}"
                 """
             )
-            f.close()
-            with DockerImage(tag=test_image_tag, path=temp_directory) as image:
-                assert image.tag == test_image_tag
-                logs = image.get_logs()
-                assert isinstance(logs, list)
-                assert logs[0] == {"stream": "Step 1/2 : FROM alpine:latest"}
-                assert logs[3] == {"stream": 'Step 2/2 : CMD echo "Hello from Docker Image!"'}
-                with DockerContainer(image.tag) as container:
-                    wait_for_logs(container, "Hello from Docker Image!")
+        with DockerImage(path=temp_directory, tag=test_image_tag, clean_up=test_cleanup) as image:
+            image_short_id = image.short_id
+            assert image.tag is test_image_tag, f"Expected {test_image_tag}, got {image.tag}"
+            assert image.short_id is not None, "Short ID should not be None"
+            logs = image.get_logs()
+            assert isinstance(logs, list), "Logs should be a list"
+            assert logs[0] == {"stream": "Step 1/2 : FROM alpine:latest"}
+            assert logs[3] == {"stream": f'Step 2/2 : CMD echo "{random_string}"'}
+            with DockerContainer(str(image)) as container:
+                assert container._container.image.short_id.endswith(image_short_id), "Image ID mismatch"
+                assert container.get_logs() == ((random_string + "\n").encode(), b""), "Container logs mismatch"
+
+        check_for_image(image_short_id, test_cleanup)

core/tests/test_core.py

@@ -0,0 +1,27 @@
+import pytest
+from typing import Optional
+from testcontainers.core.generic import SrvContainer
+
+import re
+
+
+@pytest.mark.parametrize("test_image_cleanup", [True, False])
+@pytest.mark.parametrize("test_image_tag", [None, "custom-image:test"])
+def test_srv_container(test_image_tag: Optional[str], test_image_cleanup: bool, check_for_image, port=9000):
+    with SrvContainer(
+        path="./core/tests/image_fixtures/python_server",
+        port=port,
+        tag=test_image_tag,
+        image_cleanup=test_image_cleanup,
+    ) as srv:
+        image_short_id = srv.docker_image.short_id
+        image_build_logs = srv.docker_image.get_logs()
+        # check if dict is in any of the logs
+        assert {"stream": f"Step 2/3 : EXPOSE {port}"} in image_build_logs, "Image logs mismatch"
+        assert (port, None) in srv.ports.items(), "Port mismatch"
+        with pytest.raises(NotImplementedError):
+            srv.get_api_url()
+        test_url = srv._create_connection_url()
+        assert re.match(r"http://localhost:\d+", test_url), "Connection URL mismatch"
+
+    check_for_image(image_short_id, test_image_cleanup)