Make Transferable simply a Union

Author: rhoban13

core/testcontainers/core/container.py

@@ -20,7 +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.transferable import Transferable, TransferSpec
 from testcontainers.core.utils import is_arm, setup_logger
 from testcontainers.core.wait_strategies import LogMessageWaitStrategy
 from testcontainers.core.waiting_utils import WaitStrategy, wait_container_is_ready
@@ -75,7 +75,7 @@ def __init__(
         network: Optional[Network] = None,
         network_aliases: Optional[list[str]] = None,
         _wait_strategy: Optional[WaitStrategy] = None,
-        transferables: Optional[list[Transferable]] = None,
+        transferables: Optional[list[TransferSpec]] = None,
         **kwargs: Any,
     ) -> None:
         self.env = env or {}
@@ -104,7 +104,11 @@ def __init__(
 
         self._kwargs = kwargs
         self._wait_strategy: Optional[WaitStrategy] = _wait_strategy
-        self._transferables: list[Transferable] = transferables or []
+
+        self._transferable_specs: list[TransferSpec] = []
+        if transferables:
+            for t in transferables:
+                self.with_copy_into_container(*t)
 
     def with_env(self, key: str, value: str) -> Self:
         self.env[key] = value
@@ -214,8 +218,8 @@ 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)
+        for t in self._transferable_specs:
+            self._transfer_into_container(*t)
 
         return self
 
@@ -309,24 +313,20 @@ def _configure(self) -> None:
         pass
 
     def with_copy_into_container(
-        self, file_content: Union[bytes, pathlib.Path], destination_in_container: str, mode: int = 0o644
+        self, transferable: Transferable, destination_in_container: str, mode: int = 0o644
     ) -> Self:
-        self._transferables.append(Transferable(file_content, destination_in_container, mode))
+        self._transferable_specs.append((transferable, 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 copy_into_container(self, transferable: Transferable, destination_in_container: str, mode: int = 0o644) -> None:
+        return self._transfer_into_container(transferable, 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):
-            assert source.is_file()  # Temporary, only copying file supported
-            file_content = source.read_bytes()
+    def _transfer_into_container(self, transferable: Transferable, destination_in_container: str, mode: int) -> None:
+        if isinstance(transferable, bytes):
+            file_content = transferable
+        elif isinstance(transferable, pathlib.Path):
+            assert transferable.is_file()  # Temporary, only copying file supported
+            file_content = transferable.read_bytes()
         else:
             raise TypeError("source must be bytes or PathLike")
 

core/testcontainers/core/transferable.py

@@ -1,14 +1,6 @@
-import dataclasses
 import pathlib
 from typing import Union
 
+Transferable = Union[bytes, pathlib.Path]
 
-@dataclasses.dataclass
-class Transferable:
-    """
-    Wrapper class enabling copying files into a container
-    """
-
-    source: Union[bytes, pathlib.Path]
-    destination_in_container: str
-    mode: int = 0o644
+TransferSpec = Union[tuple[Transferable, str], tuple[Transferable, str, int]]

core/tests/test_core.py

@@ -1,10 +1,9 @@
 import tempfile
 from pathlib import Path
-from typing import Union
 
 import pytest
 from testcontainers.core.container import DockerContainer
-from testcontainers.core.transferable import Transferable
+from testcontainers.core.transferable import Transferable, TransferSpec
 
 
 def test_garbage_collection_is_defensive():
@@ -51,8 +50,8 @@ def test_docker_container_with_env_file():
             print(output)
 
 
-@pytest.fixture(name="copy_source", params=(bytes, Path))
-def copy_source_fixture(request, tmp_path: Path):
+@pytest.fixture(name="transferable", params=(bytes, Path))
+def copy_sources_fixture(request, tmp_path: Path):
     """
     Provide source argument for tests of copy_into_container
     """
@@ -66,26 +65,26 @@ def copy_source_fixture(request, tmp_path: Path):
     pytest.fail("Invalid type")
 
 
-def test_copy_into_container_at_runtime(copy_source: Union[bytes, Path]):
+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(copy_source, destination_in_container)
+        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(copy_source: Union[bytes, Path]):
+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(copy_source, destination_in_container)
+    container.with_copy_into_container(transferable, destination_in_container)
 
     with container:
         # When
@@ -96,10 +95,10 @@ def test_copy_into_container_at_startup(copy_source: Union[bytes, Path]):
     assert result.output == b"hello world"
 
 
-def test_copy_into_container_via_initializer(copy_source: Union[bytes, Path]):
+def test_copy_into_container_via_initializer(transferable: Transferable):
     # Given
     destination_in_container = "/tmp/my_file"
-    transferables = [Transferable(copy_source, destination_in_container)]
+    transferables: list[TransferSpec] = [(transferable, destination_in_container, 0o644)]
 
     with DockerContainer("bash", command="sleep infinity", transferables=transferables) as container:
         # When