code: def test_output_node(shared_ray_instance):
      @ray.remote
      def f(input):
          return input
  
      with pytest.raises(ValueError):
          with InputNode() as input_data:
              dag = MultiOutputNode(f.bind(input_data))
  
      with InputNode() as input_data:
          dag = MultiOutputNode([f.bind(input_data)])
  
      assert ray.get(dag.execute(1)) == [1]
      assert ray.get(dag.execute(2)) == [2]
  
      with InputNode() as input_data:
          dag = MultiOutputNode(
              [f.bind(input_data["x"]), f.bind(input_data["y"])]
          )
  
      refs = dag.execute({"x": 1, "y": 2})
      assert len(refs) == 2
      assert ray.get(refs) == [1, 2]
  
      with InputNode() as input_data:
          dag = MultiOutputNode(
              [
                  f.bind(input_data["x"]),
                  f.bind(input_data["y"]),
                  f.bind(input_data["x"]),
              ]
          )
  
      refs = dag.execute({"x": 1, "y": 2})
      assert len(refs) == 3
      assert ray.get(refs) == [1, 2, 1]
  
file_path: ray/python/ray/dag/tests/test_output_node.py
imports: import pytest
  import ray
  from ray.dag.input_node import InputNode
  from ray.dag.output_node import MultiOutputNode
  from ray.util.state import list_tasks
  from ray._private.test_utils import wait_for_condition
label_desc: Use image node_id 1 for calling a global function with example usage: test_output_node(shared_ray_instance) and returns: input