code: def test_component_activities_hook(
      set_ray_cluster_activity_hook, call_ray_start
  ):
      """
      Tests /api/component_activities returns correctly for various
      responses of RAY_CLUSTER_ACTIVITY_HOOK.
  
      Verify no active drivers are correctly reflected in response.
      """
      external_hook = set_ray_cluster_activity_hook
  
      response = requests.get(
          "http://127.0.0.1:8265/api/component_activities"
      )
      response.raise_for_status()
  
      # Validate schema of response
      data = response.json()
      schema_path = os.path.join(
          os.path.dirname(dashboard.__file__),
          "modules/snapshot/component_activities_schema.json",
      )
      pprint.pprint(data)
      jsonschema.validate(
          instance=data, schema=json.load(open(schema_path))
      )
  
      # Validate driver response can be cast to RayActivityResponse object
      # and that there are no active drivers.
      driver_ray_activity_response = RayActivityResponse(
          **data["driver"]
      )
      assert driver_ray_activity_response.is_active == "INACTIVE"
      assert driver_ray_activity_response.reason is None
  
      # Validate external component response can be cast to RayActivityResponse object
      if external_hook[-1] == "5":
          external_activity_response = RayActivityResponse(
              **data["test_component5"]
          )
          assert external_activity_response.is_active == "ACTIVE"
          assert external_activity_response.reason == "Counter: 1"
      elif external_hook[-1] == "4":
          external_activity_response = RayActivityResponse(
              **data["external_component"]
          )
          assert external_activity_response.is_active == "ERROR"
          assert (
              "'Error in external cluster activity hook'"
              in external_activity_response.reason
          )
      elif external_hook[-1] == "3":
          external_activity_response = RayActivityResponse(
              **data["external_component"]
          )
          assert external_activity_response.is_active == "ERROR"
      elif external_hook[-1] == "2":
          external_activity_response = RayActivityResponse(
              **data["test_component2"]
          )
          assert external_activity_response.is_active == "ERROR"
      elif external_hook[-1] == "1":
          external_activity_response = RayActivityResponse(
              **data["test_component1"]
          )
          assert external_activity_response.is_active == "ACTIVE"
          assert external_activity_response.reason == "Counter: 1"
  
          # Call endpoint again to validate different response
          response = requests.get(
              "http://127.0.0.1:8265/api/component_activities"
          )
          response.raise_for_status()
          data = response.json()
          jsonschema.validate(
              instance=data, schema=json.load(open(schema_path))
          )
  
          external_activity_response = RayActivityResponse(
              **data["test_component1"]
          )
          assert external_activity_response.is_active == "ACTIVE"
          assert external_activity_response.reason == "Counter: 2"
  
file_path: ray/python/ray/dashboard/modules/snapshot/tests/test_component_activities.py
imports: import os
  import sys
  import json
  import jsonschema
  import pprint
  import pytest
  import requests
  from ray._private.test_utils import format_web_url, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.dashboard import dashboard
  from ray.dashboard.consts import RAY_CLUSTER_ACTIVITY_HOOK
  from ray.dashboard.modules.snapshot.snapshot_head import RayActivityResponse
  from ray.dashboard.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_component_activities_hook(set_ray_cluster_activity_hook, call_ray_start) without return types