code: def test_logs_list(ray_start_with_dashboard):
      assert (
          wait_until_server_available(
              ray_start_with_dashboard.address_info["webui_url"]
          )
          is True
      )
      webui_url = ray_start_with_dashboard.address_info["webui_url"]
      webui_url = format_web_url(webui_url)
      node_id = list_nodes()[0]["node_id"]
  
      def verify():
          response = requests.get(
              webui_url + f"/api/v0/logs?node_id={node_id}"
          )
          response.raise_for_status()
          result = json.loads(response.text)
          assert result["result"]
          logs = result["data"]["result"]
  
          # Test worker logs
          outs = logs["worker_out"]
          errs = logs["worker_err"]
          core_worker_logs = logs["core_worker"]
  
          assert len(outs) == len(errs) == len(core_worker_logs)
          assert len(outs) > 0
  
          # Test gcs / raylet / dashboard
          for file in ["gcs_server.out", "gcs_server.err"]:
              assert file in logs["gcs_server"]
          for file in ["raylet.out", "raylet.err"]:
              assert file in logs["raylet"]
          for file in ["dashboard.log"]:
              assert file in logs["dashboard"]
          for file in ["dashboard_agent.log"]:
              assert file in logs["agent"]
          return True
  
      wait_for_condition(verify)
  
      def verify_filter():
          # Test that logs/list can be filtered
          response = requests.get(
              webui_url + f"/api/v0/logs?node_id={node_id}&glob=*gcs*"
          )
          response.raise_for_status()
          result = json.loads(response.text)
          assert result["result"]
          logs = result["data"]["result"]
          assert "gcs_server" in logs
          assert "internal" in logs
          assert len(logs) == 2
          assert "gcs_server.out" in logs["gcs_server"]
          assert "gcs_server.err" in logs["gcs_server"]
          assert "debug_state_gcs.txt" in logs["internal"]
          return True
  
      wait_for_condition(verify_filter)
  
      def verify_worker_logs():
          response = requests.get(
              webui_url
              + f"/api/v0/logs?node_id={node_id}&glob=*worker*"
          )
          response.raise_for_status()
          result = json.loads(response.text)
          assert result["result"]
          logs = result["data"]["result"]
          worker_log_categories = [
              "core_worker",
              "worker_out",
              "worker_err",
          ]
          assert all([cat in logs for cat in worker_log_categories])
          num_workers = len(
              list(
                  filter(
                      lambda w: w["worker_type"] == "WORKER",
                      list_workers(),
                  )
              )
          )
          assert (
              len(logs["worker_out"])
              == len(logs["worker_err"])
              == len(logs["worker_out"])
          )
          assert num_workers == len(logs["worker_out"])
          return True
  
      wait_for_condition(verify_worker_logs)
  
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 12 for calling a global function with example usage: test_logs_list(ray_start_with_dashboard) and returns: True, True, True