code: def check_file(type, task_name, expected_log, expect_no_end=False):
      """Check file of type = 'out'/'err'"""
  
      def _read_file(filepath, start, end):
          with open(filepath, "r") as f:
              f.seek(start, 0)
              if end is None:
                  return f.read()
              return f.read(end - start)
  
      tasks = list_tasks(
          filters=[("name", "=", f"{task_name}")], detail=True
      )
      assert len(tasks) == 1
      task = tasks[0]
      assert task["task_log_info"] is not None
      log_info = task["task_log_info"]
  
      file = log_info.get(f"std{type}_file", None)
      start_offset = log_info.get(f"std{type}_start", None)
      end_offset = log_info.get(f"std{type}_end", None)
      if not expect_no_end:
          assert end_offset >= start_offset
      else:
          assert end_offset is None
      assert (
          start_offset > 0
      ), "offsets should be > 0 with magical log prefix"
      actual_log = _read_file(file, start_offset, end_offset)
      assert actual_log == expected_log
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 15 for calling a global function with example usage: check_file(type, task_name, expected_log, expect_no_end) and returns: f, f