diff --git a/poetry.lock b/poetry.lock index c993a8c..077bd7e 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand. [[package]] name = "appnope" @@ -51,6 +51,23 @@ files = [ {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"}, ] +[[package]] +name = "colorlog" +version = "6.8.2" +description = "Add colours to the output of Python's logging module." +optional = false +python-versions = ">=3.6" +files = [ + {file = "colorlog-6.8.2-py3-none-any.whl", hash = "sha256:4dcbb62368e2800cb3c5abd348da7e53f6c362dda502ec27c560b2e58a66bd33"}, + {file = "colorlog-6.8.2.tar.gz", hash = "sha256:3e3e079a41feb5a1b64f978b5ea4f46040a94f11f0e8bbb8261e3dbbeca64d44"}, +] + +[package.dependencies] +colorama = {version = "*", markers = "sys_platform == \"win32\""} + +[package.extras] +development = ["black", "flake8", "mypy", "pytest", "types-colorama"] + [[package]] name = "coverage" version = "7.4.4" @@ -416,6 +433,36 @@ pytest = ">=4.6" [package.extras] testing = ["fields", "hunter", "process-tests", "pytest-xdist", "virtualenv"] +[[package]] +name = "pytest-mock" +version = "3.14.0" +description = "Thin-wrapper around the mock package for easier use with pytest" +optional = false +python-versions = ">=3.8" +files = [ + {file = "pytest-mock-3.14.0.tar.gz", hash = "sha256:2719255a1efeceadbc056d6bf3df3d1c5015530fb40cf347c0f9afac88410bd0"}, + {file = "pytest_mock-3.14.0-py3-none-any.whl", hash = "sha256:0b72c38033392a5f4621342fe11e9219ac11ec9d375f8e2a0c164539e0d70f6f"}, +] + +[package.dependencies] +pytest = ">=6.2.5" + +[package.extras] +dev = ["pre-commit", "pytest-asyncio", "tox"] + +[[package]] +name = "python-simple-logger" +version = "1.0.19" +description = "A simple logger for python" +optional = false +python-versions = "<4.0,>=3.8" +files = [ + {file = "python_simple_logger-1.0.19.tar.gz", hash = "sha256:09e012b7eae28d22cd00725576aac0ec06f283838a57888d4d23b8935fc8092e"}, +] + +[package.dependencies] +colorlog = ">=6.7.0,<7.0.0" + [[package]] name = "six" version = "1.16.0" @@ -497,4 +544,4 @@ files = [ [metadata] lock-version = "2.0" python-versions = "^3.8" -content-hash = "e5320c14eaaa09de898201c3c6abdc0b9a18e901ba754b864ba798ffead0856a" +content-hash = "364f55a18d6c17d6ae0611aa8eaeb3be755e74bc4c5b7db3c79302cfe089d974" diff --git a/pyhelper_utils/shell.py b/pyhelper_utils/shell.py new file mode 100644 index 0000000..53b9afe --- /dev/null +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index d09c867..bea847e 100644 --- a/pyproject.toml +++ b/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\\d+)!)?(?P\\d+(\\.\\d+)*)" diff --git a/tests/test_shell.py b/tests/test_shell.py new file mode 100644 index 0000000..5c60ddc --- /dev/null +++ b/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)