refactor: subprocess.run proxy function for better piping

src/cipy/action.py

@@ -1,7 +1,6 @@
 """Module containing basic Action definitions, which perform linear operations"""
 
 import pathlib
-import subprocess
 import tempfile
 
 from textwrap import dedent
@@ -90,20 +89,13 @@ class NodeScript(Action):
     @cipy.runner.ipc
     @cipy.runner.preamble
     def run(self, context: Context) -> Status:
-        try:
-            subprocess.run(["node", str(self.main)], check=True)
-            return Status.SUCCESS
-        except subprocess.CalledProcessError:
-            return Status.FAILURE
+        return cipy.runner.run(self, ["node", str(self.main)])
 
     @final
     def cleanup(self, context: Context) -> None:
         if self.post is None:
             return
-        try:
-            subprocess.run(["node", str(self.post)], check=True)
-        except subprocess.CalledProcessError:
-            pass
+        cipy.runner.run(self, ["node", str(self.post)])
 
 
 class Script(Action):
@@ -140,11 +132,7 @@ class Script(Action):
         with tempfile.NamedTemporaryFile(suffix=self.shell.extension()) as file:
             file.write(script.encode("utf-8"))
             file.flush()
-            try:
-                subprocess.run(self.shell.command(file.name), check=True)
-                return Status.SUCCESS
-            except subprocess.CalledProcessError:
-                return Status.FAILURE
+            return cipy.runner.run(self, self.shell.command(file.name))
 
 
 class Composite(Action):

src/cipy/runner.py

@@ -3,11 +3,13 @@ Common functions for setting up/tearing down environments for running an action.
 """
 
 import functools
+import io
 import os
+import subprocess
 import tempfile
 
 from contextlib import contextmanager
-from typing import Any, Callable, Iterator, TypeVar
+from typing import Any, Callable, Iterator, Literal, Sequence, TypeVar, overload
 
 from dotenv import dotenv_values
 
@@ -75,6 +77,51 @@ def ipc(func: Run[Action]) -> Run[Action]:
     return wrapper
 
 
+@overload
+def run(
+    self: Action, cmd: Sequence[str], *, pipe: Literal[False] = False
+) -> Status: ...
+
+
+@overload
+def run(
+    self: Action, cmd: Sequence[str], *, pipe: Literal[True] = True
+) -> tuple[Status, io.TextIOWrapper, io.TextIOWrapper]: ...
+
+
+def run(self: Action, cmd: Sequence[str], *, pipe: bool = False):
+    """
+    Proxy for subprocess.run which limits usage to two different cases:
+    1) Execute a command, piping STDOUT and STDERR to the logger
+    2) Execute a command, captureing STDOUT and STDERR to be processed by the caller
+
+    :param Action self: The calling action
+    :param Sequence[str] cmd: A tokenized commandline
+    :param bool pipe: Sets mode to LOG (false) or CAPTURE (true)
+
+    :return: (SUCCESS/FAILURE [, STDOUT, STDERR])
+    """
+    proc = subprocess.run(
+        cmd,
+        stdout=subprocess.PIPE,
+        # Unfortunately - there's no way to actually manipulate file
+        # descriptors to allow us to apply warning/error colors to
+        # proc.stderr without becoming unable to interleave stdout
+        # and stderr in chronological order.
+        stderr=subprocess.PIPE if pipe else subprocess.STDOUT,
+        check=False,
+    )
+
+    status = Status.SUCCESS if proc.returncode == 0 else Status.FAILURE
+    if not pipe:
+        for line in proc.stdout.splitlines():
+            self.logger.info(line.decode("utf-8"))
+
+        return status
+
+    return (status, proc.stdout, proc.stderr)
+
+
 @contextmanager
 def logging_group(
     self: Action, msg: str, *args: Any, ci_only: bool = False