From 44dd8c6be9d7a260f79486177bdf8d8088bb2c7e Mon Sep 17 00:00:00 2001 From: Timothee Cezard Date: Thu, 22 Feb 2024 15:23:37 +0000 Subject: [PATCH] EVA-3438 - Allow command to output a variable log level (#51) * Allow command to output a variable log level * Revert stderr to output to ERROR by default * Add tests --- ebi_eva_common_pyutils/command_utils.py | 15 +++---- ebi_eva_common_pyutils/logger.py | 3 ++ tests/common/test_command_utils.py | 56 +++++++++++++++++++++++++ 3 files changed, 67 insertions(+), 7 deletions(-) create mode 100644 tests/common/test_command_utils.py diff --git a/ebi_eva_common_pyutils/command_utils.py b/ebi_eva_common_pyutils/command_utils.py index 1ff1445..a8952df 100644 --- a/ebi_eva_common_pyutils/command_utils.py +++ b/ebi_eva_common_pyutils/command_utils.py @@ -11,7 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. - +import logging import subprocess from ebi_eva_common_pyutils.logger import logging_config as log_cfg @@ -20,11 +20,12 @@ def run_command_with_output(command_description, command, return_process_output=False, - log_error_stream_to_output=False): + log_error_stream_to_output=False, stdout_log_level=logging.INFO, + stderr_log_level=logging.ERROR): process_output = "" - logger.info("Starting process: " + command_description) - logger.info("Running command: " + command) + logger.log(stdout_log_level, "Starting process: " + command_description) + logger.log(stdout_log_level, "Running command: " + command) stdout = subprocess.PIPE # Some lame utilities like mongodump and mongorestore output non-error messages to error stream @@ -35,18 +36,18 @@ def run_command_with_output(command_description, command, return_process_output= shell=True) as process: for line in iter(process.stdout.readline, ''): line = str(line).rstrip() - logger.info(line) + logger.log(stdout_log_level, line) if return_process_output: process_output += line + "\n" if not log_error_stream_to_output: for line in iter(process.stderr.readline, ''): line = str(line).rstrip() - logger.error(line) + logger.log(stderr_log_level, line) if process.returncode != 0: logger.error(command_description + " failed! Refer to the error messages for details.") raise subprocess.CalledProcessError(process.returncode, process.args) else: - logger.info(command_description + " - completed successfully") + logger.log(stdout_log_level, command_description + " - completed successfully") if return_process_output: return process_output diff --git a/ebi_eva_common_pyutils/logger.py b/ebi_eva_common_pyutils/logger.py index 2e441f0..126ec1d 100644 --- a/ebi_eva_common_pyutils/logger.py +++ b/ebi_eva_common_pyutils/logger.py @@ -127,6 +127,9 @@ class AppLogger: """ log_cfg = logging_config + def log(self, level, msg, *args, **kwargs): + self._logger.log(level, msg, *args, **kwargs) + def debug(self, msg, *args): self._logger.debug(msg, *args) diff --git a/tests/common/test_command_utils.py b/tests/common/test_command_utils.py new file mode 100644 index 0000000..71ee11d --- /dev/null +++ b/tests/common/test_command_utils.py @@ -0,0 +1,56 @@ +import logging +import os +import shutil +import ebi_eva_common_pyutils.command_utils +from ebi_eva_common_pyutils import command_utils +from ebi_eva_common_pyutils.command_utils import run_command_with_output +from tests.test_common import TestCommon + + +def touch(name): + open(name, 'w').close() + + +class TestCommand(TestCommon): + + def setUp(self) -> None: + # Create a directory with files + self.test_dir = os.path.join(self.resources_folder, 'test_commmands') + os.makedirs(self.test_dir) + for i in range(1, 10): + touch(os.path.join(self.test_dir, f'file_{i}')) + + def tearDown(self) -> None: + shutil.rmtree(self.test_dir) + + def test_run_command_with_output(self): + expected_output = 'file_1\nfile_2\nfile_3\nfile_4\nfile_5\nfile_6\nfile_7\nfile_8\nfile_9\n' + content = run_command_with_output('Run list command', f'ls {self.test_dir}', return_process_output=True) + assert expected_output == content + + def test_run_command_without_output_default_log(self): + expected_output = [ + 'Starting process: Run list command', + f'Running command: ls {self.test_dir}', + 'file_1', 'file_2', 'file_3', 'file_4', 'file_5', 'file_6', 'file_7', 'file_8', 'file_9', + 'Run list command - completed successfully' + ] + + with self.assertLogs(command_utils.__name__, level=logging.DEBUG) as lc: + run_command_with_output('Run list command', f'ls {self.test_dir}') + assert lc.output == ['INFO:ebi_eva_common_pyutils.command_utils:' + e for e in expected_output] + + def test_run_command_without_output_debug_log(self): + expected_output = [ + 'Starting process: Run list command', + f'Running command: ls {self.test_dir}', + 'file_1', 'file_2', 'file_3', 'file_4', 'file_5', 'file_6', 'file_7', 'file_8', 'file_9', + 'Run list command - completed successfully' + ] + + with self.assertLogs(command_utils.__name__, level=logging.DEBUG) as lc: + run_command_with_output('Run list command', f'ls {self.test_dir}', stderr_log_level=logging.DEBUG, + stdout_log_level=logging.DEBUG) + assert lc.output == ['DEBUG:ebi_eva_common_pyutils.command_utils:' + e for e in expected_output] + +