text: import asyncio
  import time
  import os
  
  import requests
  import pytest
  import starlette.responses
  
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  
  
  def test_e2e(serve_instance):
      @serve.deployment(name="api")
      def function(starlette_request):
          return {"method": starlette_request.method}
  
      function.deploy()
  
      resp = requests.get("http://127.0.0.1:8000/api").json()["method"]
      assert resp == "GET"
  
      resp = requests.post("http://127.0.0.1:8000/api").json()["method"]
      assert resp == "POST"
  
  
  def test_starlette_response(serve_instance):
      @serve.deployment(name="basic")
      def basic(_):
          return starlette.responses.Response("Hello, world!", media_type="text/plain")
  
      basic.deploy()
      assert requests.get("http://127.0.0.1:8000/basic").text == "Hello, world!"
  
      @serve.deployment(name="html")
      def html(_):
          return starlette.responses.HTMLResponse(
              "<html><body><h1>Hello, world!</h1></body></html>"
          )
  
      html.deploy()
      assert (
          requests.get("http://127.0.0.1:8000/html").text
          == "<html><body><h1>Hello, world!</h1></body></html>"
      )
  
      @serve.deployment(name="plain_text")
      def plain_text(_):
          return starlette.responses.PlainTextResponse("Hello, world!")
  
      plain_text.deploy()
      assert requests.get("http://127.0.0.1:8000/plain_text").text == "Hello, world!"
  
      @serve.deployment(name="json")
      def json(_):
          return starlette.responses.JSONResponse({"hello": "world"})
  
      json.deploy()
      assert requests.get("http://127.0.0.1:8000/json").json()["hello"] == "world"
  
      @serve.deployment(name="redirect")
      def redirect(_):
          return starlette.responses.RedirectResponse(url="http://127.0.0.1:8000/basic")
  
      redirect.deploy()
      assert requests.get("http://127.0.0.1:8000/redirect").text == "Hello, world!"
  
      @serve.deployment(name="streaming")
      def streaming(_):
          async def slow_numbers():
              for number in range(1, 4):
                  yield str(number)
                  await asyncio.sleep(0.01)
  
          return starlette.responses.StreamingResponse(
              slow_numbers(), media_type="text/plain", status_code=418
          )
  
      streaming.deploy()
      resp = requests.get("http://127.0.0.1:8000/streaming")
      assert resp.text == "123"
      assert resp.status_code == 418
  
  
  def test_deploy_sync_function_no_params(serve_instance):
      @serve.deployment()
      def sync_d():
          return "sync!"
  
      serve.start()
  
      sync_d.deploy()
      assert requests.get("http://localhost:8000/sync_d").text == "sync!"
      assert ray.get(sync_d.get_handle().remote()) == "sync!"
  
  
  def test_deploy_async_function_no_params(serve_instance):
      @serve.deployment()
      async def async_d():
          await asyncio.sleep(5)
          return "async!"
  
      serve.start()
  
      async_d.deploy()
      assert requests.get("http://localhost:8000/async_d").text == "async!"
      assert ray.get(async_d.get_handle().remote()) == "async!"
  
  
  def test_deploy_sync_class_no_params(serve_instance):
      @serve.deployment
      class Counter:
          def __init__(self):
              self.count = 0
  
          def __call__(self):
              self.count += 1
              return {"count": self.count}
  
      serve.start()
      Counter.deploy()
  
      assert requests.get("http://127.0.0.1:8000/Counter").json() == {"count": 1}
      assert requests.get("http://127.0.0.1:8000/Counter").json() == {"count": 2}
      assert ray.get(Counter.get_handle().remote()) == {"count": 3}
  
  
  def test_deploy_async_class_no_params(serve_instance):
      @serve.deployment
      class AsyncCounter:
          async def __init__(self):
              await asyncio.sleep(5)
              self.count = 0
  
          async def __call__(self):
              self.count += 1
              await asyncio.sleep(5)
              return {"count": self.count}
  
      serve.start()
      AsyncCounter.deploy()
  
      assert requests.get("http://127.0.0.1:8000/AsyncCounter").json() == {"count": 1}
      assert requests.get("http://127.0.0.1:8000/AsyncCounter").json() == {"count": 2}
      assert ray.get(AsyncCounter.get_handle().remote()) == {"count": 3}
  
  
  def test_user_config(serve_instance):
      @serve.deployment("counter", num_replicas=2, user_config={"count": 123, "b": 2})
      class Counter:
          def __init__(self):
              self.count = 10
  
          def __call__(self, *args):
              return self.count, os.getpid()
  
          def reconfigure(self, config):
              self.count = config["count"]
  
      Counter.deploy()
      handle = Counter.get_handle()
  
      def check(val, num_replicas):
          pids_seen = set()
          for i in range(100):
              result = ray.get(handle.remote())
              if str(result[0]) != val:
                  return False
              pids_seen.add(result[1])
          return len(pids_seen) == num_replicas
  
      wait_for_condition(lambda: check("123", 2))
  
      Counter = Counter.options(num_replicas=3)
      Counter.deploy()
      wait_for_condition(lambda: check("123", 3))
  
      Counter = Counter.options(user_config={"count": 456})
      Counter.deploy()
      wait_for_condition(lambda: check("456", 3))
  
  
  def test_call_method(serve_instance):
      @serve.deployment(name="method")
      class CallMethod:
          def method(self, *args):
              return "hello"
  
      CallMethod.deploy()
  
      # Test HTTP path.
      resp = requests.get(
          "http://127.0.0.1:8000/method",
          timeout=1,
          headers={"X-SERVE-CALL-METHOD": "method"},
      )
      assert resp.text == "hello"
  
      # Test serve handle path.
      handle = CallMethod.get_handle()
      assert ray.get(handle.options(method_name="method").remote()) == "hello"
  
  
  def test_reject_duplicate_route(serve_instance):
      @serve.deployment(name="A", route_prefix="/api")
      class A:
          pass
  
      A.deploy()
      with pytest.raises(ValueError):
          A.options(name="B").deploy()
  
  
  def test_scaling_replicas(serve_instance):
      @serve.deployment(name="counter", num_replicas=2)
      class Counter:
          def __init__(self):
              self.count = 0
  
          def __call__(self, _):
              self.count += 1
              return self.count
  
      Counter.deploy()
  
      counter_result = []
      for _ in range(10):
          resp = requests.get("http://127.0.0.1:8000/counter").json()
          counter_result.append(resp)
  
      # If the load is shared among two replicas. The max result cannot be 10.
      assert max(counter_result) < 10
  
      Counter.options(num_replicas=1).deploy()
  
      counter_result = []
      for _ in range(10):
          resp = requests.get("http://127.0.0.1:8000/counter").json()
          counter_result.append(resp)
      # Give some time for a replica to spin down. But majority of the request
      # should be served by the only remaining replica.
      assert max(counter_result) - min(counter_result) > 6
  
  
  def test_delete_deployment(serve_instance):
      @serve.deployment(name="delete")
      def function(_):
          return "hello"
  
      function.deploy()
  
      assert requests.get("http://127.0.0.1:8000/delete").text == "hello"
  
      function.delete()
  
      @serve.deployment(name="delete")
      def function2(_):
          return "olleh"
  
      function2.deploy()
  
      for _ in range(10):
          try:
              assert requests.get("http://127.0.0.1:8000/delete").text == "olleh"
              break
          except AssertionError:
              time.sleep(0.5)  # Wait for the change to propagate.
      else:
          assert requests.get("http://127.0.0.1:8000/delete").text == "olleh"
  
  
  def test_starlette_request(serve_instance):
      @serve.deployment(name="api")
      async def echo_body(starlette_request):
          data = await starlette_request.body()
          return data
  
      echo_body.deploy()
  
      # Long string to test serialization of multiple messages.
      UVICORN_HIGH_WATER_MARK = 65536  # max bytes in one message
      long_string = "x" * 10 * UVICORN_HIGH_WATER_MARK
  
      resp = requests.post("http://127.0.0.1:8000/api", data=long_string).text
      assert resp == long_string
  
  
  def test_start_idempotent(serve_instance):
      @serve.deployment(name="start")
      def func(*args):
          pass
  
      func.deploy()
  
      assert "start" in serve.list_deployments()
      serve.start(detached=True)
      serve.start()
      serve.start(detached=True)
      serve.start()
      assert "start" in serve.list_deployments()
  
  
  def test_shutdown_destructor(serve_instance):
      signal = SignalActor.remote()
  
      @serve.deployment
      class A:
          def __del__(self):
              signal.send.remote()
  
      A.deploy()
      A.delete()
      ray.get(signal.wait.remote(), timeout=10)
  
      # If the destructor errored, it should be logged but also cleaned up.
      @serve.deployment
      class B:
          def __del__(self):
              raise RuntimeError("Opps")
  
      B.deploy()
      B.delete()
  
  
  if __name__ == "__main__":
      import sys
  
      sys.exit(pytest.main(["-v", "-s", __file__]))
  