code: def test_cancel_on_http_client_disconnect_during_assignment(
      serve_instance,
  ):
      """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 block trying to assign a replica.
      initial_response = h.remote()
      wait_for_condition(
          lambda: ray.get(signal_actor.cur_num_waiters.remote()) == 1
      )
  
      # Intentionally time out on the client, causing it to disconnect.
      with pytest.raises(requests.exceptions.ReadTimeout):
          requests.get("http://localhost:8000", timeout=0.5)
  
      # 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 concurrent.futures
  import sys
  import pytest
  import requests
  from fastapi import FastAPI
  from starlette.requests import Request
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, async_wait_for_condition, wait_for_condition
  from ray.serve._private.test_utils import send_signal_on_cancellation
label_desc: Use image node_id 2 for calling a global function with example usage: test_cancel_on_http_client_disconnect_during_assignment(serve_instance) and returns: self