code: def test_exception_in_generator(
      serve_instance, use_async: bool, use_fastapi: bool
  ):
      async def hi_gen_async():
          yield "first result"
          raise Exception("raised in generator")
  
      def hi_gen_sync():
          yield "first result"
          raise Exception("raised in generator")
  
      if use_fastapi:
          app = FastAPI()
  
          @serve.deployment
          @serve.ingress(app)
          class SimpleGenerator:
              @app.get("/")
              def stream_hi(
                  self, request: Request
              ) -> StreamingResponse:
                  gen = hi_gen_async() if use_async else hi_gen_sync()
                  return StreamingResponse(gen, media_type="text/plain")
  
      else:
  
          @serve.deployment
          class SimpleGenerator:
              def __call__(self, request: Request) -> StreamingResponse:
                  gen = hi_gen_async() if use_async else hi_gen_sync()
                  return StreamingResponse(gen, media_type="text/plain")
  
      serve.run(SimpleGenerator.bind())
  
      r = requests.get("http://localhost:8000", stream=True)
      r.raise_for_status()
      stream_iter = r.iter_content(chunk_size=None, decode_unicode=True)
      assert next(stream_iter) == "first result"
      with pytest.raises(requests.exceptions.ChunkedEncodingError):
          next(stream_iter)
  
file_path: ray/python/ray/serve/tests/test_streaming_response.py
imports: import asyncio
  import os
  from typing import AsyncGenerator
  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
  from ray.serve.handle import RayServeHandle
label_desc: Use image node_id 4 for calling a global function with example usage: test_exception_in_generator(serve_instance, use_async, use_fastapi) and returns: StreamingResponse, StreamingResponse