code: def test_cancel_generator_async(serve_instance):
      """Test cancelling streaming handle request during execution."""
      signal_actor = SignalActor.remote()
  
      @serve.deployment
      class Downstream:
          async def __call__(self, *args):
              yield "hi"
              await send_signal_on_cancellation(signal_actor)
  
      @serve.deployment
      class Ingress:
          def __init__(self, handle):
              self._h = handle.options(
                  use_new_handle_api=True, stream=True
              )
  
          async def __call__(self, *args):
              # Send a request and wait for it to start executing.
              g = self._h.remote()
              assert await g.__anext__() == "hi"
  
              # Cancel it and verify that it is cancelled via signal.
              g.cancel()
  
              with pytest.raises(ray.exceptions.TaskCancelledError):
                  assert await g.__anext__() == "hi"
  
              await signal_actor.wait.remote()
  
      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 8 for calling a global function with example usage: test_cancel_generator_async(serve_instance) without return types