add run_command (#6)

* add run_command

* fix minor things

* review comments addressed

* Update pyhelper_utils/shell.py

Co-authored-by: Meni Yakove <441263+myakove@users.noreply.github.com>

* remove verbose from tox

* address review comments

* fix tox failure

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: Meni Yakove <441263+myakove@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

pyhelper_utils/shell.py

@@ -0,0 +1,71 @@
+import subprocess
+
+from simple_logger.logger import get_logger
+
+LOGGER = get_logger(name=__name__)
+
+TIMEOUT_30MIN = 30 * 60
+
+
+def run_command(
+    command: list,
+    verify_stderr: bool = True,
+    shell: bool = False,
+    timeout: int = None,
+    capture_output: bool = True,
+    check: bool = True,
+    hide_log_command: bool = False,
+    **kwargs,
+) -> tuple:
+    """
+    Run command locally.
+
+    Args:
+        command (list): Command to run
+        verify_stderr (bool, default True): Check command stderr
+        shell (bool, default False): run subprocess with shell toggle
+        timeout (int, optional): Command wait timeout
+        capture_output (bool, default False): Capture command output
+        check (bool, default True):  If check is True and the exit code was non-zero, it raises a
+            CalledProcessError
+        hide_log_command (bool, default False): If hide_log_command is True and check will be set to False,
+            CalledProcessError will not get raise and command will not be printed.
+
+    Returns:
+        tuple: True, out if command succeeded, False, err otherwise.
+
+    Raises:
+        CalledProcessError: when check is True and command execution fails
+    """
+    command_for_log = ["Hide", "By", "User"] if hide_log_command else command
+
+    LOGGER.info(f"Running {' '.join(command_for_log)} command")
+
+    # when hide_log_command is set to True, check should be set to False to avoid logging sensitive data in
+    # the exception
+    sub_process = subprocess.run(
+        command,
+        capture_output=capture_output,
+        check=check if not hide_log_command else False,
+        shell=shell,
+        text=True,
+        timeout=timeout,
+        **kwargs,
+    )
+    out_decoded = sub_process.stdout
+    err_decoded = sub_process.stderr
+
+    error_msg = (
+        f"Failed to run {command_for_log}. rc: {sub_process.returncode}, out: {out_decoded}, error: {err_decoded}"
+    )
+
+    if sub_process.returncode != 0:
+        LOGGER.error(error_msg)
+        return False, out_decoded, err_decoded
+
+    # From this point and onwards we are guaranteed that sub_process.returncode == 0
+    if err_decoded and verify_stderr:
+        LOGGER.error(error_msg)
+        return False, out_decoded, err_decoded
+
+    return True, out_decoded, err_decoded

pyproject.toml

@@ -39,16 +39,19 @@ classifiers = [
 
 [tool.poetry.dependencies]
 python = "^3.8"
+python-simple-logger = "^1.0.19"
 
 [tool.poetry.group.tests.dependencies]
 pytest = "^8.1.1"
 pytest-cov = "^5.0.0"
+pytest-mock = "^3.14.0"
 
 
 [tool.poetry.group.dev.dependencies]
 ipdb = "^0.13.13"
 ipython = "*"
 
+
 [tool.poetry-dynamic-versioning]
 enable = true
 pattern = "((?P<epoch>\\d+)!)?(?P<base>\\d+(\\.\\d+)*)"

tests/test_shell.py

@@ -0,0 +1,37 @@
+import shlex
+from subprocess import CalledProcessError
+import pytest
+import subprocess
+from pyhelper_utils.shell import run_command
+
+ERROR_MESSAGE = "Expected value {expected}, actual value {actual}"
+SUCCESSFUL_MESSAGE = "worked"
+FAILURE_MESSAGE = "No such file"
+
+
+def test_run_command_return_true():
+    rc, out, error = run_command(command=shlex.split(f"echo '{SUCCESSFUL_MESSAGE}'"), check=False)
+    assert rc, ERROR_MESSAGE.format(expected=True, actual=rc)
+    assert not error, ERROR_MESSAGE.format(expected="", actual="error")
+    assert SUCCESSFUL_MESSAGE in out, ERROR_MESSAGE.format(expected=SUCCESSFUL_MESSAGE, actual=out)
+
+
+def test_run_command_return_false():
+    rc, _, _ = run_command(command=shlex.split("false"), check=False)
+    assert not rc, ERROR_MESSAGE.format(expected=False, actual=rc)
+
+
+def test_run_command_no_verify_raises_exception():
+    with pytest.raises(CalledProcessError):
+        run_command(command=shlex.split("false"), check=True, verify_stderr=False)
+
+
+def test_run_command_error(mocker):
+    mocker.patch(
+        "pyhelper_utils.shell.subprocess.run",
+        return_value=subprocess.CompletedProcess(args=None, stderr=FAILURE_MESSAGE, returncode=0, stdout=""),
+    )
+    rc, out, error = run_command(command=shlex.split("true"), capture_output=False, check=False, shell=True)
+    assert not rc, ERROR_MESSAGE.format(expected=False, actual=rc)
+    assert FAILURE_MESSAGE in error, ERROR_MESSAGE.format(expected=FAILURE_MESSAGE, actual="error")
+    assert not out, ERROR_MESSAGE.format(expected="", actual=out)