code: def test_ray_start_block_and_stop(
      configure_lang, monkeypatch, tmp_path, cleanup_ray
  ):
      """Test `ray start` with `--block` as heads and workers and `ray stop`"""
      monkeypatch.setenv(
          "RAY_USAGE_STATS_CONFIG_PATH", str(tmp_path / "config.json")
      )
      runner = CliRunner(env={"RAY_USAGE_STATS_PROMPT_ENABLED": "0"})
  
      head_parent_conn, head_child_conn = mp.Pipe()
      worker_parent_conn, worker_child_conn = mp.Pipe()
  
      # Run `ray start --block --head` in another process and blocks
      head_proc = mp.Process(
          target=_start_ray_and_block,
          kwargs={
              "runner": runner,
              "child_conn": head_child_conn,
              "as_head": True,
          },
      )
      head_proc._start_method = "spawn"
  
      # Run `ray start --block --address=localhost:DEFAULT_PORT`
      worker_proc = mp.Process(
          target=_start_ray_and_block,
          kwargs={
              "runner": runner,
              "child_conn": worker_child_conn,
              "as_head": False,
          },
      )
      worker_proc._start_method = "spawn"
  
      try:
          # Run
          head_proc.start()
          worker_proc.start()
          print("head pid ", head_proc.pid)
          print("worker pid: ", worker_proc.pid)
  
          # Without sleep list_nodes sometimes failed.
          time.sleep(5)
          # Wait until all nodes are registered and started.
          wait_for_condition(lambda: len(list_nodes()) == 2)
          # When ray start --block is called, before it blocks,
          # it pings GCS to get the Ray & Python version. If GCS
          # is killed before that by ray stop, it can trigger GCS
          # timeout error which can break `ray start --block`.
          # To avoid this issue, we sleep enough before we stop ray.
          time.sleep(5)
  
          stop_result = runner.invoke(scripts.stop)
          print("result, ", stop_result.output)
          print("exit code, ", stop_result.exit_code)
          _die_on_error(stop_result)
  
          # Process with "--block" should be blocked forever w/o
          # termination by signals
          if not head_proc.is_alive():
              # NOTE(rickyyx): call recv() here is safe since the process
              # is guaranteed to be terminated.
              if head_proc.exitcode == 1:
                  assert False, (
                      "ray start --head --block is failed "
                      "due to unexpected component failures."
                  )
              _fail_if_false(
                  False,
                  head_parent_conn.recv(),
                  (
                      "`ray start --head --block` (head) should block forever even"
                      " though Ray subprocesses are stopped normally. But "
                      f"it exited with {head_proc.exitcode} early. \n"
                      f"Stop command: {stop_result.output}"
                  ),
              )
  
          if not worker_proc.is_alive():
              if worker_proc.exitcode == 1:
                  assert False, (
                      "ray start --address=<head_ip> --block` (worker) is failed "
                      "due to unexpected component failures."
                  )
              _fail_if_false(
                  False,
                  worker_parent_conn.recv(),
                  (
                      "`ray start --block` should block forever even"
                      " though Ray subprocesses are stopped normally. But"
                      f"it exited with {worker_proc.exitcode} already. \n"
                      f"Stop command: {stop_result.output}"
                  ),
              )
  
          # Stop both worker and head with SIGTERM
          head_proc.terminate()
          worker_proc.terminate()
  
          head_proc.join(15)
          worker_proc.join(15)
  
          worker_output = "cannot poll"
          head_output = "cannot poll"
          if head_parent_conn.poll(5):
              head_output = head_parent_conn.recv()
          if worker_parent_conn.poll(5):
              worker_output = worker_parent_conn.recv()
  
          print("head ", head_output)
          print("worker ", worker_output)
          assert not head_proc.is_alive(), "head node is not killed."
          assert (
              not worker_proc.is_alive()
          ), "worker node is not killed."
  
          _fail_if_false(
              head_proc.exitcode == 0,
              head_output,
              f"Head process failed unexpectedly({head_proc.exitcode})",
          )
          _fail_if_false(
              worker_proc.exitcode == 0,
              worker_output,
              f"Worker process failed unexpectedly({worker_proc.exitcode})",
          )
      finally:
          head_proc.kill()
          worker_proc.kill()
  
imports: import glob
  import multiprocessing
  import multiprocessing.connection
  import os
  import re
  import sys
  import tempfile
  import time
  import uuid
  from contextlib import contextmanager
  from pathlib import Path
  from typing import Optional
  from unittest.mock import MagicMock, patch
  import moto
  import pytest
  from click.testing import CliRunner
  from moto import mock_ec2, mock_iam
  from testfixtures import Replacer
  from testfixtures.popen import MockPopen, PopenBehaviour
  import ray
  import ray.autoscaler._private.aws.config
  import ray.autoscaler._private.constants
  import ray._private.ray_constants
  import ray.scripts.scripts
  from ray._private.test_utils import wait_for_condition
  from ray.cluster_utils import cluster_not_supported
  from ray.util.state import list_nodes
  import psutil
label_desc: Use image node_id 20 for calling a global function with example usage: test_ray_start_block_and_stop(configure_lang, monkeypatch, tmp_path, cleanup_ray) without return types