class_docstr: DAGRepr test
class_docstr_tok: ["DAGRepr","test"]
docstr: _input[(a,)]->op1[(a,)-(b, c)]->op2[(c,)-(c,)]->op3[(b,)-(e,)]->_output[(f, e)]
                                                |-->op4[(a,)-(f,)]--^
docstr_tok: ["_input","[","(","a",",",")","]","-",">","op1","[","(","a",",",")","-","(","b",",","c",")","]","-",">","op2","[","(","c",",",")","-","(","c",",",")","]","-",">","op3","[","(","b",",",")","-","(","e",",",")","]","-",">","_output","[","(","f",",","e",")","]","|","--",">","op4","[","(","a",",",")","-","(","f",",",")","]","--","^"]
code: def test_concat1(self):
      """
      _input[(a,)]->op1[(a,)-(b, c)]->op2[(c,)-(c,)]->op3[(b,)-(e,)]->_output[(f, e)]
                                                    |-->op4[(a,)-(f,)]--^
      """
      towhee_dag_test = {
          "_input": {
              "inputs": ("a",),
              "outputs": ("a",),
              "iter_info": {"type": "map", "param": None},
              "op_info": {
                  "operator": "NOPNodeOperator",
                  "type": "built_in",
                  "init_args": None,
                  "init_kws": None,
                  "tag": None,
              },
              "next_nodes": ["op1"],
          },
          "op1": {
              "inputs": ("a",),
              "outputs": ("b", "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": ["op2"],
          },
          "op2": {
              "inputs": ("c",),
              "outputs": ("c",),
              "iter_info": {"type": "flat_map", "param": None},
              "op_info": {
                  "operator": "test2",
                  "type": "hub",
                  "init_args": None,
                  "init_kws": None,
                  "tag": "main",
              },
              "config": None,
              "next_nodes": ["op3", "op4"],
          },
          "op3": {
              "inputs": ("b",),
              "outputs": ("e",),
              "iter_info": {"type": "map", "param": None},
              "op_info": {
                  "operator": "test3",
                  "type": "hub",
                  "init_args": None,
                  "init_kws": None,
                  "tag": "main",
              },
              "config": None,
              "next_nodes": ["op5"],
          },
          "op4": {
              "inputs": ("a",),
              "outputs": ("f",),
              "iter_info": {"type": "flat_map", "param": None},
              "op_info": {
                  "operator": "test4",
                  "type": "hub",
                  "init_args": None,
                  "init_kws": None,
                  "tag": "main",
              },
              "config": None,
              "next_nodes": ["op5"],
          },
          "op5": {
              "inputs": (),
              "outputs": (),
              "iter_info": {"type": "concat", "param": None},
              "op_info": {
                  "operator": "test5",
                  "type": "hub",
                  "init_args": None,
                  "init_kws": None,
                  "tag": "main",
              },
              "config": None,
              "next_nodes": ["_output"],
          },
          "_output": {
              "inputs": ("e", "f"),
              "outputs": ("e", "f"),
              "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), 9)
      self.assertEqual(len(nodes), 7)
  
      edge0 = [("a", ColumnType.SCALAR)]
      edge1 = [("a", ColumnType.SCALAR)]
      edge2 = [
          ("a", ColumnType.SCALAR),
          ("b", ColumnType.SCALAR),
          ("c", ColumnType.SCALAR),
      ]
      edge3 = [("b", ColumnType.SCALAR)]
      edge4 = [("a", ColumnType.SCALAR)]
      edge5 = [("e", ColumnType.SCALAR)]
      edge6 = [("f", ColumnType.QUEUE)]
      edge7 = [("e", ColumnType.SCALAR), ("f", ColumnType.QUEUE)]
      edge8 = [("e", ColumnType.SCALAR), ("f", 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 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 edge6),
      )
      self.assertEqual(
          dict((s, t) for s, t in edges[7]["data"]),
          dict((s, t) for s, t in edge7),
      )
      self.assertEqual(
          dict((s, t) for s, t in edges[8]["data"]),
          dict((s, t) for s, t in edge8),
      )
  
      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, 4])
      self.assertEqual(nodes["op3"].in_edges, [3])
      self.assertEqual(nodes["op3"].out_edges, [5])
      self.assertEqual(nodes["op4"].in_edges, [4])
      self.assertEqual(nodes["op4"].out_edges, [6])
      self.assertEqual(nodes["op5"].in_edges, [5, 6])
      self.assertEqual(nodes["op5"].out_edges, [7])
      self.assertEqual(nodes["_output"].in_edges, [7])
      self.assertEqual(nodes["_output"].out_edges, [8])
  
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 8 for calling the TestDAGRepr obj's underlying member method code with example usage: obj.test_concat1() without return types