code: def test_deployment_handle_to_obj_ref_detected(
      manage_ray_with_telemetry, mode
  ):
      """Check that the handle to_object_ref API is detected correctly by telemetry."""
  
      subprocess.check_output(["ray", "start", "--head"])
      wait_for_condition(check_ray_started, timeout=5)
  
      storage_handle = start_telemetry_app()
      wait_for_condition(
          lambda: ray.get(storage_handle.get_reports_received.remote())
          > 0,
          timeout=5,
      )
  
      report = ray.get(storage_handle.get_report.remote())
      print(report["extra_usage_tags"])
      assert (
          ServeUsageTag.DEPLOYMENT_HANDLE_TO_OBJECT_REF_API_USED.get_value_from_report(
              report
          )
          is None
      )
  
      @serve.deployment
      class Downstream:
          def __call__(self):
              return "hi"
  
      @serve.deployment
      class Caller:
          def __init__(self, h):
              self._h = h.options(use_new_handle_api=True)
  
          async def get(self, call_downstream=False):
              if call_downstream:
                  await self._h.remote()._to_object_ref()
              return "ok"
  
          async def __call__(self):
              return await self.get()
  
      handle = serve.run(Caller.bind(Downstream.bind()))
  
      if mode == "http":
          result = requests.get("http://localhost:8000").text
      elif mode == "outside_deployment":
          result = ray.get(
              handle.options(use_new_handle_api=True)
              .get.remote()
              ._to_object_ref_sync()
          )
      else:
          result = (
              handle.options(
                  use_new_handle_api=True,
              )
              .get.remote(call_downstream=True)
              .result()
          )
  
      assert result == "ok"
  
      def check_telemetry(tag_should_be_set: bool):
          report = ray.get(storage_handle.get_report.remote())
          print(report["extra_usage_tags"])
          tag = ServeUsageTag.DEPLOYMENT_HANDLE_TO_OBJECT_REF_API_USED
          if tag_should_be_set:
              assert tag.get_value_from_report(report) == "1"
          else:
              assert tag.get_value_from_report(report) is None
  
          return True
  
      if mode == "http":
          for _ in range(20):
              check_telemetry(tag_should_be_set=False)
              time.sleep(0.1)
      else:
          wait_for_condition(check_telemetry, tag_should_be_set=True)
  
imports: import subprocess
  import sys
  import time
  import pytest
  import requests
  import yaml
  from fastapi import FastAPI
  import ray
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray._private.usage.usage_lib import get_extra_usage_tags_to_report
  from ray.serve._private.common import ApplicationStatus
  from ray.serve._private.constants import SERVE_MULTIPLEXED_MODEL_ID
  from ray.serve._private.test_utils import TELEMETRY_ROUTE_PREFIX, TelemetryStorage, check_ray_started, receiver_app, start_telemetry_app
  from ray.serve._private.usage import ServeUsageTag
  from ray.serve.context import _get_global_client
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 5 for calling a global function with example usage: test_deployment_handle_to_obj_ref_detected(manage_ray_with_telemetry, mode) and returns: True, str, str, unknown