hunk.code_window:             ["serve", "status", "-a", "http://localhost:52365/"]
           )
           status = yaml.safe_load(cli_output)["applications"]["default"]
           assert status["status"] == "DEPLOY_FAILED"
           assert "x = (1 + 2" in status["message"]
           return True
       wait_for_condition(check_for_failed_deployment)
  
hunk.labels: keep keep keep add keep keep keep keep keep
hunk.after_edit:         assert "Traceback (most recent call last)" in status["message"]
  
hunk.file_path: python/ray/serve/tests/test_cli.py
hunk.type: add
file: """Test the reduce API."""
  import pytest
  import numpy as np
  import ray
  from ray.util.collective.types import Backend, ReduceOp
  
  from ray.util.collective.tests.cpu_util import create_collective_workers
  
  
  @pytest.mark.parametrize("backend", [Backend.GLOO])
  @pytest.mark.parametrize("group_name", ["default", "test", "123?34!"])
  @pytest.mark.parametrize("dst_rank", [0, 2, 5, 6, 7])
  def test_reduce_different_name(
      ray_start_distributed_2_nodes, group_name, backend, dst_rank
  ):
      world_size = 8
      actors, _ = create_collective_workers(
          num_workers=world_size, group_name=group_name, backend=backend
      )
      results = ray.get([a.do_reduce.remote(group_name, dst_rank) for a in actors])
      for i in range(world_size):
          if i == dst_rank:
              assert (results[i] == np.ones((10,), dtype=np.float32) * world_size).all()
          else:
              assert (results[i] == np.ones((10,), dtype=np.float32)).all()
  
  
  @pytest.mark.parametrize("backend", [Backend.GLOO])
  @pytest.mark.parametrize("array_size", [2, 2**5, 2**10, 2**15, 2**20])
  @pytest.mark.parametrize("dst_rank", [0, 2, 5, 6, 7])
  def test_reduce_different_array_size(
      ray_start_distributed_2_nodes, backend, array_size, dst_rank
  ):
      world_size = 8
      actors, _ = create_collective_workers(world_size, backend=backend)
      ray.wait(
          [a.set_buffer.remote(np.ones(array_size, dtype=np.float32)) for a in actors]
      )
      results = ray.get([a.do_reduce.remote(dst_rank=dst_rank) for a in actors])
      for i in range(world_size):
          if i == dst_rank:
              assert (
                  results[i] == np.ones((array_size,), dtype=np.float32) * world_size
              ).all()
          else:
              assert (results[i] == np.ones((array_size,), dtype=np.float32)).all()
  
  
  @pytest.mark.parametrize("backend", [Backend.GLOO])
  @pytest.mark.parametrize("dst_rank", [0, 2, 5, 6, 7])
  def test_reduce_different_op(ray_start_distributed_2_nodes, backend, dst_rank):
      world_size = 8
      actors, _ = create_collective_workers(world_size, backend=backend)
  
      # check product
      ray.wait(
          [
              a.set_buffer.remote(np.ones(10, dtype=np.float32) * (i + 2))
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.PRODUCT) for a in actors]
      )
  
      product = 1
      for i in range(world_size):
          product = product * (i + 2)
      for i in range(world_size):
          if i == dst_rank:
              assert (results[i] == np.ones((10,), dtype=np.float32) * product).all()
          else:
              assert (results[i] == np.ones((10,), dtype=np.float32) * (i + 2)).all()
      # check min
      ray.wait(
          [
              a.set_buffer.remote(np.ones(10, dtype=np.float32) * (i + 2))
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.MIN) for a in actors]
      )
      for i in range(world_size):
          if i == dst_rank:
              assert (results[i] == np.ones((10,), dtype=np.float32) * 2).all()
          else:
              assert (results[i] == np.ones((10,), dtype=np.float32) * (i + 2)).all()
  
      # check max
      ray.wait(
          [
              a.set_buffer.remote(np.ones(10, dtype=np.float32) * (i + 2))
              for i, a in enumerate(actors)
          ]
      )
      results = ray.get(
          [a.do_reduce.remote(dst_rank=dst_rank, op=ReduceOp.MAX) for a in actors]
      )
      for i in range(world_size):
          if i == dst_rank:
              assert (
                  results[i] == np.ones((10,), dtype=np.float32) * (world_size + 1)
              ).all()
          else:
              assert (results[i] == np.ones((10,), dtype=np.float32) * (i + 2)).all()
  
  
  @pytest.mark.parametrize("backend", [Backend.GLOO])
  @pytest.mark.parametrize("dst_rank", [0, 2, 5, 6, 7])
  def test_reduce_torch_numpy(ray_start_distributed_2_nodes, backend, dst_rank):
      import torch
  
      world_size = 8
      actors, _ = create_collective_workers(world_size, backend=backend)
      ray.get(
          [
              actors[1].set_buffer.remote(
                  torch.ones(
                      10,
                  )
              )
          ]
      )
      results = ray.get([a.do_reduce.remote(dst_rank=dst_rank) for a in actors])
      if dst_rank == 0:
          assert (results[0] == np.ones((10,)) * world_size).all()
          assert (results[1] == torch.ones((10,))).all()
      else:
          assert (results[0] == np.ones((10,))).all()
          assert (results[1] == torch.ones((10,))).all()
  
  
  @pytest.mark.parametrize("backend", [Backend.GLOO])
  def test_reduce_invalid_rank(ray_start_distributed_2_nodes, backend, dst_rank=9):
      world_size = 8
      actors, _ = create_collective_workers(world_size, backend=backend)
      with pytest.raises(ValueError):
          _ = ray.get([a.do_reduce.remote(dst_rank=dst_rank) for a in actors])
  
  
  if __name__ == "__main__":
      import pytest
      import sys
  
      sys.exit(pytest.main(["-v", "-x", __file__]))
  