spec: added raise_subproc_error (xonsh#5494)

### Motivation

Add `spec.raise_subproc_error` to have an ability to use
`SpecModifierAlias` to manage the errors. Also this is the first step to
xonsh#4351.

Generally in scripts it's good to have RAISE_SUBPROC_ERROR=True to avoid
processing the error for every executed command. But in some cases (e.g.
`![]`) it's needed to avoid raising the error. To more elegant doing
this we can make an ability to create SpecModifier.

### After

```xsh
from xonsh.procs.specs import SpecModifierAlias
class SpecModifierNoErrAlias(SpecModifierAlias):
    def on_modifer_added(self, spec):
        spec.raise_subproc_error = False
aliases['noraise'] = SpecModifierNoErrAlias()

$RAISE_SUBPROC_ERROR = True

if ![noraise git pull]:
    git add --all
```

Cc xonsh#5443 

## For community
⬇️ **Please click the 👍 reaction instead of leaving a `+1` or 👍
comment**

---------

Co-authored-by: a <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

news/spec_raise_subproc_error.rst

@@ -0,0 +1,23 @@
+**Added:**
+
+* Added ``spec.raise_subproc_error`` for fine-tuning exceptions via ``SpecModifierAlias`` (#5494).
+
+**Changed:**
+
+* <news item>
+
+**Deprecated:**
+
+* <news item>
+
+**Removed:**
+
+* <news item>
+
+**Fixed:**
+
+* <news item>
+
+**Security:**
+
+* <news item>

tests/procs/test_specs.py

@@ -3,7 +3,7 @@
 import itertools
 import signal
 import sys
-from subprocess import Popen
+from subprocess import CalledProcessError, Popen
 
 import pytest
 
@@ -163,6 +163,51 @@ def test_interrupted_process_returncode(xonsh_session, captured, interactive):
  assert p.proc.returncode == -signal.SIGINT
 
 
+@skip_if_on_windows
+def test_proc_raise_subproc_error(xonsh_session):
+ xonsh_session.env["RAISE_SUBPROC_ERROR"] = False
+
+ specs = cmds_to_specs(cmd := [["ls"]], captured="stdout")
+ specs[-1].raise_subproc_error = True
+ exception = None
+ try:
+ (p := _run_command_pipeline(specs, cmd)).end()
+ assert p.proc.returncode == 0
+ except Exception as e:
+ exception = e
+ assert exception is None
+
+ specs = cmds_to_specs(cmd := [["ls", "nofile"]], captured="stdout")
+ specs[-1].raise_subproc_error = False
+ exception = None
+ try:
+ (p := _run_command_pipeline(specs, cmd)).end()
+ assert p.proc.returncode > 0
+ except Exception as e:
+ exception = e
+ assert exception is None
+
+ specs = cmds_to_specs(cmd := [["ls", "nofile"]], captured="stdout")
+ specs[-1].raise_subproc_error = True
+ exception = None
+ try:
+ (p := _run_command_pipeline(specs, cmd)).end()
+ except Exception as e:
+ assert p.proc.returncode > 0
+ exception = e
+ assert isinstance(exception, CalledProcessError)
+
+ xonsh_session.env["RAISE_SUBPROC_ERROR"] = True
+ specs = cmds_to_specs(cmd := [["ls", "nofile"]], captured="stdout")
+ exception = None
+ try:
+ (p := _run_command_pipeline(specs, cmd)).end()
+ except Exception as e:
+ assert p.proc.returncode > 0
+ exception = e
+ assert isinstance(exception, CalledProcessError)
+
+
 @skip_if_on_windows
 @pytest.mark.parametrize(
  "suspended_pipeline",

xonsh/procs/pipelines.py

@@ -627,14 +627,21 @@ def _raise_subproc_error(self):
  spec = self.spec
  rtn = self.returncode
 
- if rtn is None or rtn == 0 or not XSH.env.get("RAISE_SUBPROC_ERROR"):
+ if rtn is None or rtn == 0:
  return
 
- try:
- raise subprocess.CalledProcessError(rtn, spec.args, output=self.output)
- finally:
- # this is need to get a working terminal in interactive mode
- self._return_terminal()
+ raise_subproc_error = spec.raise_subproc_error
+ if callable(raise_subproc_error):
+ raise_subproc_error = raise_subproc_error(spec, self)
+ if raise_subproc_error is False:
+ return
+
+ if raise_subproc_error or XSH.env.get("RAISE_SUBPROC_ERROR", True):
+ try:
+ raise subprocess.CalledProcessError(rtn, spec.args, output=self.output)
+ finally:
+ # this is need to get a working terminal in interactive mode
+ self._return_terminal()
 
  #
  # Properties

xonsh/procs/specs.py

@@ -421,6 +421,7 @@ def __init__(
  self.stack = None
  self.spec_modifiers = [] # List of SpecModifierAlias objects that applied to spec.
  self.output_format = XSH.env.get("XONSH_SUBPROC_OUTPUT_FORMAT", "stream_lines")
+ self.raise_subproc_error = None # Spec-based $RAISE_SUBPROC_ERROR.
 
  def __str__(self):
  s = self.__class__.__name__ + "(" + str(self.cmd) + ", "