EVA-3438 - Allow command to output a variable log level

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
 

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)
 

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]
+
+