code: def test_log_rotation(shutdown_only, monkeypatch):
      max_bytes = 1
      backup_count = 3
      set_logging_config(monkeypatch, max_bytes, backup_count)
      ray.init(num_cpus=1)
      session_dir = ray._private.worker.global_worker.node.address_info[
          "session_dir"
      ]
      session_path = Path(session_dir)
      log_dir_path = session_path / "logs"
  
      log_rotating_component = [
          ray_constants.PROCESS_TYPE_DASHBOARD,
          ray_constants.PROCESS_TYPE_DASHBOARD_AGENT,
          ray_constants.PROCESS_TYPE_LOG_MONITOR,
          ray_constants.PROCESS_TYPE_MONITOR,
          ray_constants.PROCESS_TYPE_PYTHON_CORE_WORKER_DRIVER,
          ray_constants.PROCESS_TYPE_PYTHON_CORE_WORKER,
          # Below components are not log rotating now.
          # ray_constants.PROCESS_TYPE_RAYLET,
          # ray_constants.PROCESS_TYPE_GCS_SERVER,
          # ray_constants.PROCESS_TYPE_WORKER,
      ]
  
      # Run the basic workload.
      @ray.remote
      def f():
          for i in range(10):
              print(f"test {i}")
  
      # Create a runtime env to make sure dashboard agent is alive.
      ray.get(
          f.options(
              runtime_env={"env_vars": {"A": "a", "B": "b"}}
          ).remote()
      )
  
      # Filter out only paths that end in .log, .log.1, etc.
      # These paths are handled by the logger; the others (.out, .err) are not.
      paths = []
      for path in log_dir_path.iterdir():
          if re.search(r".*\.log(\.\d+)?", str(path)):
              paths.append(path)
  
      def component_exist(component, paths):
          for path in paths:
              filename = path.stem
              if component in filename:
                  return True
          return False
  
      def component_file_only_one_log_entry(component):
          """Since max_bytes is 1, the log file should
          only have at most one log entry.
          """
          for path in paths:
              if not component_exist(component, [path]):
                  continue
  
              with open(path) as file:
                  found = False
                  for line in file:
                      if re.match(
                          r"^\[?\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d", line
                      ):
                          if found:
                              return False
                          found = True
          return True
  
      for component in log_rotating_component:
          assert component_exist(component, paths), paths
          assert component_file_only_one_log_entry(component)
  
      # Check if the backup count is respected.
      file_cnts = defaultdict(int)
      for path in paths:
          filename = path.name
          parts = filename.split(".")
          if len(parts) == 3:
              filename_without_suffix = parts[0]
              file_cnts[filename_without_suffix] += 1
      for filename, file_cnt in file_cnts.items():
          assert file_cnt <= backup_count, (
              f"{filename} has files that are more than "
              f"backup count {backup_count}, file count: {file_cnt}"
          )
  
imports: import io
  import os
  import re
  import subprocess
  import sys
  import tempfile
  import time
  import logging
  from collections import Counter, defaultdict
  from contextlib import redirect_stderr, redirect_stdout
  from pathlib import Path
  from unittest.mock import Mock, MagicMock
  import pytest
  import ray
  from ray._private import ray_constants
  from ray._private.log_monitor import LOG_NAME_UPDATE_INTERVAL_S, RAY_LOG_MONITOR_MANY_FILES_THRESHOLD, LogFileInfo, LogMonitor, is_proc_alive
  from ray._private.test_utils import get_log_batch, get_log_message, get_log_data, init_log_pubsub, run_string_as_driver, wait_for_condition
  from ray.cross_language import java_actor_class
  from ray.autoscaler._private.cli_logger import cli_logger
label_desc: Use image node_id 6 for calling a global function with example usage: test_log_rotation(shutdown_only, monkeypatch) and returns: False, True, True, False