testcontainers · Tranquility2 · Apr 1, 2026 · Apr 1, 2026 · Apr 1, 2026 · Apr 1, 2026
diff --git a/core/README.rst b/core/README.rst
@@ -51,3 +51,20 @@ Using `DockerContainer` and `DockerImage` to create a container:
 
 The `DockerImage` class is used to build the image from the specified path and tag.
 The `DockerContainer` class is then used to create a container from the image.
+
+Copying a file from disk into a container:
+
+.. doctest::
+
+    >>> import tempfile
+    >>> from pathlib import Path
+    >>> from testcontainers.core.container import DockerContainer
+
+    >>> with tempfile.TemporaryDirectory() as tmp:
+    ...     my_file = Path(tmp) / "my_file.txt"
+    ...     _ = my_file.write_text("file content")
+    ...     with DockerContainer("bash", command="sleep infinity") as container:
+    ...         container.copy_into_container(my_file, "/tmp/my_file.txt")
+    ...         result = container.exec("cat /tmp/my_file.txt")
+    ...         result.output
+    b'file content'
diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py
@@ -21,7 +21,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.transferable import Transferable, TransferSpec
+from testcontainers.core.transferable import Transferable, TransferSpec, build_transfer_tar
 from testcontainers.core.utils import is_arm, setup_logger
 from testcontainers.core.wait_strategies import LogMessageWaitStrategy
 from testcontainers.core.waiting_utils import WaitStrategy
