Move execute_command to separate class

pysqa/utils/basic.py

@@ -5,10 +5,10 @@
 import importlib
 from jinja2 import Template
 import os
-import subprocess
 import re
 import pandas
 from pysqa.utils.queues import Queues
+from pysqa.utils.execute import execute_command
 
 
 class BasisQueueAdapter(object):
@@ -17,8 +17,10 @@ class BasisQueueAdapter(object):
     locally.
 
     Args:
+        config (dict):
         directory (str): directory containing the queue.yaml files as well as corresponding jinja2 templates for the
                          individual queues.
+        execute_command(funct):
 
     Attributes:
 
@@ -39,7 +41,7 @@ class BasisQueueAdapter(object):
             Queues available for auto completion QueueAdapter().queues.<queue name> returns the queue name.
     """
 
-    def __init__(self, config, directory="~/.queues"):
+    def __init__(self, config, directory="~/.queues", execute_command=execute_command):
         self._config = config
         self._fill_queue_dict(queue_lst_dict=self._config["queues"])
         self._load_templates(queue_lst_dict=self._config["queues"], directory=directory)
@@ -71,6 +73,7 @@ def __init__(self, config, directory="~/.queues"):
         self._queues = Queues(self.queue_list)
         self._remote_flag = False
         self._ssh_delete_file_on_remote = True
+        self._execute_command_function = execute_command
 
     @property
     def ssh_delete_file_on_remote(self):
@@ -405,18 +408,13 @@ def _job_submission_template(
             **kwargs
         )
 
-    @staticmethod
-    def _get_user():
-        """
-
-        Returns:
-            str:
-        """
-        return getpass.getuser()
-
-    @staticmethod
     def _execute_command(
-        commands, working_directory=None, split_output=True, shell=False
+        self,
+        commands,
+        working_directory=None,
+        split_output=True,
+        shell=False,
+        error_filename="pysqa.err",
     ):
         """
 
@@ -425,28 +423,27 @@ def _execute_command(
             working_directory (str):
             split_output (bool):
             shell (bool):
+            error_filename (str):
 
         Returns:
             str:
         """
-        if shell and isinstance(commands, list):
-            commands = " ".join(commands)
-        try:
-            out = subprocess.check_output(
-                commands,
-                cwd=working_directory,
-                stderr=subprocess.STDOUT,
-                universal_newlines=True,
-                shell=not isinstance(commands, list),
-            )
-        except subprocess.CalledProcessError as e:
-            with open(os.path.join(working_directory, "pysqa.err"), "w") as f:
-                print(e.stdout, file=f)
-            out = None
-        if out is not None and split_output:
-            return out.split("\n")
-        else:
-            return out
+        return self._execute_command_function(
+            commands=commands,
+            working_directory=working_directory,
+            split_output=split_output,
+            shell=shell,
+            error_filename=error_filename,
+        )
+
+    @staticmethod
+    def _get_user():
+        """
+
+        Returns:
+            str:
+        """
+        return getpass.getuser()
 
     @staticmethod
     def _fill_queue_dict(queue_lst_dict):

pysqa/utils/execute.py

@@ -0,0 +1,42 @@
+import os
+import subprocess
+
+
+def execute_command(
+    commands,
+    working_directory=None,
+    split_output=True,
+    shell=False,
+    error_filename="pysqa.err",
+):
+    """
+    A wrapper around the subprocess.check_output function.
+
+    Args:
+        commands (list/str): These commands are executed on the command line
+        working_directory (str/None): The directory where the command is executed
+        split_output (bool): Boolean flag to split newlines in the output
+        shell (bool): Additional switch to convert list of commands to one string
+        error_filename (str): In case the execution fails the output is written to this file
+
+    Returns:
+        str/list: output of the shell command either as string or as a list of strings
+    """
+    if shell and isinstance(commands, list):
+        commands = " ".join(commands)
+    try:
+        out = subprocess.check_output(
+            commands,
+            cwd=working_directory,
+            stderr=subprocess.STDOUT,
+            universal_newlines=True,
+            shell=not isinstance(commands, list),
+        )
+    except subprocess.CalledProcessError as e:
+        with open(os.path.join(working_directory, error_filename), "w") as f:
+            print(e.stdout, file=f)
+        out = None
+    if out is not None and split_output:
+        return out.split("\n")
+    else:
+        return out

tests/test_execute_command.py

@@ -0,0 +1,59 @@
+import os
+import unittest
+from pysqa.utils.execute import execute_command
+
+
+class TestExecuteCommand(unittest.TestCase):
+    def test_commands_as_lst(self):
+        output = execute_command(
+            commands=["echo", "hello"],
+            working_directory=".",
+            split_output=True,
+            shell=False,
+            error_filename="pysqa.err"
+        )
+        self.assertEqual(output, ['hello', ''])
+
+    def test_commands_as_lst_no_split(self):
+        output = execute_command(
+            commands=["echo", "hello"],
+            working_directory=".",
+            split_output=False,
+            shell=False,
+            error_filename="pysqa.err"
+        )
+        self.assertEqual(output, 'hello\n')
+
+    def test_commands_as_lst_shell_true(self):
+        output = execute_command(
+            commands=["echo", "hello"],
+            working_directory=".",
+            split_output=True,
+            shell=True,
+            error_filename="pysqa.err"
+        )
+        self.assertEqual(output, ['hello', ''])
+
+    def test_commands_as_str(self):
+        output = execute_command(
+            commands="echo hello",
+            working_directory=".",
+            split_output=True,
+            shell=False,
+            error_filename="pysqa.err"
+        )
+        self.assertEqual(output, ['hello', ''])
+
+    def test_commands_fails(self):
+        output = execute_command(
+            commands="exit 1",
+            working_directory=".",
+            split_output=True,
+            shell=False,
+            error_filename="pysqa_fails.err"
+        )
+        self.assertIsNone(output)
+        with open("pysqa_fails.err") as f:
+            error = f.readlines()
+        self.assertEqual(error, ["\n"])
+        os.remove("pysqa_fails.err")