[elbe-devel] [PATCH 2/5] elbepack: shellhelper: introduce run() function

[Thomas Weißschuh]

the run() function is a thin wrapper around subprocess.run(),
with the following changes:

* It defaults to check=True for more correctness
* It accepts shellper.ELBE_LOGGING for stdin and stdout to forward these
  through the elbe log

In contrast with the original shellhelper functions, this has
many advantages:

It is much more flexible for callers to achieve the exact behaviour they
require. Specifying stdout, stderr, cwd, check, etc; capturing output and
validating return codes are all available independently from each other.

Executing through a shell needs opt-in as that is more error-prone.

Signed-off-by: Thomas Weißschuh <thomas.weissschuh at linutronix.de>
---
 elbepack/shellhelper.py | 69 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 69 insertions(+)

diff --git a/elbepack/shellhelper.py b/elbepack/shellhelper.py
index dd7312c913bc..d04699d9977c 100644
--- a/elbepack/shellhelper.py
+++ b/elbepack/shellhelper.py
@@ -3,6 +3,7 @@
 # SPDX-FileCopyrightText: 2014-2017 Linutronix GmbH
 # SPDX-FileCopyrightText: 2014 Ferdinand Schwenk <ferdinand.schwenk at emtrion.de>
 
+import contextlib
 import logging
 import os
 import shlex
@@ -11,6 +12,12 @@ import subprocess
 from elbepack.log import async_logging_ctx
 
 
+"""
+Forward to elbe logging system.
+"""
+ELBE_LOGGING = object()
+
+
 def _is_shell_cmd(cmd):
     return isinstance(cmd, str)
 
@@ -22,6 +29,68 @@ def _log_cmd(cmd):
         return shlex.join(map(os.fspath, cmd))
 
 
+def run(cmd, /, *, check=True, log_cmd=None, **kwargs):
+    """
+    Like subprocess.run() but
+     * defaults to check=True
+     * logs the executed command
+     * accepts ELBE_LOGGING for stdout and stderr
+
+    --
+
+    Let's quiet the loggers
+
+    >>> import os
+    >>> import sys
+    >>> from elbepack.log import open_logging
+    >>> open_logging({"files":os.devnull})
+
+    >>> run(['echo', 'ELBE'])
+    CompletedProcess(args=['echo', 'ELBE'], returncode=0)
+
+    >>> run(['echo', 'ELBE'], capture_output=True)
+    CompletedProcess(args=['echo', 'ELBE'], returncode=0, stdout=b'ELBE\\n', stderr=b'')
+
+    >>> run(['false']) # doctest: +ELLIPSIS
+    Traceback (most recent call last):
+    ...
+    subprocess.CalledProcessError: ...
+
+    >>> run('false', check=False).returncode
+    1
+
+    >>> run(['cat', '-'], input=b'ELBE', capture_output=True).stdout
+    b'ELBE'
+
+    >>> run(['echo', 'ELBE'], stdout=ELBE_LOGGING)
+    CompletedProcess(args=['echo', 'ELBE'], returncode=0)
+
+    Let's redirect the loggers to current stdout
+
+    >>> from elbepack.log import open_logging
+    >>> open_logging({"streams":sys.stdout})
+
+    >>> run(['echo', 'ELBE'], stdout=ELBE_LOGGING)
+    [CMD] echo ELBE
+    ELBE
+    ELBE
+    CompletedProcess(args=['echo', 'ELBE'], returncode=0)
+    """
+    stdout = kwargs.pop('stdout', None)
+    stderr = kwargs.pop('stderr', None)
+
+    with contextlib.ExitStack() as stack:
+        if stdout is ELBE_LOGGING or stderr is ELBE_LOGGING:
+            log_fd = stack.enter_context(async_logging_ctx())
+            if stdout is ELBE_LOGGING:
+                stdout = log_fd
+            if stderr is ELBE_LOGGING:
+                stderr = log_fd
+
+        logging.info(log_cmd or _log_cmd(cmd), extra={'context': '[CMD] '})
+        return subprocess.run(cmd, stdout=stdout, stderr=stderr, check=check, **kwargs)
+
+
 def do(cmd, /, *, check=True, env_add=None, log_cmd=None, **kwargs):
     """do() - Execute cmd in a shell and redirect outputs to logging.
 

-- 
2.45.1