hunk.code_window:         if isinstance(module, str):
               # module_path is a local path or a URI.
               module_path = module
           elif isinstance(module, ModuleType):
               # NOTE(edoakes): Python allows some installed Python packages to
               # be split into multiple directories. We could probably handle
  
hunk.labels: keep keep add keep keep keep
hunk.after_edit:         elif isinstance(module, Path):
               module_path = str(module)
  
hunk.file_path: python/ray/_private/runtime_env/py_modules.py
hunk.type: add
file: """
  The test file for all standalone tests that doesn't
  requires a shared Serve instance.
  """
  import os
  import subprocess
  import sys
  import socket
  from typing import Optional
  from tempfile import mkstemp
  
  import pytest
  import pydantic
  import requests
  
  import ray
  from ray import serve
  from ray.cluster_utils import Cluster
  from ray.serve.constants import SERVE_ROOT_URL_ENV_KEY, SERVE_PROXY_NAME
  from ray.serve.exceptions import RayServeException
  from ray.serve.utils import (block_until_http_ready, get_all_node_ids,
                               format_actor_name)
  from ray.serve.config import HTTPOptions
  from ray.serve.api import _get_global_client
  from ray._private.test_utils import (run_string_as_driver, wait_for_condition,
                                       convert_actor_state)
  from ray._private.services import new_port
  import ray._private.gcs_utils as gcs_utils
  
  # Explicitly importing it here because it is a ray core tests utility (
  # not in the tree)
  from ray.tests.conftest import ray_start_with_dashboard  # noqa: F401
  
  
  @pytest.fixture
  def ray_cluster():
      cluster = Cluster()
      yield Cluster()
      serve.shutdown()
      ray.shutdown()
      cluster.shutdown()
  
  
  def test_shutdown(ray_shutdown):
      ray.init(num_cpus=16)
      serve.start(http_options=dict(port=8003))
  
      @serve.deployment
      def f():
          pass
  
      f.deploy()
  
      serve_controller_name = serve.api._global_client._controller_name
      actor_names = [
          serve_controller_name,
          format_actor_name(SERVE_PROXY_NAME,
                            serve.api._global_client._controller_name,
                            get_all_node_ids()[0][0])
      ]
  
      def check_alive():
          alive = True
          for actor_name in actor_names:
              try:
                  if actor_name == serve_controller_name:
                      ray.get_actor(
                          actor_name,
                          namespace=ray.get_runtime_context().namespace)
                  else:
                      ray.get_actor(actor_name)
              except ValueError:
                  alive = False
          return alive
  
      wait_for_condition(check_alive)
  
      serve.shutdown()
      with pytest.raises(RayServeException):
          serve.list_deployments()
  
      def check_dead():
          for actor_name in actor_names:
              try:
                  if actor_name == serve_controller_name:
                      ray.get_actor(
                          actor_name,
                          namespace=ray.get_runtime_context().namespace)
                  else:
                      ray.get_actor(actor_name)
                  return False
              except ValueError:
                  pass
          return True
  
      wait_for_condition(check_dead)
  
  
  def test_detached_deployment(ray_cluster):
      # https://github.com/ray-project/ray/issues/11437
  
      cluster = ray_cluster
      head_node = cluster.add_node(num_cpus=6)
  
      # Create first job, check we can run a simple serve endpoint
      ray.init(head_node.address, namespace="serve")
      first_job_id = ray.get_runtime_context().job_id
      serve.start(detached=True)
  
      @serve.deployment
      def f(*args):
          return "hello"
  
      f.deploy()
      assert ray.get(f.get_handle().remote()) == "hello"
  
      serve.api._global_client = None
      ray.shutdown()
  
      # Create the second job, make sure we can still create new deployments.
      ray.init(head_node.address, namespace="serve")
      assert ray.get_runtime_context().job_id != first_job_id
  
      @serve.deployment
      def g(*args):
          return "world"
  
      g.deploy()
      assert ray.get(g.get_handle().remote()) == "world"
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  @pytest.mark.parametrize("detached", [True, False])
  def test_connect(detached, ray_shutdown):
      # Check that you can make API calls from within a deployment for both
      # detached and non-detached instances.
      ray.init(num_cpus=16, namespace="serve")
      serve.start(detached=detached)
  
      @serve.deployment
      def connect_in_deployment(*args):
          connect_in_deployment.options(name="deployment-ception").deploy()
  
      connect_in_deployment.deploy()
      ray.get(connect_in_deployment.get_handle().remote())
      assert "deployment-ception" in serve.list_deployments()
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  @pytest.mark.parametrize("controller_cpu", [True, False])
  @pytest.mark.parametrize("num_proxy_cpus", [0, 1, 2])
  def test_dedicated_cpu(controller_cpu, num_proxy_cpus, ray_cluster):
      cluster = ray_cluster
      num_cluster_cpus = 8
      head_node = cluster.add_node(num_cpus=num_cluster_cpus)
  
      ray.init(head_node.address)
      wait_for_condition(
          lambda: ray.cluster_resources().get("CPU") == num_cluster_cpus)
  
      num_cpus_used = int(controller_cpu) + num_proxy_cpus
  
      serve.start(
          dedicated_cpu=controller_cpu,
          http_options=HTTPOptions(num_cpus=num_proxy_cpus))
      available_cpus = num_cluster_cpus - num_cpus_used
      wait_for_condition(
          lambda: (ray.available_resources().get("CPU") == available_cpus))
      serve.shutdown()
      ray.shutdown()
  
  
  @pytest.mark.skipif(
      not hasattr(socket, "SO_REUSEPORT"),
      reason=("Port sharing only works on newer verion of Linux. "
              "This test can only be ran when port sharing is supported."))
  def test_multiple_routers(ray_cluster):
      cluster = ray_cluster
      head_node = cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
  
      ray.init(head_node.address)
      node_ids = ray.state.node_ids()
      assert len(node_ids) == 2
      serve.start(http_options=dict(port=8005, location="EveryNode"))
  
      def get_proxy_names():
          proxy_names = []
          for node_id, _ in get_all_node_ids():
              proxy_names.append(
                  format_actor_name(SERVE_PROXY_NAME,
                                    serve.api._global_client._controller_name,
                                    node_id))
          return proxy_names
  
      wait_for_condition(lambda: len(get_proxy_names()) == 2)
      proxy_names = get_proxy_names()
  
      # Two actors should be started.
      def get_first_two_actors():
          try:
              ray.get_actor(proxy_names[0])
              ray.get_actor(proxy_names[1])
              return True
          except ValueError:
              return False
  
      wait_for_condition(get_first_two_actors)
  
      # Wait for the actors to come up.
      ray.get(block_until_http_ready.remote("http://127.0.0.1:8005/-/routes"))
  
      # Kill one of the servers, the HTTP server should still function.
      ray.kill(ray.get_actor(get_proxy_names()[0]), no_restart=True)
      ray.get(block_until_http_ready.remote("http://127.0.0.1:8005/-/routes"))
  
      # Add a new node to the cluster. This should trigger a new router to get
      # started.
      new_node = cluster.add_node()
  
      wait_for_condition(lambda: len(get_proxy_names()) == 3)
      third_proxy = get_proxy_names()[2]
  
      def get_third_actor():
          try:
              ray.get_actor(third_proxy)
              return True
          # IndexErrors covers when cluster resources aren't updated yet.
          except (IndexError, ValueError):
              return False
  
      wait_for_condition(get_third_actor)
  
      # Remove the newly-added node from the cluster. The corresponding actor
      # should be removed as well.
      cluster.remove_node(new_node)
  
      def third_actor_removed():
          try:
              ray.get_actor(third_proxy)
              return False
          except ValueError:
              return True
  
      # Check that the actor is gone and the HTTP server still functions.
      wait_for_condition(third_actor_removed)
      ray.get(block_until_http_ready.remote("http://127.0.0.1:8005/-/routes"))
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_middleware(ray_shutdown):
      from starlette.middleware import Middleware
      from starlette.middleware.cors import CORSMiddleware
  
      port = new_port()
      serve.start(
          http_options=dict(
              port=port,
              middlewares=[
                  Middleware(
                      CORSMiddleware, allow_origins=["*"], allow_methods=["*"])
              ]))
      ray.get(block_until_http_ready.remote(f"http://127.0.0.1:{port}/-/routes"))
  
      # Snatched several test cases from Starlette
      # https://github.com/encode/starlette/blob/master/tests/
      # middleware/test_cors.py
      headers = {
          "Origin": "https://example.org",
          "Access-Control-Request-Method": "GET",
      }
      root = f"http://localhost:{port}"
      resp = requests.options(root, headers=headers)
      assert resp.headers["access-control-allow-origin"] == "*"
  
      resp = requests.get(f"{root}/-/routes", headers=headers)
      assert resp.headers["access-control-allow-origin"] == "*"
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_http_root_url(ray_shutdown):
      @serve.deployment
      def f(_):
          pass
  
      root_url = "https://my.domain.dev/prefix"
  
      port = new_port()
      os.environ[SERVE_ROOT_URL_ENV_KEY] = root_url
      serve.start(http_options=dict(port=port))
      f.deploy()
      assert f.url == root_url + "/f"
      serve.shutdown()
      ray.shutdown()
      del os.environ[SERVE_ROOT_URL_ENV_KEY]
  
      port = new_port()
      serve.start(http_options=dict(port=port))
      f.deploy()
      assert f.url != root_url + "/f"
      assert f.url == f"http://127.0.0.1:{port}/f"
      serve.shutdown()
      ray.shutdown()
  
      ray.init(runtime_env={"env_vars": {SERVE_ROOT_URL_ENV_KEY: root_url}})
      port = new_port()
      serve.start(http_options=dict(port=port))
      f.deploy()
      assert f.url == root_url + "/f"
      serve.shutdown()
      ray.shutdown()
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_http_proxy_fail_loudly(ray_shutdown):
      # Test that if the http server fail to start, serve.start should fail.
      with pytest.raises(ValueError):
          serve.start(http_options={"host": "bad.ip.address"})
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_no_http(ray_shutdown):
      # The following should have the same effect.
      options = [
          {
              "http_options": {
                  "host": None
              }
          },
          {
              "http_options": {
                  "location": None
              }
          },
          {
              "http_options": {
                  "location": "NoServer"
              }
          },
      ]
  
      ray.init(num_cpus=16)
      for i, option in enumerate(options):
          print(f"[{i+1}/{len(options)}] Running with {option}")
          serve.start(**option)
  
          # Only controller actor should exist
          live_actors = [
              actor for actor in ray.state.actors().values() if actor["State"] ==
              convert_actor_state(gcs_utils.ActorTableData.ALIVE)
          ]
          assert len(live_actors) == 1
          controller = serve.api._global_client._controller
          assert len(ray.get(controller.get_http_proxies.remote())) == 0
  
          # Test that the handle still works.
          @serve.deployment
          def hello(*args):
              return "hello"
  
          hello.deploy()
  
          assert ray.get(hello.get_handle().remote()) == "hello"
          serve.shutdown()
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_http_head_only(ray_cluster):
      cluster = ray_cluster
      head_node = cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
  
      ray.init(head_node.address)
      node_ids = ray.state.node_ids()
      assert len(node_ids) == 2
  
      serve.start(http_options={"port": new_port(), "location": "HeadOnly"})
  
      # Only the controller and head node actor should be started
      assert len(ray.state.actors()) == 2
  
      # They should all be placed on the head node
      cpu_per_nodes = {
          r["CPU"]
          for r in ray.state.state._available_resources_per_node().values()
      }
      assert cpu_per_nodes == {4, 4}
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  @pytest.mark.skipif(
      not hasattr(socket, "SO_REUSEPORT"),
      reason=("Port sharing only works on newer verion of Linux. "
              "This test can only be ran when port sharing is supported."))
  def test_fixed_number_proxies(ray_cluster):
      cluster = ray_cluster
      head_node = cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
  
      ray.init(head_node.address)
      node_ids = ray.state.node_ids()
      assert len(node_ids) == 3
  
      with pytest.raises(
              pydantic.ValidationError,
              match="you must specify the `fixed_number_replicas` parameter."):
          serve.start(http_options={
              "location": "FixedNumber",
          })
  
      serve.start(http_options={
          "port": new_port(),
          "location": "FixedNumber",
          "fixed_number_replicas": 2
      })
  
      # Only the controller and two http proxy should be started.
      controller_handle = _get_global_client()._controller
      node_to_http_actors = ray.get(controller_handle.get_http_proxies.remote())
      assert len(node_to_http_actors) == 2
  
      serve.shutdown()
      ray.shutdown()
      cluster.shutdown()
  
  
  def test_serve_shutdown(ray_shutdown):
      ray.init(namespace="serve")
      serve.start(detached=True)
  
      @serve.deployment
      class A:
          def __call__(self, *args):
              return "hi"
  
      A.deploy()
  
      assert len(serve.list_deployments()) == 1
  
      serve.shutdown()
      serve.start(detached=True)
  
      assert len(serve.list_deployments()) == 0
  
      A.deploy()
  
      assert len(serve.list_deployments()) == 1
  
  
  def test_detached_namespace_default_ray_init(ray_shutdown):
      # Can start detached instance when ray is not initialized.
      serve.start(detached=True)
  
  
  def test_detached_instance_in_non_anonymous_namespace(ray_shutdown):
      # Can start detached instance in non-anonymous namespace.
      ray.init(namespace="foo")
      serve.start(detached=True)
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  @pytest.mark.parametrize("namespace", [None, "test_namespace"])
  @pytest.mark.parametrize("detached", [True, False])
  def test_serve_controller_namespace(ray_shutdown, namespace: Optional[str],
                                      detached: bool):
      """
      Tests the serve controller is started in the current namespace if not
      anonymous or in the "serve" namespace if no namespace is specified.
      When the controller is started in the "serve" namespace, this also tests
      that we can get the serve controller from another namespace.
      """
  
      ray.init(namespace=namespace)
      serve.start(detached=detached)
      client = serve.api._global_client
      if namespace:
          controller_namespace = namespace
      elif detached:
          controller_namespace = "serve"
      else:
          controller_namespace = ray.get_runtime_context().namespace
  
      assert ray.get_actor(
          client._controller_name, namespace=controller_namespace)
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows")
  def test_checkpoint_isolation_namespace(ray_shutdown):
      info = ray.init(namespace="test_namespace1")
  
      address = info["redis_address"]
  
      driver_template = """
  import ray
  from ray import serve
  
  ray.init(address="{address}", namespace="{namespace}")
  
  serve.start(detached=True, http_options={{"port": {port}}})
  
  @serve.deployment
  class A:
      pass
  
  A.deploy()"""
  
      run_string_as_driver(
          driver_template.format(
              address=address, namespace="test_namespace1", port=8000))
      run_string_as_driver(
          driver_template.format(
              address=address, namespace="test_namespace2", port=8001))
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows.")
  def test_local_store_recovery(ray_shutdown):
      _, tmp_path = mkstemp()
  
      @serve.deployment
      def hello(_):
          return "hello"
  
      # https://github.com/ray-project/ray/issues/19987
      @serve.deployment
      def world(_):
          return "world"
  
      def check(name):
          try:
              resp = requests.get(f"http://localhost:8000/{name}")
              assert resp.text == name
              return True
          except Exception:
              return False
  
      # https://github.com/ray-project/ray/issues/20159
      # https://github.com/ray-project/ray/issues/20158
      def clean_up_leaked_processes():
          import psutil
          for proc in psutil.process_iter():
              try:
                  cmdline = " ".join(proc.cmdline())
                  if "ray::" in cmdline:
                      print(f"Kill {proc} {cmdline}")
                      proc.kill()
              except Exception:
                  pass
  
      def crash():
          subprocess.call(["ray", "stop", "--force"])
          clean_up_leaked_processes()
          ray.shutdown()
          serve.shutdown()
  
      serve.start(detached=True, _checkpoint_path=f"file://{tmp_path}")
      hello.deploy()
      world.deploy()
      assert check("hello")
      assert check("world")
      crash()
  
      # Simulate a crash
  
      serve.start(detached=True, _checkpoint_path=f"file://{tmp_path}")
      wait_for_condition(lambda: check("hello"))
      # wait_for_condition(lambda: check("world"))
      crash()
  
  
  @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows.")
  @pytest.mark.parametrize(
      "ray_start_with_dashboard", [{
          "num_cpus": 4
      }], indirect=True)
  def test_snapshot_always_written_to_internal_kv(
          ray_start_with_dashboard,  # noqa: F811
          ray_shutdown):
      # https://github.com/ray-project/ray/issues/19752
      _, tmp_path = mkstemp()
  
      @serve.deployment()
      def hello(_):
          return "hello"
  
      def check():
          try:
              resp = requests.get("http://localhost:8000/hello")
              assert resp.text == "hello"
              return True
          except Exception:
              return False
  
      serve.start(detached=True, _checkpoint_path=f"file://{tmp_path}")
      hello.deploy()
      check()
  
      webui_url = ray_start_with_dashboard["webui_url"]
  
      def get_deployment_snapshot():
          snapshot = requests.get(f"http://{webui_url}/api/snapshot").json()[
              "data"]["snapshot"]
          return snapshot["deployments"]
  
      # Make sure /api/snapshot return non-empty deployment status.
      def verify_snapshot():
          return get_deployment_snapshot() != {}
  
      wait_for_condition(verify_snapshot)
  
      # Sanity check the snapshot is correct
      snapshot = get_deployment_snapshot()
      assert len(snapshot) == 1
      hello_deployment = list(snapshot.values())[0]
      assert hello_deployment["name"] == "hello"
      assert hello_deployment["status"] == "RUNNING"
  
  
  if __name__ == "__main__":
      sys.exit(pytest.main(["-v", "-s", __file__]))
  