code: def test_fetch_log(fetch_log_setup):
      def f():
          print("test")
  
      r = mr.spawn(f)
      r.execute()
  
      log = r.fetch_log()
      assert str(log).strip() == "test"
  
      # test multiple functions
      def f1(size):
          print("f1" * size)
          sys.stdout.flush()
  
      fs = mr.ExecutableTuple([mr.spawn(f1, 30), mr.spawn(f1, 40)])
      execute(*fs)
      log = fetch_log(*fs, offsets=20, sizes=10)
      assert str(log[0]).strip() == ("f1" * 30)[20:30]
      assert str(log[1]).strip() == ("f1" * 40)[20:30]
      assert len(log[0].offsets) > 0
      assert all(s > 0 for s in log[0].offsets)
      assert len(log[1].offsets) > 0
      assert all(s > 0 for s in log[1].offsets)
      assert len(log[0].chunk_op_keys) > 0
  
      # test negative offsets
      log = fs.fetch_log(offsets=-20, sizes=10)
      assert str(log[0]).strip() == ("f1" * 30 + os.linesep)[-20:-10]
      assert str(log[1]).strip() == ("f1" * 40 + os.linesep)[-20:-10]
      assert all(s > 0 for s in log[0].offsets) is True
      assert len(log[1].offsets) > 0
      assert all(s > 0 for s in log[1].offsets) is True
      assert len(log[0].chunk_op_keys) > 0
  
      # test negative offsets which represented in string
      log = fetch_log(*fs, offsets="-0.02K", sizes="0.01K")
      assert str(log[0]).strip() == ("f1" * 30 + os.linesep)[-20:-10]
      assert str(log[1]).strip() == ("f1" * 40 + os.linesep)[-20:-10]
      assert all(s > 0 for s in log[0].offsets) is True
      assert len(log[1].offsets) > 0
      assert all(s > 0 for s in log[1].offsets) is True
      assert len(log[0].chunk_op_keys) > 0
  
      def test_nested():
          print("level0")
          fr = mr.spawn(f1, 1)
          fr.execute()
          print(fr.fetch_log())
  
      r = mr.spawn(test_nested)
      r.execute()
      log = str(r.fetch_log())
      assert "level0" in log
      assert "f1" in log
  
      df = md.DataFrame(mt.random.rand(10, 3), chunk_size=5)
  
      def df_func(c):
          print("df func")
          return c
  
      df2 = df.map_chunk(df_func)
      df2.execute()
      log = df2.fetch_log()
      assert "Chunk op key:" in str(log)
      assert "df func" in repr(log)
      assert len(str(df.fetch_log())) == 0
  
      def test_host(rndf):
          rm = mr.spawn(nested, rndf)
          rm.execute()
          print(rm.fetch_log())
  
      def nested(_rndf):
          print("log_content")
  
      ds = [
          mr.spawn(test_host, n, retry_when_fail=False)
          for n in np.random.rand(4)
      ]
      xtp = execute(ds)
      for log in fetch_log(xtp):
          assert str(log).strip() == "log_content"
  
      def test_threaded():
          import threading
  
          exc_info = None
  
          def print_fun():
              nonlocal exc_info
              try:
                  print("inner")
              except:  # noqa: E722  # nosec  # pylint: disable=bare-except
                  exc_info = sys.exc_info()
  
          print_thread = threading.Thread(target=print_fun)
          print_thread.start()
          print_thread.join()
  
          if exc_info is not None:
              raise exc_info[1].with_traceback(exc_info[-1])
  
          print("after")
  
      rm = mr.spawn(test_threaded)
      rm.execute()
      logs = str(rm.fetch_log()).strip()
      assert logs == "inner\nafter"
  
imports: import io
  import os
  import re
  import sys
  import tempfile
  from collections import namedtuple
  import numpy
  import pandas
  import pytest
  from ..None import tensor
  from ..None import dataframe
  from ..None import remote
  from ..config import option_context
  from ..deploy.utils import load_service_config_file
  from ..session import execute, fetch, fetch_log
label_desc: Use image node_id 13 for calling a global function with example usage: test_fetch_log(fetch_log_setup) and returns: c