code: def test_logs_stream_and_tail(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_basic():
          stream_response = requests.get(
              webui_url
              + f"/api/v0/logs/file?node_id={node_id}&filename=gcs_server.out&lines=5",
              stream=True,
          )
          if stream_response.status_code != 200:
              raise ValueError(stream_response.content.decode("utf-8"))
          lines = []
          for line in stream_response.iter_lines():
              lines.append(line.decode("utf-8"))
          assert len(lines) == 5 or len(lines) == 6
          return True
  
      wait_for_condition(verify_basic)
  
      @ray.remote
      class Actor:
          def write_log(self, strings):
              for s in strings:
                  print(s)
  
          def getpid(self):
              return os.getpid()
  
      test_log_text = "test_log_text_日志_{}"
      actor = Actor.remote()
      ray.get(actor.write_log.remote([test_log_text.format("XXXXXX")]))
  
      # Test stream and fetching by actor id
      stream_response = requests.get(
          webui_url
          + "/api/v0/logs/stream?&lines=-1"
          + f"&actor_id={actor._ray_actor_id.hex()}",
          stream=True,
      )
      if stream_response.status_code != 200:
          raise ValueError(stream_response.content.decode("utf-8"))
      stream_iterator = stream_response.iter_content(chunk_size=None)
      actual_output = next(stream_iterator).decode("utf-8")
      assert "actor_name:Actor\n" in actual_output
      assert test_log_text.format("XXXXXX") in actual_output
  
      streamed_string = ""
      for i in range(5):
          strings = []
          for j in range(3):
              strings.append(test_log_text.format(f"{3*i + j:06d}"))
  
          ray.get(actor.write_log.remote(strings))
  
          string = ""
          for s in strings:
              string += s + "\n"
          streamed_string += string
          # NOTE: Prefix 1 indicates the stream has succeeded.
          assert string in next(stream_iterator).decode("utf-8")
      del stream_response
  
      # Test tailing log by actor id
      LINES = 150
      file_response = requests.get(
          webui_url
          + f"/api/v0/logs/file?&lines={LINES}"
          + "&actor_id="
          + actor._ray_actor_id.hex(),
      ).content.decode("utf-8")
      # NOTE: Prefix 1 indicates the stream has succeeded.
      for line in streamed_string.split("\n")[-(LINES + 1) :]:
          assert line in file_response
  
      # Test query by pid & node_ip instead of actor id.
      node_ip = list(ray.nodes())[0]["NodeManagerAddress"]
      pid = ray.get(actor.getpid.remote())
      file_response = requests.get(
          webui_url
          + f"/api/v0/logs/file?node_ip={node_ip}&lines={LINES}"
          + f"&pid={pid}",
      ).content.decode("utf-8")
      # NOTE: Prefix 1 indicates the stream has succeeded.
      for line in streamed_string.split("\n")[-(LINES + 1) :]:
          assert line in file_response
  
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 13 for calling a global function with example usage: test_logs_stream_and_tail(ray_start_with_dashboard) and returns: True, os