code: def test_backpressure_invalid(shutdown_only):
      """
      Verify invalid cases.
      1. Verify using backpressure + async actor raises an exception
      2. Verify _generator_backpressure_num_objects == 0 is not allowed.
      """
      ray.init(num_cpus=1)
  
      @ray.remote
      class A:
          async def f(self):
              for i in range(10):
                  print("yield", i)
                  yield i
  
      a = A.remote()
      gen = a.f.options(_generator_backpressure_num_objects=1).remote()
      with pytest.raises(ValueError):
          ray.get(next(gen))
  
      with pytest.raises(
          ValueError, match="backpressure_num_objects=0 is not allowed"
      ):
  
          @ray.remote(_generator_backpressure_num_objects=0)
          def f():
              pass
  
file_path: ray/python/ray/tests/test_streaming_generator_backpressure.py
imports: import asyncio
  import pytest
  import numpy
  import sys
  import time
  import signal
  import os
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.util.state import list_tasks
label_desc: Use image node_id 4 for calling a global function with example usage: test_backpressure_invalid(shutdown_only) without return types