def test_mapnode_json(tmpdir):
 wd = str(tmpdir)
 os.chdir(wd)
 from nipype import MapNode, Function, Workflow
 def func1(in1):
  return (in1 + 1)
 n1 = MapNode(Function(input_names=[u'in1'], output_names=[u'out'], function=func1), iterfield=[u'in1'], name=u'n1')
 n1.inputs.in1 = [1]
 w1 = Workflow(name=u'test')
 w1.base_dir = wd
 w1.config[u'execution'][u'crashdump_dir'] = wd
 w1.add_nodes([n1])
 w1.run()
 n1.inputs.in1 = [2]
 w1.run()
 n1.inputs.in1 = [1]
 eg = w1.run()
 node = eg.nodes()[0]
 outjson = glob(os.path.join(node.output_dir(), u'_0x*.json'))
 assert (len(outjson) == 1)
 with open(os.path.join(node.output_dir(), u'test.json'), u'wt') as fp:
  fp.write(u'dummy file')
 w1.config[u'execution'].update(**{u'stop_on_first_rerun': True})
 w1.run()

def test_mapnode_json ( tmpdir ) : wd = str ( tmpdir ) os . chdir ( wd ) from nipype import MapNode , Function , Workflow def func1 ( in1 ) : return ( in1 + 1 ) n1 = MapNode ( Function ( input_names = [ u'in1' ] , output_names = [ u'out' ] , function = func1 ) , iterfield = [ u'in1' ] , name = u'n1' ) n1 . inputs . in1 = [ 1 ] w1 = Workflow ( name = u'test' ) w1 . base_dir = wd w1 . config [ u'execution' ] [ u'crashdump_dir' ] = wd w1 . add_nodes ( [ n1 ] ) w1 . run ( ) n1 . inputs . in1 = [ 2 ] w1 . run ( ) n1 . inputs . in1 = [ 1 ] eg = w1 . run ( ) node = eg . nodes ( ) [ 0 ] outjson = glob ( os . path . join ( node . output_dir ( ) , u'_0x*.json' ) ) assert ( len ( outjson ) == 1 ) with open ( os . path . join ( node . output_dir ( ) , u'test.json' ) , u'wt' ) as fp : fp . write ( u'dummy file' ) w1 . config [ u'execution' ] . update ( ** { u'stop_on_first_rerun' : True } ) w1 . run ( )
tests that mapnodes dont generate excess jsons .