    def test_edges(self):
        """
        _input(map)[(a, b, c)]->op1(flat_map)[(a, c)-(a, c)]->op2(filter)[(a, b)-(d, e)]->_output(map)[(d, c)]
        """
        towhee_dag_test = copy.deepcopy(self.dag_dict)
        dr = DAGRepr.from_dict(towhee_dag_test)
        edges = dr.edges
        nodes = dr.nodes
        self.assertEqual(len(edges), 5)
        self.assertEqual(len(nodes), 4)

        edge0 = [('a', ColumnType.SCALAR), ('b', ColumnType.SCALAR), ('c', ColumnType.SCALAR)]
        edge1 = [('a', ColumnType.QUEUE), ('c', ColumnType.QUEUE), ('b', ColumnType.SCALAR)]
        edge2 = [('d', ColumnType.QUEUE), ('c', ColumnType.QUEUE)]
        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 edge0))
        self.assertEqual(dict((s, t) for s, t in edges[2]['data']), dict((s, t) for s, t in edge1))
        self.assertEqual(dict((s, t) for s, t in edges[3]['data']), dict((s, t) for s, t in edge2))
        self.assertEqual(dict((s, t) for s, t in edges[4]['data']), dict((s, t) for s, t in edge2))

        self.assertEqual(nodes['_input'].in_edges, [0])
        self.assertEqual(nodes['_input'].out_edges, [1])
        self.assertEqual(nodes['op1'].in_edges, [1])
        self.assertEqual(nodes['op1'].out_edges, [2])
        self.assertEqual(nodes['op2'].in_edges, [2])
        self.assertEqual(nodes['op2'].out_edges, [3])
        self.assertEqual(nodes['_output'].in_edges, [3])
        self.assertEqual(nodes['_output'].out_edges, [4])
_input(map)[(a, b, c)]->op1(flat_map)[(a, c)-(a, c)]->op2(filter)[(a, b)-(d, e)]->_output(map)[(d, c)]
tests/unittests/runtime/test_dag_repr.py