code: def test_cancel_async_handle_call_during_assignment(serve_instance):
      """Test cancelling handle request during assignment (async context)."""
      signal_actor = SignalActor.remote()
  
      @serve.deployment(max_concurrent_queries=1)
      class Downstream:
          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
  
      @serve.deployment
      class Ingress:
          def __init__(self, handle):
              self._h = handle.options(use_new_handle_api=True)
  
          async def __call__(self, *args):
              # 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 = self._h.remote()
  
              async def one_waiter():
                  return (
                      await signal_actor.cur_num_waiters.remote() == 1
                  )
  
              await async_wait_for_condition(one_waiter)
  
              # Make a second request, cancel it, and verify that it is cancelled.
              second_response = self._h.remote()
              second_response.cancel()
              with pytest.raises(asyncio.CancelledError):
                  await second_response
  
              # Now signal the initial request to finish and check that the second request
              # never reached the replica.
              await signal_actor.send.remote()
              assert await initial_response == 1
              for i in range(2, 12):
                  assert await self._h.remote() == i
  
      h = serve.run(Ingress.bind(Downstream.bind())).options(
          use_new_handle_api=True
      )
      h.remote().result()
  
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 6 for calling a global function with example usage: test_cancel_async_handle_call_during_assignment(serve_instance) and returns: self, unknown