Implement & test copying a directory

Author: rhoban13

core/testcontainers/core/container.py

@@ -323,13 +323,20 @@ def copy_into_container(self, transferable: Transferable, destination_in_contain
 
     def _transfer_into_container(self, transferable: Transferable, destination_in_container: str, mode: int) -> None:
         if isinstance(transferable, bytes):
-            file_content = transferable
+            self._transfer_file_content_into_container(transferable, destination_in_container, mode)
         elif isinstance(transferable, pathlib.Path):
-            assert transferable.is_file()  # Temporary, only copying file supported
-            file_content = transferable.read_bytes()
+            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)
@@ -341,6 +348,20 @@ def _transfer_into_container(self, transferable: Transferable, destination_in_co
         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
+
+        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
+
     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)

core/tests/test_core.py

@@ -125,3 +125,32 @@ def test_copy_file_from_container(tmp_path: Path):
     # 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"