code: def test_cancel_on_http_timeout_during_assignment(
      ray_instance, shutdown_serve
  ):
      """Test the client disconnecting while the proxy is assigning the request."""
      signal_actor = SignalActor.remote()
  
      @serve.deployment(max_concurrent_queries=1)
      class Ingress:
          def __init__(self):
              self._num_requests = 0
  
          async def __call__(self, *args):
              self._num_requests += 1
              await signal_actor.wait.remote()
  
              return self._num_requests
  
      h = serve.run(Ingress.bind()).options(use_new_handle_api=True)
  
      # Send a request and wait for it to be ongoing so we know that further requests
      # will be blocking trying to assign a replica.
      initial_response = h.remote()
      wait_for_condition(
          lambda: ray.get(signal_actor.cur_num_waiters.remote()) == 1
      )
  
      # Request should time out, causing the handler and handle call to be cancelled.
      assert requests.get("http://localhost:8000").status_code == 408
  
      # Now signal the initial request to finish and check that the request sent via HTTP
      # never reaches the replica.
      ray.get(signal_actor.send.remote())
      assert initial_response.result() == 1
      for i in range(2, 12):
          assert h.remote().result() == i
  
imports: import asyncio
  import os
  import sys
  import time
  from typing import Generator, Set
  import pytest
  import requests
  from fastapi import FastAPI
  from starlette.requests import Request
  from starlette.responses import StreamingResponse
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.dashboard.modules.serve.sdk import ServeSubmissionClient
  from ray.serve._private.common import ApplicationStatus
  from ray.serve._private.test_utils import send_signal_on_cancellation
  from ray.serve.schema import ServeInstanceDetails
  from ray.util.state import list_tasks
label_desc: Use image node_id 10 for calling a global function with example usage: test_cancel_on_http_timeout_during_assignment(ray_instance, shutdown_serve) and returns: self