ENH: test doesn't work yet but oar.py works

Author: demianw

nipype/pipeline/plugins/oar.py

@@ -2,6 +2,7 @@
 """
 
 import os
+import stat
 from time import sleep
 import subprocess
 import json
@@ -26,6 +27,7 @@ class OARPlugin(SGELikeBatchManagerBase):
 
     # Addtional class variables
     _max_jobname_len = 15
+    _oarsub_args = ''
 
     def __init__(self, **kwargs):
         template = """
@@ -53,9 +55,10 @@ def _is_pending(self, taskid):
             stderr=subprocess.PIPE
         )
         o, e = proc.communicate()
+        parsed_result = json.loads(o)[taskid].lower()
+        is_pending = 'error' not in parsed_result
 
-        parsed_result = json.loads(o)[taskid]
-        return 'error' not in parsed_result
+        return is_pending
 
     def _submit_batchtask(self, scriptfile, node):
         cmd = CommandLine('oarsub', environ=os.environ.data,
@@ -72,10 +75,7 @@ def _submit_batchtask(self, scriptfile, node):
                 oarsubargs = node.plugin_args['oarsub_args']
             else:
                 oarsubargs += (" " + node.plugin_args['oarsub_args'])
-        if '-o' not in oarsubargs:
-            oarsubargs = '%s -o %s' % (oarsubargs, path)
-        if '-E' not in oarsubargs:
-            oarsubargs = '%s -E %s' % (oarsubargs, path)
+
         if node._hierarchy:
             jobname = '.'.join((os.environ.data['LOGNAME'],
                                 node._hierarchy,
@@ -87,12 +87,28 @@ def _submit_batchtask(self, scriptfile, node):
         jobnameitems.reverse()
         jobname = '.'.join(jobnameitems)
         jobname = jobname[0:self._max_jobname_len]
+
+        if '-O' not in oarsubargs:
+            oarsubargs = '%s -O %s' % (
+                oarsubargs,
+                os.path.join(path, jobname + '.stdout')
+            )
+        if '-E' not in oarsubargs:
+            oarsubargs = '%s -E %s' % (
+                oarsubargs,
+                os.path.join(path, jobname + '.stderr')
+            )
+        if '-J' not in oarsubargs:
+            oarsubargs = '%s -J' % (oarsubargs)
+
+        os.chmod(scriptfile, stat.S_IEXEC | stat.S_IREAD | stat.S_IWRITE)
         cmd.inputs.args = '%s -n %s -S %s' % (
            oarsubargs,
            jobname,
            scriptfile
         )
 
+        print cmd.inputs.args
         oldlevel = iflogger.level
         iflogger.setLevel(logging.getLevelName('CRITICAL'))
         tries = 0
@@ -106,7 +122,7 @@ def _submit_batchtask(self, scriptfile, node):
                     # sleep 2 seconds and try again.
                 else:
                     iflogger.setLevel(oldlevel)
-                    raise RuntimeError('\n'.join((('Could not submit pbs task'
+                    raise RuntimeError('\n'.join((('Could not submit OAR task'
                                                    ' for node %s') % node._id,
                                                   str(e))))
             else:
@@ -126,5 +142,4 @@ def _submit_batchtask(self, scriptfile, node):
         taskid = json.loads(o)['job_id']
         self._pending[taskid] = node.output_dir()
         logger.debug('submitted OAR task: %s for node %s' % (taskid, node._id))
-
         return taskid

nipype/pipeline/plugins/tests/test_oar.py

@@ -1,4 +1,9 @@
+from nipype import config, logging
+config.enable_debug_mode()
+logging.update_logging(config)
+
 import os
+from os import path
 from shutil import rmtree
 from tempfile import mkdtemp
 
@@ -7,6 +12,7 @@
 import nipype.pipeline.engine as pe
 
 
+
 class InputSpec(nib.TraitedSpec):
     input1 = nib.traits.Int(desc='a random int')
     input2 = nib.traits.Int(desc='a random int')
@@ -31,9 +37,9 @@ def _list_outputs(self):
 
 
 @skipif(False)
-def test_run_oargraph():
+def test_run_oar():
     cur_dir = os.getcwd()
-    temp_dir = mkdtemp(prefix='test_engine_')
+    temp_dir = mkdtemp(prefix='test_engine_', dir=os.getcwd())
     os.chdir(temp_dir)
 
     pipe = pe.Workflow(name='pipe')
@@ -45,6 +51,7 @@ def test_run_oargraph():
     pipe.base_dir = os.getcwd()
     mod1.inputs.input1 = 1
     execgraph = pipe.run(plugin="OAR")
+    print "Run"
     names = [
         '.'.join((node._hierarchy, node.name))
         for node in execgraph.nodes()