basic transferable demo

Author: alexanderankin

core/testcontainers/core/docker_client.py

@@ -16,7 +16,7 @@
 from docker.models.containers import Container, ContainerCollection
 import functools as ft
 import os
-from typing import List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 import urllib
 
 from .utils import default_gateway_ip, inside_container, setup_logger
@@ -65,7 +65,7 @@ def port(self, container_id: str, port: int) -> int:
                                   'not available')
         return port_mappings[0]["HostPort"]
 
-    def get_container(self, container_id: str) -> Container:
+    def get_container(self, container_id: str) -> Dict[str, Any]:
         """
         Get the container with a given identifier.
         """
@@ -110,3 +110,10 @@ def host(self) -> str:
                 if ip_address:
                     return ip_address
         return "localhost"
+
+    def status(self, container_id: str) -> str:
+        """
+        Get the status of a container. (running, stopped, etc)
+        """
+        container = self.get_container(container_id)
+        return container['State']

core/testcontainers/core/transferable.py

@@ -105,3 +105,11 @@ def wait_for_logs(container: "DockerContainer", predicate: Union[Callable, str],
             raise TimeoutError(f"Container did not emit logs satisfying predicate in {timeout:.3f} "
                                "seconds")
         time.sleep(interval)
+
+
+def wait_for_status(container: "DockerContainer", status: str = 'running') -> None:
+    start = time.time()
+    while config.TIMEOUT > time.time() - start:
+        if status == container.get_docker_client().status(container.get_wrapped_container().id):
+            break
+        time.sleep(config.SLEEP_TIME)

core/testcontainers/core/waiting_utils.py

@@ -0,0 +1,42 @@
+from pathlib import Path
+from tarfile import open as tarfile_open, TarInfo
+from io import BytesIO
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.docker_client import DockerClient
+from testcontainers.core.transferable import *
+from testcontainers.core.waiting_utils import *
+from docker.models.containers import Container
+
+
+def test_docker_custom_image():
+    alpine = DockerContainer('alpine')
+    alpine.with_command('sleep 1000000')
+    alpine.start()
+    wait_for_status(alpine)
+
+    def put_file(dest: str, content: str):
+        alpine.exec(f'touch "{dest}"')
+
+        # setup tar info
+        content_encode = content.encode()
+        content_bytes = BytesIO(content_encode)
+        into = TarInfo(name=Path(dest).name)
+        # into = TarInfo(name=dest)
+        into.size = len(content_encode)
+
+        # setup tar
+        output_bytes = BytesIO()
+        with tarfile_open(fileobj=output_bytes, mode='w:gz') as tar:
+            tar.addfile(into, content_bytes)
+
+        # create tar
+        data = output_bytes.getvalue()
+
+        # peform docker api call
+        alpine.get_wrapped_container() \
+            .put_archive(str(Path(dest).parent), data)
+
+    put_file('/etc/stuff', 'content')
+
+    print(alpine.exec('cat /etc/stuff'))