Fixes Popen output to use universal newlines.

maestrowf/datastructures/environment/gitdependency.py

@@ -32,9 +32,9 @@
 import logging
 import os
 import re
-import subprocess
 
 from maestrowf.abstracts import Dependency
+from maestrowf.utils import start_process
 
 logger = logging.getLogger(__name__)
 
@@ -148,8 +148,7 @@ def acquire(self, substitutions=None):
 
         path = os.path.join(self.path, self.name)
         logger.info("Cloning %s from %s...", self.name, self.url)
-        clone = subprocess.Popen(["git", "clone", self.url, path],
-                                 stdout=subprocess.PIPE)
+        clone = start_process(["git", "clone", self.url, path], shell=False)
         retcode = clone.wait()
         if retcode != 0:
             if retcode == 128:
@@ -166,7 +165,8 @@ def acquire(self, substitutions=None):
 
         if self.hash:
             logger.info("Checking out SHA1 hash '{}'...", self.hash)
-            chkout = subprocess.Popen(["git", "checkout", self.hash], cwd=path)
+            chkout = start_process(["git", "checkout", self.hash],
+                                   cwd=path, shell=False)
             retcode = chkout.wait()
 
             if retcode != 0:
@@ -179,7 +179,8 @@ def acquire(self, substitutions=None):
         if self.tag:
             logger.info("Checking out git tag '{}'...", self.tag)
             tag = "tags/{}".format(self.tag)
-            chkout = subprocess.Popen(["git", "checkout", tag], cwd=path)
+            chkout = start_process(["git", "checkout", tag],
+                                   cwd=path, shell=False)
 
             retcode = chkout.wait()
 
@@ -191,8 +192,8 @@ def acquire(self, substitutions=None):
 
         if self.branch:
             logger.info("Checking out git branch '{}'...", self.branch)
-            chkout = subprocess.Popen(["git", "checkout", self.branch],
-                                      cwd=path)
+            chkout = start_process(["git", "checkout", self.branch],
+                                   cwd=path, shell=False)
 
             retcode = chkout.wait()
 

maestrowf/interfaces/script/localscriptadapter.py

@@ -30,11 +30,11 @@
 """Local interface implementation."""
 import logging
 import os
-from subprocess import PIPE, Popen
 
 from maestrowf.abstracts.enums import JobStatusCode, SubmissionCode, \
     CancelCode
 from maestrowf.abstracts.interfaces import ScriptAdapter
+from maestrowf.utils import start_process
 
 LOGGER = logging.getLogger(__name__)
 
@@ -135,8 +135,7 @@ def submit(self, step, path, cwd, job_map=None, env=None):
         """
         LOGGER.debug("cwd = %s", cwd)
         LOGGER.debug("Script to execute: %s", path)
-        p = Popen(path, shell=False, stdout=PIPE, stderr=PIPE, cwd=cwd,
-                  env=env)
+        p = start_process(path, cwd=cwd, env=env)
         pid = p.pid
         output, err = p.communicate()
         retcode = p.wait()

maestrowf/interfaces/script/slurmscriptadapter.py

@@ -32,11 +32,11 @@
 import logging
 import os
 import re
-from subprocess import PIPE, Popen
 
 from maestrowf.abstracts.interfaces import SchedulerScriptAdapter
 from maestrowf.abstracts.enums import JobStatusCode, State, SubmissionCode, \
     CancelCode
+from maestrowf.utils import start_process
 
 LOGGER = logging.getLogger(__name__)
 
@@ -182,7 +182,7 @@ def submit(self, step, path, cwd, job_map=None, env=None):
 
         LOGGER.debug("cwd = %s", cwd)
         LOGGER.debug("Command to execute: %s", cmd)
-        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE, cwd=cwd, env=env)
+        p = start_process(cmd, cwd=cwd, env=env)
         output, err = p.communicate()
         retcode = p.wait()
 
@@ -213,7 +213,7 @@ def check_jobs(self, joblist):
         # -u = username to search queues for.
         # -t = list of job states to search for. 'all' for all states.
         cmd = "squeue -u $USER -t all"
-        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
+        p = start_process(cmd)
         output, err = p.communicate()
         retcode = p.wait()
 
@@ -275,7 +275,7 @@ def cancel_jobs(self, joblist):
             return CancelCode.OK
 
         cmd = "scancel --quiet {}".format(" ".join(joblist))
-        p = Popen(cmd, shell=True, stdout=PIPE, stderr=PIPE)
+        p = start_process(cmd)
         output, err = p.communicate()
         retcode = p.wait()
 

maestrowf/maestro.py

@@ -34,7 +34,6 @@
 import logging
 import os
 import shutil
-from subprocess import Popen, PIPE
 import six
 import sys
 import tabulate
@@ -44,7 +43,8 @@
 from maestrowf.datastructures import YAMLSpecification
 from maestrowf.datastructures.core import Study
 from maestrowf.datastructures.environment import Variable
-from maestrowf.utils import create_parentdir, csvtable_to_dict, make_safe_path
+from maestrowf.utils import \
+    create_parentdir, csvtable_to_dict, make_safe_path, start_process
 
 
 # Program Globals
@@ -266,7 +266,7 @@ def run_study(args):
                    "&>", "{}.txt".format(os.path.join(
                     study.output_path, exec_dag.name))]
             LOGGER.debug(" ".join(cmd))
-            Popen(" ".join(cmd), shell=True, stdout=PIPE, stderr=PIPE)
+            start_process(cmd)
 
     return 0
 

maestrowf/utils.py

@@ -33,6 +33,7 @@
 import logging
 import os
 import string
+from subprocess import PIPE, Popen
 import time
 
 LOGGER = logging.getLogger(__name__)
@@ -153,3 +154,17 @@ def make_safe_path(*args):
         arg = "".join(c for c in arg if c in valid)
         arg = arg.replace(" ", "_")
     return os.path.join(*args)
+
+
+def start_process(cmd, cwd=None, env=None, shell=True):
+    """
+    Starts a new process using a specified command.
+
+    :param cmd: A string or a list representing the command to be run.
+    :param cwd: Current working path that the process will be started in.
+    :param env: A dictionary containing the environment the process will use.
+    :param shell: Boolean that determines if the process will run a shell.
+    """
+    return Popen(cmd,
+                 shell=shell, stdout=PIPE, stderr=PIPE,
+                 universal_newlines=True)