    def test_concat(self):
        """
        _input[(a,b)]->op1[(a,)-(c,)]->op3[(c, d)-(c, d)]->_output[(c, d)]
                |------>op2[(b,)-(d,)]----^
        """
        towhee_dag_test = {
            '_input': {
                'inputs': ('a', 'b'),
                'outputs': ('a', 'b'),
                'iter_info': {'type': 'map', 'param': None},
                'op_info': {'operator': 'NOPNodeOperator', 'type': 'built_in', 'init_args': None, 'init_kws': None, 'tag': None},
                'next_nodes': ['op1', 'op2']
            },
            'op1': {
                'inputs': ('a',),
                'outputs': ('c',),
                'iter_info': {'type': 'map', 'param': None},
                'op_info': {'operator': 'test1', 'type': 'hub', 'init_args': None, 'init_kws': None, 'tag': 'main'},
                'config': None,
                'next_nodes': ['op3']
            },
            'op2': {
                'inputs': ('b',),
                'outputs': ('d',),
                'iter_info': {'type': 'map', 'param': None},
                'op_info': {'operator': 'test2', 'type': 'hub', 'init_args': None, 'init_kws': None, 'tag': 'main'},
                'config': None,
                'next_nodes': ['op3']
            },
            'op3': {
                'inputs': (),
                'outputs': (),
                'iter_info': {'type': 'concat', 'param': None},
                'op_info': {'operator': 'test3', 'type': 'hub', 'init_args': None, 'init_kws': None, 'tag': 'main'},
                'config': None,
                'next_nodes': ['_output']
            },
            '_output': {
                'inputs': ('c', 'd'),
                'outputs': ('c', 'd'),
                'iter_info': {'type': 'map', 'param': None},
                'op_info': {'operator': 'NOPNodeOperator', 'type': 'built_in', 'init_args': None, 'init_kws': None, 'tag': None},
                'next_nodes': None
            },
        }
        dr = DAGRepr.from_dict(towhee_dag_test)
        edges = dr.edges
        nodes = dr.nodes
        self.assertEqual(len(edges), 7)
        self.assertEqual(len(nodes), 5)

        edge0 = [('a', ColumnType.SCALAR), ('b', ColumnType.SCALAR)]
        edge1 = [('a', ColumnType.SCALAR)]
        edge2 = [('b', ColumnType.SCALAR)]
        edge3 = [('c', ColumnType.SCALAR)]
        edge4 = [('d', ColumnType.SCALAR)]
        edge5 = [('c', ColumnType.SCALAR), ('d', ColumnType.SCALAR)]
        self.assertEqual(dict((s, t) for s, t in edges[0]['data']), dict((s, t) for s, t in edge0))
        self.assertEqual(dict((s, t) for s, t in edges[1]['data']), dict((s, t) for s, t in edge1))
        self.assertEqual(dict((s, t) for s, t in edges[2]['data']), dict((s, t) for s, t in edge2))
        self.assertEqual(dict((s, t) for s, t in edges[3]['data']), dict((s, t) for s, t in edge3))
        self.assertEqual(dict((s, t) for s, t in edges[4]['data']), dict((s, t) for s, t in edge4))
        self.assertEqual(dict((s, t) for s, t in edges[5]['data']), dict((s, t) for s, t in edge5))
        self.assertEqual(dict((s, t) for s, t in edges[6]['data']), dict((s, t) for s, t in edge5))

        self.assertEqual(nodes['_input'].in_edges, [0])
        self.assertEqual(nodes['_input'].out_edges, [1, 2])
        self.assertEqual(nodes['op1'].in_edges, [1])
        self.assertEqual(nodes['op1'].out_edges, [3])
        self.assertEqual(nodes['op2'].in_edges, [2])
        self.assertEqual(nodes['op2'].out_edges, [4])
        self.assertEqual(nodes['op3'].in_edges, [3, 4])
        self.assertEqual(nodes['op3'].out_edges, [5])
        self.assertEqual(nodes['_output'].in_edges, [5])
        self.assertEqual(nodes['_output'].out_edges, [6])
_input[(a,b)]->op1[(a,)-(c,)]->op3[(c, d)-(c, d)]->_output[(c, d)]
|------>op2[(b,)-(d,)]----^
tests/unittests/runtime/test_dag_repr.py