class_docstr: DAGRepr test
class_docstr_tok: ["DAGRepr","test"]
docstr: _input(map)[(a, b, c)]->op1(flat_map)[(a, c)-(a, c)]->op2(filter)[(a, b)-(d, e)]->_output(map)[(d, c)]
docstr_tok: ["_input","(","map",")","[","(","a",",","b",",","c",")","]","-",">","op1","(","flat_map",")","[","(","a",",","c",")","-","(","a",",","c",")","]","-",">","op2","(","filter",")","[","(","a",",","b",")","-","(","d",",","e",")","]","-",">","_output","(","map",")","[","(","d",",","c",")","]"]
code: 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])
  
file_path: towhee/tests/unittests/runtime/test_dag_repr.py
imports: import copy
  import unittest
  from towhee import pipe
  from towhee.runtime.dag_repr import DAGRepr, NodeRepr
  from towhee.runtime.data_queue import ColumnType
  from towhee.runtime.schema_repr import SchemaRepr
label_desc: Use image node_id 6 for calling the TestDAGRepr obj's underlying member method code with example usage: obj.test_edges() without return types