diff --git a/core/README.rst b/core/README.rst
index 2d364d0a..32ce157b 100644
--- a/core/README.rst
+++ b/core/README.rst
@@ -14,6 +14,8 @@ Testcontainers Core
.. autoclass:: testcontainers.core.generic.DbContainer
+.. autoclass:: testcontainers.core.transferable.Transferable
+
.. raw:: html
diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py
index e0456fa0..c56b3743 100644
--- a/core/testcontainers/core/container.py
+++ b/core/testcontainers/core/container.py
@@ -1,4 +1,7 @@
import contextlib
+import io
+import pathlib
+import tarfile
from os import PathLike
from socket import socket
from types import TracebackType
@@ -17,6 +20,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
from testcontainers.core.utils import is_arm, setup_logger
from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
@@ -69,6 +73,7 @@ def __init__(
volumes: Optional[list[tuple[str, str, str]]] = None,
network: Optional[Network] = None,
network_aliases: Optional[list[str]] = None,
+ transferables: Optional[list[Transferable]] = None,
**kwargs: Any,
) -> None:
self.env = env or {}
@@ -96,6 +101,7 @@ def __init__(
self.with_network_aliases(*network_aliases)
self._kwargs = kwargs
+ self._transferables: list[Transferable] = transferables or []
def with_env(self, key: str, value: str) -> Self:
self.env[key] = value
@@ -196,6 +202,10 @@ def start(self) -> Self:
)
logger.info("Container started: %s", self._container.short_id)
+
+ for t in self._transferables:
+ self._transfer_into_container(t.source, t.destination_in_container, t.mode)
+
return self
def stop(self, force: bool = True, delete_volume: bool = True) -> None:
@@ -273,6 +283,50 @@ def _configure(self) -> None:
# placeholder if subclasses want to define this and use the default start method
pass
+ def with_copy_into_container(
+ self, file_content: Union[bytes, pathlib.Path], destination_in_container: str, mode: int = 0o644
+ ) -> Self:
+ self._transferables.append(Transferable(file_content, destination_in_container, mode))
+ return self
+
+ def copy_into_container(
+ self, file_content: Union[bytes, pathlib.Path], destination_in_container: str, mode: int = 0o644
+ ) -> None:
+ return self._transfer_into_container(file_content, destination_in_container, mode)
+
+ def _transfer_into_container(
+ self, source: Union[bytes, pathlib.Path], destination_in_container: str, mode: int
+ ) -> None:
+ if isinstance(source, bytes):
+ file_content = source
+ elif isinstance(source, pathlib.Path):
+ file_content = source.read_bytes()
+ else:
+ raise TypeError("source must be bytes or PathLike")
+
+ 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 copy_from_container(self, source_in_container: str, destination_on_host: pathlib.Path) -> None:
+ assert self._container is not None
+ 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())
+
class Reaper:
"""
diff --git a/core/testcontainers/core/transferable.py b/core/testcontainers/core/transferable.py
new file mode 100644
index 00000000..5d3f6a80
--- /dev/null
+++ b/core/testcontainers/core/transferable.py
@@ -0,0 +1,14 @@
+import dataclasses
+import pathlib
+from typing import Union
+
+
+@dataclasses.dataclass
+class Transferable:
+ """
+ Wrapper class enabling copying files into a container
+ """
+
+ source: Union[bytes, pathlib.Path]
+ destination_in_container: str
+ mode: int = 0o644
diff --git a/core/tests/test_core.py b/core/tests/test_core.py
index 9312b0bc..2c03fc16 100644
--- a/core/tests/test_core.py
+++ b/core/tests/test_core.py
@@ -2,6 +2,7 @@
from pathlib import Path
from testcontainers.core.container import DockerContainer
+from testcontainers.core.transferable import Transferable
def test_garbage_collection_is_defensive():
@@ -46,3 +47,119 @@ 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)
+
+
+def test_copy_file_into_container_at_runtime(tmp_path: Path):
+ # Given
+ my_file = tmp_path / "my_file"
+ my_file.write_text("hello world")
+ destination_in_container = "/tmp/my_file"
+
+ with DockerContainer("bash", command="sleep infinity") as container:
+ # When
+ container.copy_into_container(my_file, 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_file_into_container_at_startup(tmp_path: Path):
+ # Given
+ my_file = tmp_path / "my_file"
+ my_file.write_text("hello world")
+ destination_in_container = "/tmp/my_file"
+
+ container = DockerContainer("bash", command="sleep infinity")
+ container.with_copy_into_container(my_file, 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_file_into_container_via_initializer(tmp_path: Path):
+ # Given
+ my_file = tmp_path / "my_file"
+ my_file.write_text("hello world")
+ destination_in_container = "/tmp/my_file"
+ transferables = [Transferable(my_file, destination_in_container)]
+
+ 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_bytes_to_container_at_runtime():
+ # Given
+ file_content = b"hello world"
+ destination_in_container = "/tmp/my_file"
+
+ with DockerContainer("bash", command="sleep infinity") as container:
+ # When
+ container.copy_into_container(file_content, destination_in_container)
+
+ # Then
+ result = container.exec(f"cat {destination_in_container}")
+
+ assert result.exit_code == 0
+ assert result.output == b"hello world"
+
+
+def test_copy_bytes_to_container_at_startup():
+ # Given
+ file_content = b"hello world"
+ destination_in_container = "/tmp/my_file"
+
+ container = DockerContainer("bash", command="sleep infinity")
+ container.with_copy_into_container(file_content, 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_bytes_to_container_via_initializer():
+ # Given
+ file_content = b"hello world"
+ destination_in_container = "/tmp/my_file"
+ transferables = [Transferable(file_content, destination_in_container)]
+
+ 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"