def test_identity_join_node(tmpdir):
 global _sum_operands
 _sum_operands = []
 os.chdir(str(tmpdir))
 wf = pe.Workflow(name=u'test')
 inputspec = pe.Node(IdentityInterface(fields=[u'n']), name=u'inputspec')
 inputspec.iterables = [(u'n', [1, 2, 3])]
 pre_join1 = pe.Node(IncrementInterface(), name=u'pre_join1')
 wf.connect(inputspec, u'n', pre_join1, u'input1')
 join = pe.JoinNode(IdentityInterface(fields=[u'vector']), joinsource=u'inputspec', joinfield=u'vector', name=u'join')
 wf.connect(pre_join1, u'output1', join, u'vector')
 post_join1 = pe.Node(SumInterface(), name=u'post_join1')
 wf.connect(join, u'vector', post_join1, u'input1')
 result = wf.run()
 assert (len(result.nodes()) == 5), u'The number of expanded nodes is incorrect.'
 assert (_sum_operands[0] == [2, 3, 4]), (u'The join Sum input is incorrect: %s.' % _sum_operands[0])

def test_identity_join_node ( tmpdir ) : global _sum_operands _sum_operands = [ ] os . chdir ( str ( tmpdir ) ) wf = pe . Workflow ( name = u'test' ) inputspec = pe . Node ( IdentityInterface ( fields = [ u'n' ] ) , name = u'inputspec' ) inputspec . iterables = [ ( u'n' , [ 1 , 2 , 3 ] ) ] pre_join1 = pe . Node ( IncrementInterface ( ) , name = u'pre_join1' ) wf . connect ( inputspec , u'n' , pre_join1 , u'input1' ) join = pe . JoinNode ( IdentityInterface ( fields = [ u'vector' ] ) , joinsource = u'inputspec' , joinfield = u'vector' , name = u'join' ) wf . connect ( pre_join1 , u'output1' , join , u'vector' ) post_join1 = pe . Node ( SumInterface ( ) , name = u'post_join1' ) wf . connect ( join , u'vector' , post_join1 , u'input1' ) result = wf . run ( ) assert ( len ( result . nodes ( ) ) == 5 ) , u'The number of expanded nodes is incorrect.' assert ( _sum_operands [ 0 ] == [ 2 , 3 , 4 ] ) , ( u'The join Sum input is incorrect: %s.' % _sum_operands [ 0 ] )
test an identityinterface join .