def test_nested_workflow_join(tmpdir):
 wd = str(tmpdir)
 os.chdir(wd)
 def nested_wf(i, name=u'smallwf'):
  inputspec = pe.Node(IdentityInterface(fields=[u'n']), name=u'inputspec')
  inputspec.iterables = [(u'n', i)]
  pre_join = pe.Node(IncrementInterface(), name=u'pre_join')
  join = pe.JoinNode(IdentityInterface(fields=[u'n']), joinsource=u'inputspec', joinfield=u'n', name=u'join')
  wf = pe.Workflow(name=(u'wf_%d' % i[0]))
  wf.connect(inputspec, u'n', pre_join, u'input1')
  wf.connect(pre_join, u'output1', join, u'n')
  return wf
 meta_wf = pe.Workflow(name=u'meta', base_dir=u'.')
 for i in [[1, 3], [2, 4]]:
  mini_wf = nested_wf(i)
  meta_wf.add_nodes([mini_wf])
 result = meta_wf.run()
 assert (len(result.nodes()) == 6), u'The number of expanded nodes is incorrect.'

def test_nested_workflow_join ( tmpdir ) : wd = str ( tmpdir ) os . chdir ( wd ) def nested_wf ( i , name = u'smallwf' ) : inputspec = pe . Node ( IdentityInterface ( fields = [ u'n' ] ) , name = u'inputspec' ) inputspec . iterables = [ ( u'n' , i ) ] pre_join = pe . Node ( IncrementInterface ( ) , name = u'pre_join' ) join = pe . JoinNode ( IdentityInterface ( fields = [ u'n' ] ) , joinsource = u'inputspec' , joinfield = u'n' , name = u'join' ) wf = pe . Workflow ( name = ( u'wf_%d' % i [ 0 ] ) ) wf . connect ( inputspec , u'n' , pre_join , u'input1' ) wf . connect ( pre_join , u'output1' , join , u'n' ) return wf meta_wf = pe . Workflow ( name = u'meta' , base_dir = u'.' ) for i in [ [ 1 , 3 ] , [ 2 , 4 ] ] : mini_wf = nested_wf ( i ) meta_wf . add_nodes ( [ mini_wf ] ) result = meta_wf . run ( ) assert ( len ( result . nodes ( ) ) == 6 ) , u'The number of expanded nodes is incorrect.'
test collecting join inputs within a nested workflow .