code: def test_dag_with_actor_handle(shared_ray_instance):
      """Verify DAG API works with actor created by .remote"""
  
      @ray.remote
      class Worker:
          def __init__(self):
              self.forward_called = 0
              self.init_called = 0
  
          def forward(self, input):
              print("forward")
              self.forward_called += 1
              return input
  
          def initialize(self, input):
              print("initialize")
              self.init_called += 1
              return input
  
          def get(self):
              return (self.forward_called, self.init_called)
  
      worker = Worker.remote()
      with InputNode() as input_node:
          init_dag = worker.initialize.bind(input_node)
      with InputNode() as input_node:
          forward_dag = worker.forward.bind(input_node)
  
      assert ray.get(init_dag.execute(1)) == 1
      assert ray.get(forward_dag.execute(2)) == 2
  
      # Make sure both forward/initialize called only once
      assert ray.get(worker.get.remote()) == (1, 1)
  
      # Double check the actor is resued.
      assert ray.get(init_dag.execute(1)) == 1
      assert ray.get(worker.get.remote()) == (1, 2)
  
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 2 for calling a global function with example usage: test_dag_with_actor_handle(shared_ray_instance) and returns: input, input, self,  self