[Serve][Part1] Update the tests to use graph deploy (ray-project#26310)

Signed-off-by: Stefan van der Kleij <[email protected]>

python/ray/serve/deployment_function_node.py

@@ -29,7 +29,6 @@ def __init__(
             func_options,
             other_args_to_resolve=other_args_to_resolve,
         )
-
         if "deployment_schema" in self._bound_other_args_to_resolve:
             deployment_schema: DeploymentSchema = self._bound_other_args_to_resolve[
                 "deployment_schema"

python/ray/serve/deployment_graph_build.py

@@ -180,6 +180,7 @@ def replace_with_handle(node):
         deployment_schema: DeploymentSchema = dag_node._bound_other_args_to_resolve[
             "deployment_schema"
         ]
+
         deployment_shell: Deployment = schema_to_deployment(deployment_schema)
 
         # Prefer user specified name to override the generated one.

python/ray/serve/tests/test_advanced.py

@@ -14,9 +14,7 @@ def sleeper():
         while True:
             time.sleep(1000)
 
-    sleeper.deploy()
-
-    handle = sleeper.get_handle()
+    handle = serve.run(sleeper.bind())
     ref = handle.remote()
     sleeper.delete()
 
@@ -37,8 +35,7 @@ class Wait:
         async def __call__(self, signal_actor):
             await signal_actor.wait.remote()
 
-    Wait.deploy()
-    handle = Wait.get_handle()
+    handle = serve.run(Wait.bind())
     refs = [handle.remote(signal) for _ in range(10)]
 
     # Wait for all the queries to be enqueued
@@ -95,8 +92,8 @@ def __init__(self):
         def __call__(self):
             return "Ready"
 
-    LongStartingServable.deploy()
-    ray.get(LongStartingServable.get_handle().remote(), timeout=10)
+    handle = serve.run(LongStartingServable.bind())
+    ray.get(handle.remote(), timeout=10)
 
 
 if __name__ == "__main__":

python/ray/serve/tests/test_autoscaling_metrics.py

@@ -87,8 +87,7 @@ class A:
         def __call__(self):
             time.sleep(0.5)
 
-    A.deploy()
-    handle = A.get_handle()
+    handle = serve.run(A.bind())
     [handle.remote() for _ in range(100)]
 
     # Wait for metrics to propagate

python/ray/serve/tests/test_autoscaling_policy.py

@@ -195,12 +195,11 @@ class A:
         def __call__(self):
             ray.get(signal.wait.remote())
 
-    A.deploy()
+    handle = serve.run(A.bind())
 
     controller = serve_instance._controller
     start_time = get_deployment_start_time(controller, A)
 
-    handle = A.get_handle()
     [handle.remote() for _ in range(100)]
 
     # scale up one more replica from min_replicas
@@ -241,12 +240,11 @@ class A:
         def __call__(self):
             ray.get(signal.wait.remote())
 
-    A.deploy()
+    handle = serve.run(A.bind())
 
     controller = serve_instance._controller
     start_time = get_deployment_start_time(controller, A)
 
-    handle = A.get_handle()
     [handle.remote() for _ in range(100)]
 
     # scale up one more replica from min_replicas
@@ -280,7 +278,7 @@ class A:
         def __call__(self):
             return "ok!"
 
-    A.deploy()
+    serve.run(A.bind())
 
     controller = serve_instance._controller
     assert get_num_running_replicas(controller, A) == 2
@@ -607,12 +605,11 @@ class A:
         def __call__(self):
             ray.get(signal.wait.remote())
 
-    A.deploy()
+    handle = serve.run(A.bind())
 
     controller = serve_instance._controller
     start_time = get_deployment_start_time(controller, A)
 
-    handle = A.get_handle()
     [handle.remote() for _ in range(100)]
 
     wait_for_condition(lambda: get_num_running_replicas(controller, A) >= 2)
@@ -665,12 +662,12 @@ class A:
         def __call__(self):
             ray.get(signal.wait.remote())
 
-    A.deploy()
+    handle = serve.run(A.bind())
 
     controller = serve_instance._controller
     start_time = get_deployment_start_time(controller, A)
 
-    handle = A.get_handle()
+    A.get_handle()
     [handle.remote() for _ in range(50)]
 
     wait_for_condition(
@@ -720,15 +717,14 @@ class A:
         def __call__(self):
             ray.get(signal.wait.remote())
 
-    A.deploy()
+    handle = serve.run(A.bind())
     print("Deployed A with min_replicas 1 and max_replicas 10.")
 
     controller = serve_instance._controller
     start_time = get_deployment_start_time(controller, A)
 
     assert get_num_running_replicas(controller, A) == 0
 
-    handle = A.get_handle()
     [handle.remote() for _ in range(400)]
     print("Issued 400 requests.")
 
@@ -742,17 +738,19 @@ def __call__(self):
     time.sleep(3)
     print("Issued 458 requests. Request routing in-progress.")
 
-    A.options(
-        _autoscaling_config={
-            "metrics_interval_s": 0.1,
-            "min_replicas": 2,
-            "max_replicas": 20,
-            "look_back_period_s": 0.2,
-            "downscale_delay_s": 0.2,
-            "upscale_delay_s": 0.2,
-        },
-        version="v1",
-    ).deploy()
+    serve.run(
+        A.options(
+            _autoscaling_config={
+                "metrics_interval_s": 0.1,
+                "min_replicas": 2,
+                "max_replicas": 20,
+                "look_back_period_s": 0.2,
+                "downscale_delay_s": 0.2,
+                "upscale_delay_s": 0.2,
+            },
+            version="v1",
+        ).bind()
+    )
     print("Redeployed A.")
 
     wait_for_condition(lambda: get_num_running_replicas(controller, A) >= 20)
@@ -774,17 +772,19 @@ def __call__(self):
     assert get_deployment_start_time(controller, A) == start_time
 
     # scale down to 0
-    A.options(
-        _autoscaling_config={
-            "metrics_interval_s": 0.1,
-            "min_replicas": 0,
-            "max_replicas": 20,
-            "look_back_period_s": 0.2,
-            "downscale_delay_s": 0.2,
-            "upscale_delay_s": 0.2,
-        },
-        version="v1",
-    ).deploy()
+    serve.run(
+        A.options(
+            _autoscaling_config={
+                "metrics_interval_s": 0.1,
+                "min_replicas": 0,
+                "max_replicas": 20,
+                "look_back_period_s": 0.2,
+                "downscale_delay_s": 0.2,
+                "upscale_delay_s": 0.2,
+            },
+            version="v1",
+        ).bind()
+    )
     print("Redeployed A.")
 
     wait_for_condition(lambda: get_num_running_replicas(controller, A) < 1)

python/ray/serve/tests/test_batching.py

@@ -21,10 +21,9 @@ async def handle_batch(self, requests):
         async def __call__(self, request):
             return await self.handle_batch(request)
 
-    BatchingExample.deploy()
+    handle = serve.run(BatchingExample.bind())
 
     future_list = []
-    handle = BatchingExample.get_handle()
     for _ in range(20):
         f = handle.remote(1)
         future_list.append(f)
@@ -50,9 +49,8 @@ async def __call__(self, request):
             return await self.handle_batch(request)
 
     # Set the max batch size.
-    NoListReturned.deploy()
+    handle = serve.run(NoListReturned.bind())
 
-    handle = NoListReturned.get_handle()
     with pytest.raises(ray.exceptions.RayTaskError):
         assert ray.get(handle.remote(1))
 

python/ray/serve/tests/test_cluster.py

@@ -97,7 +97,7 @@ def get_pids(expected, timeout=30):
     serve.start(detached=True)
 
     print("Initial deploy.")
-    D.deploy()
+    serve.run(D.bind())
     pids1 = get_pids(5)
 
     # Remove the node. There should still be three replicas running.

python/ray/serve/tests/test_constructor_failure.py

@@ -19,10 +19,10 @@ async def serve(self, request):
             return "hi"
 
     with pytest.raises(RuntimeError):
-        ConstructorFailureDeploymentOneReplica.deploy()
+        serve.run(ConstructorFailureDeploymentOneReplica.bind())
 
     # Assert no replicas are running in deployment deployment after failed
-    # deploy() call
+    # deploy call
     deployment_dict = ray.get(serve_instance._controller._all_running_replicas.remote())
     assert deployment_dict["ConstructorFailureDeploymentOneReplica"] == []
 
@@ -36,10 +36,10 @@ async def serve(self, request):
             return "hi"
 
     with pytest.raises(RuntimeError):
-        ConstructorFailureDeploymentTwoReplicas.deploy()
+        serve.run(ConstructorFailureDeploymentTwoReplicas.bind())
 
     # Assert no replicas are running in deployment deployment after failed
-    # deploy() call
+    # deploy call
     deployment_dict = ray.get(serve_instance._controller._all_running_replicas.remote())
     assert deployment_dict["ConstructorFailureDeploymentTwoReplicas"] == []
 
@@ -70,10 +70,10 @@ def __init__(self):
             async def serve(self, request):
                 return "hi"
 
-        PartialConstructorFailureDeployment.deploy()
+        serve.run(PartialConstructorFailureDeployment.bind())
 
     # Assert 2 replicas are running in deployment deployment after partially
-    # successful deploy() call
+    # successful deploy call
     deployment_dict = ray.get(serve_instance._controller._all_running_replicas.remote())
     assert len(deployment_dict["PartialConstructorFailureDeployment"]) == 2
 
@@ -97,9 +97,9 @@ def __init__(self):
             async def serve(self, request):
                 return "hi"
 
-        TransientConstructorFailureDeployment.deploy()
+        serve.run(TransientConstructorFailureDeployment.bind())
     # Assert 2 replicas are running in deployment deployment after partially
-    # successful deploy() call with transient error
+    # successful deploy call with transient error
     deployment_dict = ray.get(serve_instance._controller._all_running_replicas.remote())
     assert len(deployment_dict["TransientConstructorFailureDeployment"]) == 2
 

python/ray/serve/tests/test_controller.py

@@ -17,7 +17,7 @@ def test_redeploy_start_time(serve_instance):
     def test(_):
         return "1"
 
-    test.deploy()
+    serve.run(test.bind())
     deployment_route = DeploymentRoute.FromString(
         ray.get(controller.get_deployment_info.remote("test"))
     )
@@ -30,7 +30,7 @@ def test(_):
     def test(_):
         return "2"
 
-    test.deploy()
+    serve.run(test.bind())
     deployment_route = DeploymentRoute.FromString(
         ray.get(controller.get_deployment_info.remote("test"))
     )