feat: run_command log error before raising

pyaptly/util.py

@@ -3,12 +3,14 @@
 import logging
 import os
 import subprocess
+import traceback
 from pathlib import Path
-
-from colorama import Fore, init
 from subprocess import PIPE, CalledProcessError  # noqa: F401
+from tempfile import NamedTemporaryFile
 from typing import Optional, Sequence
 
+from colorama import Fore, init
+
 _DEFAULT_KEYSERVER: str = "hkps://keys.openpgp.org"
 _PYTEST_KEYSERVER: Optional[str] = None
 
@@ -28,6 +30,13 @@
 lg = logging.getLogger(__name__)
 
 
+def write_traceback():
+    with NamedTemporaryFile("w", delete=False) as tmp:
+        tmp.write(traceback.format_exc())
+        tmp.close()
+    return tmp.name
+
+
 def isatty():
     global _isatty_cache
     if _isatty_cache is None:
@@ -75,27 +84,47 @@ def run_command(
     """
     added_stdout = False
     added_stderr = False
-    # TODO assert PIPE or None
     if "stdout" not in kwargs:
         kwargs["stdout"] = PIPE
         added_stdout = True
+    else:
+        assert kwargs["stdout"] in (PIPE, None)
     if "stderr" not in kwargs:
         kwargs["stderr"] = PIPE
         added_stderr = True
+    else:
+        assert kwargs["stdout"] in (PIPE, None)  # pragma: no cover
+    # If we want to log stdout/err before raising CalledProcessError we have to
+    # check ourselves
+    check = False
+    if "check" in kwargs:
+        check = kwargs["check"]
+        del kwargs["check"]
     result = None
+    tb = ""
     if decode and "encoding" not in kwargs:
         kwargs["encoding"] = "UTF-8"
     try:
         result = subprocess.run(cmd_args, **kwargs)
+        if check and result.returncode:
+            raise CalledProcessError(
+                result.returncode,
+                result.args,
+                output=result.stdout,
+                stderr=result.stderr,
+            )
+    except Exception:
+        tb = write_traceback()
+        raise
     finally:
         if result:
-            log_msg = format_run_result(result, result.returncode)
+            log_msg = format_run_result(result, result.returncode, tb)
             if result.returncode == 0:
                 lg.info(log_msg)
             else:
                 if not hide_error or lg.root.level <= 20:
                     lg.error(log_msg)
-            # Do not change returned result by debug mode
+            # This function should not alter the returned result
             if added_stdout:
                 delattr(result, "stdout")
             if added_stderr:
@@ -132,7 +161,7 @@ def indent_out(output: bytes | str) -> str:
     return "\n".join(result)
 
 
-def format_run_result(result: subprocess.CompletedProcess, returncode: int):
+def format_run_result(result: subprocess.CompletedProcess, returncode: int, tb: str):
     """Format a CompletedProcess result log."""
     color_begin = ""
     color_end = ""