code: def test_streaming_request_already_sent_and_timed_out(
      ray_instance, shutdown_serve
  ):
      """
      Verify that streaming requests are timed out even if some chunks have already
      been sent.
      """
  
      @serve.deployment(
          graceful_shutdown_timeout_s=0, max_concurrent_queries=1
      )
      class SleepForNSeconds:
          def __init__(self, sleep_s: int):
              self.sleep_s = sleep_s
  
          def generate_numbers(self) -> Generator[str, None, None]:
              for i in range(2):
                  yield f"generated {i}"
                  time.sleep(self.sleep_s)
  
          def __call__(self, request: Request) -> StreamingResponse:
              gen = self.generate_numbers()
              return StreamingResponse(
                  gen, status_code=200, media_type="text/plain"
              )
  
      serve.run(SleepForNSeconds.bind(0.11))  # 0.11s > 0.1s timeout
  
      r = requests.get("http://localhost:8000", stream=True)
      iterator = r.iter_content(chunk_size=None, decode_unicode=True)
  
      # The first chunk should be received successfully.
      assert iterator.__next__() == "generated 0"
      assert r.status_code == 200
  
      # The second chunk should time out and raise error.
      with pytest.raises(
          requests.exceptions.ChunkedEncodingError
      ) as request_error:
          iterator.__next__()
      assert "Connection broken" in str(request_error.value)
  
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 7 for calling a global function with example usage: test_streaming_request_already_sent_and_timed_out(ray_instance, shutdown_serve) and returns: StreamingResponse