code: def test_handle_apis_detected(
      manage_ray_with_telemetry, use_new_handle_api, call_in_deployment
  ):
      """Check that the various handles are 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_API_USED.get_value_from_report(
              report
          )
          is None
      )
      assert (
          ServeUsageTag.RAY_SERVE_HANDLE_API_USED.get_value_from_report(
              report
          )
          is None
      )
      assert (
          ServeUsageTag.RAY_SERVE_SYNC_HANDLE_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=use_new_handle_api)
  
          async def __call__(self, call_downstream=True):
              if call_downstream:
                  await self._h.remote()
              return "ok"
  
      handle = serve.run(Caller.bind(Downstream.bind()))
  
      if call_in_deployment:
          result = requests.get("http://localhost:8000").text
      elif use_new_handle_api:
          result = handle.remote(call_downstream=False).result()
      else:
          result = ray.get(
              handle.options(use_new_handle_api=False).remote(
                  call_downstream=False
              )
          )
  
      assert result == "ok"
  
      def check_telemetry():
          report = ray.get(storage_handle.get_report.remote())
          print(report["extra_usage_tags"])
          if use_new_handle_api:
              assert (
                  ServeUsageTag.DEPLOYMENT_HANDLE_API_USED.get_value_from_report(
                      report
                  )
                  == "1"
              )
          elif call_in_deployment:
              assert (
                  ServeUsageTag.RAY_SERVE_HANDLE_API_USED.get_value_from_report(
                      report
                  )
                  == "1"
              )
          else:
              assert (
                  ServeUsageTag.RAY_SERVE_SYNC_HANDLE_API_USED.get_value_from_report(
                      report
                  )
                  == "1"
              )
          return True
  
      wait_for_condition(check_telemetry)
  
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 4 for calling a global function with example usage: test_handle_apis_detected(manage_ray_with_telemetry, use_new_handle_api, call_in_deployment) and returns: True, str, str