code: def test_sigterm_while_ray_get_and_wait(shutdown_only, is_get):
      """Verify when sigterm is received while running
      ray.get, it will clean up the worker process properly.
      """
  
      @ray.remote(max_retries=0)
      def f():
          time.sleep(300)
  
      @ray.remote(max_retries=0)
      def ray_get_wait_task():
          ref = f.remote()
          if is_get:
              ray.get(ref)
          else:
              ray.wait([ref])
  
      # TODO(sang): The task failure is not properly propagated.
      r = ray_get_wait_task.remote()
  
      def wait_for_task():
          t = list_tasks(filters=[("name", "=", "ray_get_wait_task")])[
              0
          ]
          return t.state == "RUNNING"
  
      wait_for_condition(wait_for_task)
  
      # Kill a task that's blocked by ray.get
      t = list_tasks(filters=[("name", "=", "ray_get_wait_task")])[0]
      os.kill(t.worker_pid, signal.SIGTERM)
  
      with pytest.raises(ray.exceptions.WorkerCrashedError):
          ray.get(r)
  
      def verify():
          t = list_tasks(filters=[("name", "=", "ray_get_wait_task")])[
              0
          ]
          w = get_worker(t.worker_id)
          assert t.state == "FAILED"
          assert w.exit_type == "SYSTEM_ERROR"
          assert "SIGTERM" in w.exit_detail
          return True
  
      wait_for_condition(verify)
  
imports: import sys
  import time
  import platform
  import os
  import signal
  import multiprocessing
  import pytest
  import ray
  import psutil
  from ray._private.test_utils import wait_for_condition, run_string_as_driver_nonblocking
  from ray.util.state import get_worker, list_tasks
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 15 for calling a global function with example usage: test_sigterm_while_ray_get_and_wait(shutdown_only, is_get) and returns: unknown, True