Implemented run method and added CompletedProcess (inspired by subprocess.run)

src/dockerblade/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
 from . import exceptions
 from .files import FileSystem
-from .shell import Shell, ShellFactory
+from .shell import Shell, ShellFactory, CompletedProcess
 from .stopwatch import Stopwatch

src/dockerblade/shell.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
-__all__ = ('Shell', 'ShellFactory')
+__all__ = ('Shell', 'ShellFactory', 'CompletedProcess')
 
-from typing import Tuple, Optional
+from typing import Tuple, Optional, Generic, TypeVar
 import shlex
 
 from loguru import logger
@@ -11,6 +11,29 @@
 
 from .stopwatch import Stopwatch
 
+T = TypeVar('T', str, bytes)
+
+
+@attr.s(auto_attribs=True, frozen=True)
+class CompletedProcess(Generic[T]):
+    """Stores the result of a completed process.
+
+    Attributes
+    ----------
+    args: str
+        The arguments that were used to launch the process.
+    returncode: int
+        The returncode that was produced by the process.
+    duration: float
+        The number of seconds taken to complete the process.
+    output: T, optional
+        The output, if any, that was produced by the process.
+    """
+    args: str
+    returncode: int
+    duration: float
+    output: Optional[T]
+
 
 @attr.s(eq=False, hash=False)
 class Shell:
@@ -58,36 +81,34 @@ def environ(self, var: str) -> str:
         """
         raise NotImplementedError
 
-    def execute(self,
-                command: str,
-                *,
-                context: str = '/'
-                ) -> Tuple[int, str, float]:
+    def run(self,
+            args: str,
+            *,
+            context: str = '/'
+            ) -> CompletedProcess:
         """Executes a given command and blocks until its completion.
 
         Returns
         -------
-        Tuple[int, str, float]
-            The return code, output, and wall-clock running time of the
-            execution, measured in seconds.
+        CompletedProcess
+            A summary of the outcome of the command execution.
         """
-        logger.debug(f"executing command: {command}")
+        logger.debug(f"executing command: {args}")
         container = self._container
-        command_instrumented = self._instrument(command)
+        args_instrumented = self._instrument(args)
 
         with Stopwatch() as timer:
             retcode, output = container.exec_run(
-                command_instrumented,
+                args_instrumented,
                 workdir=context)
 
-        duration = timer.duration
         output = output.decode('utf-8').rstrip('\n')
-        logger.debug("executed command [{command}] "
-                     "(retcode: {retcode}; time: {time:.3f} s)"
-                     "\n{output}",
-                     command=command, retcode=retcode, time=duration,
-                     output=output)
-        return retcode, output, duration
+        result = CompletedProcess(args=args,
+                                  returncode=retcode,
+                                  duration=timer.duration,
+                                  output=output)
+        logger.debug(f"executed command: {result}")
+        return result
 
 
 @attr.s(slots=True, frozen=True)

test/test_shell.py

@@ -26,6 +26,6 @@ def shell_factory():
 
 def test_hello_world(alpine_310, shell_factory):
     shell = shell_factory.build(alpine_310.id, '/bin/sh')
-    retcode, output, duration = shell.execute("echo 'hello world'")
-    assert retcode == 0
-    assert output == 'hello world'
+    result = shell.run("echo 'hello world'")
+    assert result.returncode == 0
+    assert result.output == 'hello world'