feat: support with_copy_to

Author: guenhter

core/testcontainers/core/container.py

@@ -1,6 +1,7 @@
 import contextlib
 import sys
 from os import PathLike
+from pathlib import Path
 from socket import socket
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union, cast
@@ -18,7 +19,7 @@
 from testcontainers.core.exceptions import ContainerConnectException, ContainerStartException
 from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
 from testcontainers.core.network import Network
-from testcontainers.core.utils import is_arm, setup_logger
+from testcontainers.core.utils import build_tar_file, is_arm, setup_logger
 from testcontainers.core.wait_strategies import LogMessageWaitStrategy
 from testcontainers.core.waiting_utils import WaitStrategy
 
@@ -97,6 +98,7 @@ def __init__(
 
         self._kwargs = kwargs
         self._wait_strategy: Optional[WaitStrategy] = _wait_strategy
+        self._copy_to_container: list[tuple[str, Union[bytes, Path]]] = []
 
     def with_env(self, key: str, value: str) -> Self:
         self.env[key] = value
@@ -190,17 +192,21 @@ def start(self) -> Self:
             else {}
         )
 
-        self._container = docker_client.run(
+        self._container = docker_client.create(
             self.image,
             command=self._command,
-            detach=True,
             environment=self.env,
             ports=cast("dict[int, Optional[int]]", self.ports),
             name=self._name,
             volumes=self.volumes,
             **{**network_kwargs, **self._kwargs},
         )
 
+        for target, source in self._copy_to_container:
+            self._container.put_archive("/", build_tar_file(target, source))
+
+        docker_client.start(self._container)
+
         if self._wait_strategy is not None:
             self._wait_strategy.wait_until_ready(self)
 
@@ -270,6 +276,27 @@ def with_volume_mapping(self, host: Union[str, PathLike[str]], container: str, m
         self.volumes[str(host)] = mapping
         return self
 
+    def with_copy_to(self, target: str, source: Union[bytes, str, PathLike[str]]) -> Self:
+        """
+        Copy a file, directory, or raw bytes into the container at startup.
+
+        :param target: Absolute path inside the container where the data should be placed.
+        :param source: Either ``bytes``/``bytearray`` (raw file content) or a path
+            (``str`` / :class:`pathlib.Path`) to a local file or directory.
+
+        :doctest:
+
+        >>> from testcontainers.core.container import DockerContainer
+        >>> container = DockerContainer("alpine")
+        >>> container = container.with_copy_to("/tmp/hello.txt", b"hello world")
+
+        """
+        if isinstance(source, (bytes, bytearray)):
+            self._copy_to_container.append((target, bytes(source)))
+        else:
+            self._copy_to_container.append((target, Path(source)))
+        return self
+
     def get_wrapped_container(self) -> "Container":
         return self._container
 
@@ -301,6 +328,13 @@ def exec(self, command: Union[str, list[str]]) -> ExecResult:
             raise ContainerStartException("Container should be started before executing a command")
         return self._container.exec_run(command)
 
+    def wait(self) -> int:
+        """Wait for the container to stop and return its exit code."""
+        if not self._container:
+            raise ContainerStartException("Container should be started before waiting")
+        result = self._container.wait()
+        return int(result["StatusCode"])
+
     def _configure(self) -> None:
         # placeholder if subclasses want to define this and use the default start method
         pass

core/testcontainers/core/docker_client.py

@@ -49,6 +49,14 @@ def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
     return wrapper
 
 
+def _wrapped_container_collection_create(function: Callable[_P, _T]) -> Callable[_P, _T]:
+    @ft.wraps(ContainerCollection.create)
+    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        return function(*args, **kwargs)
+
+    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:
@@ -114,6 +122,42 @@ def run(
         )
         return container
 
+    @_wrapped_container_collection_create
+    def create(
+        self,
+        image: str,
+        command: Optional[Union[str, list[str]]] = None,
+        environment: Optional[dict[str, str]] = None,
+        ports: Optional[dict[int, Optional[int]]] = None,
+        labels: Optional[dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> Container:
+        """Create a container without starting it, pulling the image first if not present locally."""
+        if "network" not in kwargs and not get_docker_host():
+            host_network = self.find_host_network()
+            if host_network:
+                kwargs["network"] = host_network
+
+        try:
+            # This is more or less a replication of what the self.client.containers.start does internally
+            self.client.images.get(image)
+        except docker.errors.ImageNotFound:
+            self.client.images.pull(image)
+
+        container = self.client.containers.create(
+            image,
+            command=command,
+            environment=environment,
+            ports=ports,
+            labels=create_labels(image, labels),
+            **kwargs,
+        )
+        return container
+
+    def start(self, container: Container) -> None:
+        """Start a previously created container."""
+        container.start()
+
     @_wrapped_image_collection
     def build(
         self, path: str, tag: Optional[str], rm: bool = True, **kwargs: Any

core/testcontainers/core/utils.py

@@ -1,10 +1,12 @@
+import io
 import logging
 import os
 import platform
 import subprocess
 import sys
+import tarfile
 from pathlib import Path
-from typing import Any, Final, Optional
+from typing import Any, Final, Optional, Union
 
 LINUX = "linux"
 MAC = "mac"
@@ -98,3 +100,20 @@ def get_running_in_container_id() -> Optional[str]:
         if path.startswith("/docker"):
             return path.removeprefix("/docker/")
     return None
+
+
+def build_tar_file(target: str, source: Union[bytes, Path]) -> bytes:
+    """Pack *source* into an in-memory tar archive whose member path equals *target* (relative to /)."""
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w") as tar:
+        # Docker's put_archive extracts relative to the given path; we upload to "/"
+        # so the member name must be the target path stripped of its leading slash.
+        arcname = target.lstrip("/")
+        if isinstance(source, bytes):
+            info = tarfile.TarInfo(name=arcname)
+            info.size = len(source)
+            info.mode = 0o644
+            tar.addfile(info, io.BytesIO(source))
+        else:
+            tar.add(str(source), arcname=arcname)
+    return buf.getvalue()

core/tests/test_core.py

@@ -46,3 +46,50 @@ def test_docker_container_with_env_file():
             assert "ADMIN_EMAIL=admin@example.org" in output
             assert "ROOT_URL=example.org/app" in output
             print(output)
+
+
+# ---------------------------------------------------------------------------
+# with_copy_to
+# ---------------------------------------------------------------------------
+
+
+def test_with_copy_to_bytes():
+    """Bytes passed to with_copy_to should be readable inside the running container."""
+    with (
+        DockerContainer("alpine")
+        .with_command(["cat", "/tmp/hello.txt"])
+        .with_copy_to("/tmp/hello.txt", b"hello from bytes") as c
+    ):
+        c.wait()
+        stdout, _ = c.get_logs()
+        assert stdout.decode() == "hello from bytes"
+
+
+def test_with_copy_to_file(tmp_path: Path):
+    """A local file passed to with_copy_to should be readable inside the running container."""
+    src = tmp_path / "copied.txt"
+    src.write_bytes(b"hello from file")
+
+    with DockerContainer("alpine").with_command(["cat", "/tmp/copied.txt"]).with_copy_to("/tmp/copied.txt", src) as c:
+        c.wait()
+        stdout, _ = c.get_logs()
+        assert stdout.decode() == "hello from file"
+
+
+def test_with_copy_to_directory(tmp_path: Path):
+    """A local directory passed to with_copy_to should be readable inside the running container."""
+    (tmp_path / "a.txt").write_text("aaa")
+    sub = tmp_path / "sub"
+    sub.mkdir()
+    (sub / "b.txt").write_text("bbb")
+
+    with (
+        DockerContainer("alpine")
+        .with_command(["sh", "-c", "cat /mydata/a.txt && cat /mydata/sub/b.txt"])
+        .with_copy_to("/mydata", tmp_path)
+    ) as c:
+        c.wait()
+        stdout, _ = c.get_logs()
+        output = stdout.decode()
+        assert "aaa" in output
+        assert "bbb" in output

core/tests/test_utils.py

@@ -1,3 +1,5 @@
+import io
+import tarfile
 from pathlib import Path
 
 import pytest
@@ -76,3 +78,32 @@ def test_get_running_container_id(fake_cgroup: Path) -> None:
     container_id = "b78eebb08f89158ed6e2ed2fe"
     fake_cgroup.write_text(f"13:cpuset:/docker/{container_id}")
     assert utils.get_running_in_container_id() == container_id
+
+
+# ---------------------------------------------------------------------------
+# build_copy_to_tar
+# ---------------------------------------------------------------------------
+
+
+def test_build_copy_to_tar_bytes() -> None:
+    data = b"hello world"
+    tar_bytes = utils.build_tar_file("/tmp/hello.txt", data)
+
+    with tarfile.open(fileobj=io.BytesIO(tar_bytes)) as tar:
+        members = tar.getmembers()
+        assert len(members) == 1
+        assert members[0].name == "tmp/hello.txt"
+        assert tar.extractfile(members[0]).read() == data
+
+
+def test_build_copy_to_tar_file(tmp_path: Path) -> None:
+    src = tmp_path / "myfile.txt"
+    src.write_bytes(b"file content")
+
+    tar_bytes = utils.build_tar_file("/tmp/myfile.txt", src)
+
+    with tarfile.open(fileobj=io.BytesIO(tar_bytes)) as tar:
+        members = tar.getmembers()
+        assert len(members) == 1
+        assert members[0].name == "tmp/myfile.txt"
+        assert tar.extractfile(members[0]).read() == b"file content"

modules/oracle-free/example_basic.py

@@ -1,5 +1,4 @@
 import oracledb
-
 from testcontainers.oracle_free import OracleFreeContainer