@@ -289,7 +289,7 @@ def with_volume_mapping(self, host: Union[str, PathLike[str]], container: str, m
 
     def with_tmpfs_mount(self, container_path: str, size: Optional[str] = None) -> Self:
         """Mount a tmpfs volume on the container.
-        
+
         :param container_path: Container path to mount tmpfs on (e.g., '/data')
         :param size: Optional size limit (e.g., '256m', '1g'). If None, unbounded.
         :return: Self for chaining
@@ -342,57 +342,24 @@ def copy_into_container(self, transferable: Transferable, destination_in_contain
         return self._transfer_into_container(transferable, destination_in_container, mode)
 
     def _transfer_into_container(self, transferable: Transferable, destination_in_container: str, mode: int) -> None:
-        if isinstance(transferable, bytes):
-            self._transfer_file_content_into_container(transferable, destination_in_container, mode)
-        elif isinstance(transferable, pathlib.Path):
-            if transferable.is_file():
-                self._transfer_file_content_into_container(transferable.read_bytes(), destination_in_container, mode)
-            elif transferable.is_dir():
-                self._transfer_directory_into_container(transferable, destination_in_container, mode)
-            else:
-                raise TypeError(f"Path {transferable} is neither a file nor directory")
-        else:
-            raise TypeError("source must be bytes or PathLike")
-
-    def _transfer_file_content_into_container(
-        self, file_content: bytes, destination_in_container: str, mode: int
-    ) -> None:
-        fileobj = io.BytesIO()
-        with tarfile.open(fileobj=fileobj, mode="w") as tar:
-            tarinfo = tarfile.TarInfo(name=destination_in_container)
-            tarinfo.size = len(file_content)
-            tarinfo.mode = mode
-            tar.addfile(tarinfo, io.BytesIO(file_content))
-        fileobj.seek(0)
-        assert self._container is not None
-        rv = self._container.put_archive(path="/", data=fileobj.getvalue())
-        assert rv is True
-
-    def _transfer_directory_into_container(
-        self, source_directory: pathlib.Path, destination_in_container: str, mode: int
-    ) -> None:
-        assert self._container is not None
-        result = self._container.exec_run(["mkdir", "-p", destination_in_container])
-        assert result.exit_code == 0
+        if not self._container:
+            raise ContainerStartException("Container must be started before transferring files")
 
-        fileobj = io.BytesIO()
-        with tarfile.open(fileobj=fileobj, mode="w") as tar:
-            tar.add(source_directory, arcname=source_directory.name)
-        fileobj.seek(0)
-        rv = self._container.put_archive(path=destination_in_container, data=fileobj.getvalue())
-        assert rv is True
+        data = build_transfer_tar(transferable, destination_in_container, mode)
+        if not self._container.put_archive(path="/", data=data):
+            raise OSError(f"Failed to put archive into container at {destination_in_container}")
 
     def copy_from_container(self, source_in_container: str, destination_on_host: pathlib.Path) -> None:
-        assert self._container is not None
+        if not self._container:
+            raise ContainerStartException("Container must be started before copying files")
+
         tar_stream, _ = self._container.get_archive(source_in_container)
 
-        for chunk in tar_stream:
-            with tarfile.open(fileobj=io.BytesIO(chunk)) as tar:
-                for member in tar.getmembers():
-                    with open(destination_on_host, "wb") as f:
-                        fileobj = tar.extractfile(member)
-                        assert fileobj is not None
-                        f.write(fileobj.read())
+        with tarfile.open(fileobj=io.BytesIO(b"".join(tar_stream))) as tar:
+            for member in tar.getmembers():
+                extracted = tar.extractfile(member)
+                if extracted is not None:
+                    destination_on_host.write_bytes(extracted.read())
 
 
 class Reaper:

diff --git a/core/testcontainers/core/transferable.py b/core/testcontainers/core/transferable.py
@@ -1,6 +1,33 @@
+import io
 import pathlib
+import tarfile
 from typing import Union
 
 Transferable = Union[bytes, pathlib.Path]
 
 TransferSpec = Union[tuple[Transferable, str], tuple[Transferable, str, int]]
+
+
+def build_transfer_tar(transferable: Transferable, destination: str, mode: int = 0o644) -> bytes:
+    """Build a tar archive containing the transferable, ready for put_archive(path="/")."""
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w") as tar:
+        if isinstance(transferable, bytes):
+            info = tarfile.TarInfo(name=destination)
+            info.size = len(transferable)
+            info.mode = mode
+            tar.addfile(info, io.BytesIO(transferable))
+        elif isinstance(transferable, pathlib.Path):
+            if transferable.is_file():
+                info = tarfile.TarInfo(name=destination)
+                info.size = transferable.stat().st_size
+                info.mode = mode
+                with transferable.open("rb") as f:
+                    tar.addfile(info, f)
+            elif transferable.is_dir():
+                tar.add(str(transferable), arcname=f"{destination.rstrip('/')}/{transferable.name}")
+            else:
+                raise TypeError(f"Path {transferable} is neither a file nor directory")
+        else:
+            raise TypeError("source must be bytes or Path")
+    return buf.getvalue()
diff --git a/core/tests/test_core.py b/core/tests/test_core.py
@@ -1,9 +1,7 @@
 import tempfile
 from pathlib import Path
 
-import pytest
 from testcontainers.core.container import DockerContainer
-from testcontainers.core.transferable import Transferable, TransferSpec
 
 
 def test_garbage_collection_is_defensive():
@@ -48,109 +46,3 @@ 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)
-
-
-@pytest.fixture(name="transferable", params=(bytes, Path))
-def copy_sources_fixture(request, tmp_path: Path):
-    """
-    Provide source argument for tests of copy_into_container
-    """
-    raw_data = b"hello world"
-    if request.param is bytes:
-        return raw_data
-    elif request.param is Path:
-        my_file = tmp_path / "my_file"
-        my_file.write_bytes(raw_data)
-        return my_file
-    pytest.fail("Invalid type")
-
-
-def test_copy_into_container_at_runtime(transferable: Transferable):
-    # Given
-    destination_in_container = "/tmp/my_file"
-
-    with DockerContainer("bash", command="sleep infinity") as container:
-        # When
-        container.copy_into_container(transferable, destination_in_container)
-        result = container.exec(f"cat {destination_in_container}")
-
-    # Then
-    assert result.exit_code == 0
-    assert result.output == b"hello world"
-
-
-def test_copy_into_container_at_startup(transferable: Transferable):
-    # Given
-    destination_in_container = "/tmp/my_file"
-
-    container = DockerContainer("bash", command="sleep infinity")
-    container.with_copy_into_container(transferable, destination_in_container)
-
-    with container:
-        # When
-        result = container.exec(f"cat {destination_in_container}")
-
-    # Then
-    assert result.exit_code == 0
-    assert result.output == b"hello world"
-
-
-def test_copy_into_container_via_initializer(transferable: Transferable):
-    # Given
-    destination_in_container = "/tmp/my_file"
-    transferables: list[TransferSpec] = [(transferable, destination_in_container, 0o644)]
-
-    with DockerContainer("bash", command="sleep infinity", transferables=transferables) as container:
-        # When
-        result = container.exec(f"cat {destination_in_container}")
-
-    # Then
-    assert result.exit_code == 0
-    assert result.output == b"hello world"
-
-
-def test_copy_file_from_container(tmp_path: Path):
-    # Given
-    file_in_container = "/tmp/foo.txt"
-    destination_on_host = tmp_path / "foo.txt"
-    assert not destination_on_host.is_file()
-
-    with DockerContainer("bash", command="sleep infinity") as container:
-        result = container.exec(f'bash -c "echo -n hello world > {file_in_container}"')
-        assert result.exit_code == 0
-
-        # When
-        container.copy_from_container(file_in_container, destination_on_host)
-
-    # Then
-    assert destination_on_host.is_file()
-    assert destination_on_host.read_text() == "hello world"
-
-
-def test_copy_directory_into_container(tmp_path: Path):
-    # Given
-    source_dir = tmp_path / "my_directory"
-    source_dir.mkdir()
-    my_file = source_dir / "my_file"
-    my_file.write_bytes(b"hello world")
-
-    destination_in_container = "/tmp/my_destination_directory"
-
-    with DockerContainer("bash", command="sleep infinity") as container:
-        # When
-        container.copy_into_container(source_dir, destination_in_container)
-        result = container.exec(f"ls {destination_in_container}")
-
-        # Then - my_directory exists
-        assert result.exit_code == 0
-        assert result.output == b"my_directory\n"
-
-        # Then - my_file is in directory
-        result = container.exec(f"ls {destination_in_container}/my_directory")
-        assert result.exit_code == 0
-        assert result.output == b"my_file\n"
-
-        # Then - my_file contents are correct
-        result = container.exec(f"cat {destination_in_container}/my_directory/my_file")
-        assert result.exit_code == 0
-        assert result.output == b"hello world"