feat(core): Image build (Dockerfile support)

core/README.rst

@@ -4,3 +4,17 @@ testcontainers-core
 :code:`testcontainers-core` is the core functionality for spinning up Docker containers in test environments.
 
 .. autoclass:: testcontainers.core.container.DockerContainer
+
+.. autoclass:: testcontainers.core.image.DockerImage
+
+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(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/docker_client.py

@@ -16,10 +16,12 @@
 import os
 import urllib
 import urllib.parse
+from collections.abc import Iterable
 from typing import Callable, Optional, TypeVar, Union
 
 import docker
 from docker.models.containers import Container, ContainerCollection
+from docker.models.images import Image, ImageCollection
 from typing_extensions import ParamSpec
 
 from testcontainers.core.config import testcontainers_config as c
@@ -40,6 +42,14 @@ def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
     return wrapper
 
 
+def _wrapped_image_collection(function: Callable[_P, _T]) -> Callable[_P, _T]:
+    @ft.wraps(ImageCollection.build)
+    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        return function(*args, **kwargs)
+
+    return wrapper
+
+
 class DockerClient:
     """
     Thin wrapper around :class:`docker.DockerClient` for a more functional interface.
@@ -94,6 +104,17 @@ def run(
         )
         return container
 
+    @_wrapped_image_collection
+    def build(self, path: str, tag: str, rm: bool = True, **kwargs) -> tuple[Image, Iterable[dict]]:
+        """
+        Build a Docker image from a directory containing the Dockerfile.
+
+        :return: A tuple containing the image object and the build logs.
+        """
+        image_object, image_logs = self.client.images.build(path=path, tag=tag, rm=rm, **kwargs)
+
+        return image_object, image_logs
+
     def find_host_network(self) -> Optional[str]:
         """
         Try to find the docker host network.

core/testcontainers/core/image.py

@@ -0,0 +1,88 @@
+from typing import TYPE_CHECKING, Optional
+
+from typing_extensions import Self
+
+from testcontainers.core.docker_client import DockerClient
+from testcontainers.core.utils import setup_logger
+
+if TYPE_CHECKING:
+    from docker.models.containers import Image
+
+logger = setup_logger(__name__)
+
+
+class DockerImage:
+    """
+    Basic image object to build Docker images.
+
+    .. doctest::
+
+            >>> from testcontainers.core.image import DockerImage
+
+            >>> with DockerImage(path="./core/tests/image_fixtures/sample/", tag="test-image") as image:
+            ...    logs = image.get_logs()
+
+    :param tag: Tag for the image to be built (default: None)
+    :param path: Path to the Dockerfile to build the image
+    """
+
+    def __init__(
+        self,
+        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 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.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 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()
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.remove()
+
+    def get_wrapped_image(self) -> "Image":
+        return self._image
+
+    def get_docker_client(self) -> DockerClient:
+        return self._docker
+
+    def get_logs(self) -> list[dict]:
+        return list(self._logs)

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/image_fixtures/sample/Dockerfile

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

core/tests/test_core.py

@@ -1,6 +1,11 @@
 import pytest
+import tempfile
+import random
+
+from typing import Optional
 
 from testcontainers.core.container import DockerContainer
+from testcontainers.core.image import DockerImage
 from testcontainers.core.waiting_utils import wait_for_logs
 
 
@@ -31,3 +36,31 @@ def test_can_get_logs():
         assert isinstance(stdout, bytes)
         assert isinstance(stderr, bytes)
         assert stdout, "There should be something on stdout"
+
+
+@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 "{random_string}"
+                """
+            )
+        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_docker_client.py

@@ -9,6 +9,7 @@
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.docker_client import DockerClient
 from testcontainers.core.utils import parse_docker_auth_config
+from testcontainers.core.image import DockerImage
 
 
 def test_docker_client_from_env():
@@ -54,3 +55,12 @@ def test_container_docker_client_kw():
         DockerContainer(image="", docker_client_kw=test_kwargs)
 
     mock_docker.from_env.assert_called_with(**test_kwargs)
+
+
+def test_image_docker_client_kw():
+    test_kwargs = {"test_kw": "test_value"}
+    mock_docker = MagicMock(spec=docker)
+    with patch("testcontainers.core.docker_client.docker", mock_docker):
+        DockerImage(name="", path="", docker_client_kw=test_kwargs)
+
+    mock_docker.from_env.assert_called_with(**test_kwargs)