code: def test_dag_to_workflow_execution(workflow_start_regular_shared):
      """This test constructs a DAG with complex dependencies
      and turns it into a workflow."""
  
      @ray.remote
      def begin(x, pos, a):
          return x * a + pos  # 23.14
  
      @ray.remote
      def left(x, c, a):
          return f"left({x}, {c}, {a})"
  
      @ray.remote
      def right(x, b, pos):
          return f"right({x}, {b}, {pos})"
  
      @ray.remote
      def end(lf, rt, b):
          return f"{lf},{rt};{b}"
  
      with pytest.raises(TypeError):
          workflow.run_async(begin.remote(1, 2, 3))
  
      with InputNode() as dag_input:
          f = begin.bind(2, dag_input[1], a=dag_input.a)
          lf = left.bind(f, "hello", dag_input.a)
          rt = right.bind(f, b=dag_input.b, pos=dag_input[0])
          b = end.bind(lf, rt, b=dag_input.b)
  
      assert (
          workflow.run(b, 2, 3.14, a=10, b="ok")
          == "left(23.14, hello, 10),right(23.14, ok, 2);ok"
      )
  
file_path: ray/python/ray/workflow/tests/test_dag_to_workflow.py
imports: from ray.tests.conftest import *
  import pytest
  import ray
  from ray import workflow
  from ray.dag import InputNode
label_desc: Use image node_id 1 for calling a global function with example usage: test_dag_to_workflow_execution(workflow_start_regular_shared) and returns: unknown, str, x, str, c, str, a, str, str, x, str, b, str, pos, str, lf, str, rt, str, b