Add new function shell_tools.run

dev_tools/shell_tools.py

@@ -118,6 +118,53 @@ def abbreviate_command_arguments_after_switches(cmd: Tuple[str, ...]) -> Tuple[s
     return tuple(result)
 
 
+def run(
+    args: Union[str, List[str]],
+    *,
+    log_run_to_stderr: bool = True,
+    abbreviate_non_option_arguments: bool = False,
+    check: bool = True,
+    text: bool = True,
+    **subprocess_run_kwargs,
+) -> subprocess.CompletedProcess:
+    """Call subprocess.run with an option to log executed command to stderr.
+
+    Args:
+        args: The arguments for launching the process.  This may be a list
+            or a string.  The string type may need to be used with
+            ``shell=True`` to allow invocation as a shell command;
+            otherwise the string is used as a command name with no arguments.
+        log_run_to_stderr: Determines whether the fact that this command
+            was executed is logged to sys.stderr or not.
+        abbreviate_non_option_arguments: When logging to stderr, this cuts off
+            the potentially-huge tail of the command listing off e.g. hundreds
+            of file paths. No effect if log_run_to_stderr is not set.
+        check: Raise the CalledProcessError exception if this flag is
+            set and the process returns a non-zero exit code.  This sets
+            the default check argument for the `subprocess.run` to True.
+        text: Use text mode for the stdout and stderr streams from the
+            process.  This changes the default text argument to the
+            `subprocess.run` call to True.
+        **subprocess_run_kwargs: Arguments passed to `subprocess.run`.
+            See `subprocess.run` for a full detail of supported arguments.
+
+    Returns:
+        subprocess.CompletedProcess: The return value from `subprocess.run`.
+
+    Raises:
+        subprocess.CalledProcessError: The process returned a non-zero error
+            code and the check argument was set.
+    """
+    # setup our default for subprocess.run flag arguments
+    subprocess_run_kwargs.update(check=check, text=text)
+    if log_run_to_stderr:
+        cmd_desc: Tuple[str, ...] = (args,) if isinstance(args, str) else tuple(args)
+        if abbreviate_non_option_arguments:
+            cmd_desc = abbreviate_command_arguments_after_switches(cmd_desc)
+        print('run:', cmd_desc, file=sys.stderr)
+    return subprocess.run(args, **subprocess_run_kwargs)
+
+
 def run_cmd(
     *cmd: Optional[str],
     out: Optional[Union[TeeCapture, IO[str]]] = sys.stdout,

dev_tools/shell_tools_test.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import contextlib
+import io
 import subprocess
 
 import pytest
@@ -20,6 +22,10 @@
 from dev_tools.test_utils import only_on_posix
 
 
+def run(*args, **kwargs):
+    return shell_tools.run(*args, log_run_to_stderr=False, **kwargs)
+
+
 def run_cmd(*args, **kwargs):
     return shell_tools.run_cmd(*args, log_run_to_stderr=False, **kwargs)
 
@@ -28,6 +34,31 @@ def run_shell(*args, **kwargs):
     return shell_tools.run_shell(*args, log_run_to_stderr=False, **kwargs)
 
 
+@only_on_posix
+def test_run_raises_on_failure():
+    assert run('true').returncode == 0
+    with pytest.raises(subprocess.CalledProcessError):
+        run('false')
+    assert run('false', check=False).returncode == 1
+
+
+def test_run_returns_string_output():
+    result = run(['echo', 'hello', 'world'], capture_output=True)
+    assert result.stdout == 'hello world\n'
+
+
+def test_run_with_command_logging():
+    catch_stderr = io.StringIO()
+    kw = {'stdout': subprocess.DEVNULL}
+    with contextlib.redirect_stderr(catch_stderr):
+        shell_tools.run(['echo', '-n', 'a', 'b'], **kw)
+    assert catch_stderr.getvalue() == "run: ('echo', '-n', 'a', 'b')\n"
+    catch_stderr = io.StringIO()
+    with contextlib.redirect_stderr(catch_stderr):
+        shell_tools.run(['echo', '-n', 'a', 'b'], abbreviate_non_option_arguments=True, **kw)
+    assert catch_stderr.getvalue() == "run: ('echo', '-n', '[...]')\n"
+
+
 @only_on_posix
 def test_run_cmd_raise_on_fail():
     assert run_cmd('true') == (None, None, 0)