code: def test_log_cli(shutdown_only):
      ray.init(num_cpus=1)
      runner = CliRunner()
  
      # Test the head node is chosen by default.
      def verify():
          result = runner.invoke(logs_state_cli_group, ["cluster"])
          assert result.exit_code == 0, result.exception
          assert "raylet.out" in result.output
          assert "raylet.err" in result.output
          assert "gcs_server.out" in result.output
          assert "gcs_server.err" in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test when there's only 1 match, it prints logs.
      def verify():
          result = runner.invoke(
              logs_state_cli_group, ["cluster", "raylet.out"]
          )
          assert result.exit_code == 0
          assert "raylet.out" not in result.output
          assert "raylet.err" not in result.output
          assert "gcs_server.out" not in result.output
          assert "gcs_server.err" not in result.output
          # Make sure it prints the log message.
          assert "NodeManager server started" in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test when there's more than 1 match, it prints a list of logs.
      def verify():
          result = runner.invoke(
              logs_state_cli_group, ["cluster", "raylet.*"]
          )
          assert result.exit_code == 0, result.exception
          assert "raylet.out" in result.output
          assert "raylet.err" in result.output
          assert "gcs_server.out" not in result.output
          assert "gcs_server.err" not in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test actor log: `ray logs actor`
      ACTOR_LOG_LINE = "test actor log"
  
      @ray.remote
      class Actor:
          def __init__(self):
              print(ACTOR_LOG_LINE)
  
      actor = Actor.remote()
      actor_id = actor._actor_id.hex()
  
      def verify():
          result = runner.invoke(
              logs_state_cli_group, ["actor", "--id", actor_id]
          )
          assert result.exit_code == 0, result.exception
          assert ACTOR_LOG_LINE in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test worker log: `ray logs worker`
      WORKER_LOG_LINE = "test worker log"
  
      @ray.remote
      def worker_func():
          import os
  
          print(WORKER_LOG_LINE)
          return os.getpid()
  
      pid = ray.get(worker_func.remote())
  
      def verify():
          result = runner.invoke(
              logs_state_cli_group, ["worker", "--pid", pid]
          )
          assert result.exit_code == 0, result.exception
          assert WORKER_LOG_LINE in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test `ray logs raylet.*` forwarding to `ray logs cluster raylet.*`
      def verify():
          result = runner.invoke(logs_state_cli_group, ["raylet.*"])
          assert result.exit_code == 0, result.exception
          assert "raylet.out" in result.output
          assert "raylet.err" in result.output
          assert "gcs_server.out" not in result.output
          assert "gcs_server.err" not in result.output
          return True
  
      wait_for_condition(verify)
  
      # Test binary binary files and encodings.
      log_dir = (
          ray._private.worker.global_worker.node.get_logs_dir_path()
      )
      file = "test.bin"
      binary_file = os.path.join(log_dir, file)
      with open(binary_file, "wb") as f:
          data = bytearray(i for i in range(256))
          f.write(data)
  
      def verify():
          # Tailing with lines is not supported for binary files, thus the `tail=-1`
          result = runner.invoke(
              logs_state_cli_group,
              [
                  file,
                  "--encoding",
                  "iso-8859-1",
                  "--encoding-errors",
                  "replace",
                  "--tail",
                  "-1",
              ],
          )
          assert result.exit_code == 0, result.exception
          assert result.output == data.decode(
              encoding="iso-8859-1", errors="replace"
          )
          return True
  
      wait_for_condition(verify)
  
imports: import json
  import os
  import sys
  import asyncio
  from typing import List
  import urllib
  from unittest.mock import MagicMock, AsyncMock
  import pytest
  from ray.util.state.state_cli import logs_state_cli_group
  from ray.util.state import list_jobs
  import requests
  from click.testing import CliRunner
  import grpc
  from pathlib import Path
  import ray
  from ray._private.test_utils import format_web_url, wait_for_condition, wait_until_server_available
  from ray._private.ray_constants import LOG_PREFIX_TASK_ATTEMPT_START, LOG_PREFIX_TASK_ATTEMPT_END
  from ray._raylet import ActorID, NodeID, TaskID, WorkerID
  from ray.core.generated.common_pb2 import Address
  from ray.core.generated.gcs_service_pb2 import GetTaskEventsReply
  from ray.core.generated.reporter_pb2 import ListLogsReply, StreamLogReply
  from ray.core.generated.gcs_pb2 import ActorTableData, TaskEvents, TaskStateUpdate, TaskLogInfo
  from ray.dashboard.modules.actor.actor_head import actor_table_data_to_dict
  from ray.dashboard.modules.log.log_agent import find_offset_of_content_in_file, find_end_offset_file, find_end_offset_next_n_lines_from_offset, find_start_offset_last_n_lines_from_offset, LogAgentV1Grpc
  from ray.dashboard.modules.log.log_agent import _stream_log_in_chunk
  from ray.dashboard.modules.log.log_manager import LogsManager
  from ray.dashboard.tests.conftest import *
  from ray.util.state import get_log, list_logs, list_nodes, list_workers
  from ray.util.state.common import GetLogOptions
  from ray.util.state.exception import DataSourceUnavailable, RayStateApiException
  from ray.util.state.state_manager import StateDataSourceClient
label_desc: Use image node_id 20 for calling a global function with example usage: test_log_cli(shutdown_only) and returns: True, True, True, True, os, True, True, True