Skip to content
Next Next commit
TEST: Skip PBS test in absence of qsub, use tmp_path fixture
  • Loading branch information
effigies committed Mar 29, 2020
commit 933f64f769d66670b750a16b77de6c45f9b05b80
18 changes: 4 additions & 14 deletions nipype/pipeline/plugins/tests/test_pbs.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,5 @@
# -*- coding: utf-8 -*-
import os
from shutil import rmtree
from tempfile import mkdtemp
from time import sleep
from shutil import which

import nipype.interfaces.base as nib
import pytest
Expand Down Expand Up @@ -32,22 +29,15 @@ def _list_outputs(self):
return outputs


@pytest.mark.xfail(reason="not known")
def test_run_pbsgraph():
cur_dir = os.getcwd()
temp_dir = mkdtemp(prefix="test_engine_")
os.chdir(temp_dir)

pipe = pe.Workflow(name="pipe")
@pytest.mark.skipif(which('qsub') is None, reason="PBS not installed")
def test_run_pbsgraph(tmp_path):
pipe = pe.Workflow(name="pipe", base_dir=str(tmp_path))
mod1 = pe.Node(interface=PbsTestInterface(), name="mod1")
mod2 = pe.MapNode(interface=PbsTestInterface(), iterfield=["input1"], name="mod2")
pipe.connect([(mod1, mod2, [("output1", "input1")])])
pipe.base_dir = os.getcwd()
mod1.inputs.input1 = 1
execgraph = pipe.run(plugin="PBSGraph")
names = [".".join((node._hierarchy, node.name)) for node in execgraph.nodes()]
node = list(execgraph.nodes())[names.index("pipe.mod1")]
result = node.get_output("output1")
assert result == [1, 1]
os.chdir(cur_dir)
rmtree(temp_dir)