code: def test_task_logs(
      task_type: str, task_config: Dict[str, Any], log_regex: Any
  ) -> None:
      master_url = conf.make_master_url()
      certs.cli_cert = certs.default_load(conf.make_master_url())
      authentication.cli_auth = authentication.Authentication(
          conf.make_master_url()
      )
      session = api.Session(
          master_url,
          "determined",
          authentication.cli_auth,
          certs.cli_cert,
      )
  
      rps = bindings.get_GetResourcePools(session)
      assert (
          rps.resourcePools and len(rps.resourcePools) > 0
      ), "missing resource pool"
  
      if task_type == "tensorboard":
          exp_id = exp.run_basic_test(
              conf.fixtures_path("no_op/single.yaml"),
              conf.fixtures_path("no_op"),
              1,
          )
          treq = bindings.v1LaunchTensorboardRequest(
              config=task_config, experimentIds=[exp_id]
          )
          task_id = bindings.post_LaunchTensorboard(
              session, body=treq
          ).tensorboard.id
      elif task_type == "notebook":
          nreq = bindings.v1LaunchNotebookRequest(config=task_config)
          task_id = bindings.post_LaunchNotebook(
              session, body=nreq
          ).notebook.id
      elif task_type == "command":
          creq = bindings.v1LaunchCommandRequest(config=task_config)
          task_id = bindings.post_LaunchCommand(
              session, body=creq
          ).command.id
      elif task_type == "shell":
          sreq = bindings.v1LaunchShellRequest(config=task_config)
          task_id = bindings.post_LaunchShell(
              session, body=sreq
          ).shell.id
      else:
          raise ValueError("unknown task type: {task_type}")
  
      def task_logs(**kwargs: Any) -> Iterable[Log]:
          return api.task_logs(session, task_id, **kwargs)
  
      def task_log_fields(
          follow: Optional[bool] = None,
      ) -> Iterable[LogFields]:
          return bindings.get_TaskLogsFields(
              session, taskId=task_id, follow=follow
          )
  
      try:
          result: Optional[Exception] = None
  
          def do_check_logs() -> None:
              nonlocal result
              try:
                  check_logs(
                      log_regex,
                      functools.partial(
                          api.task_logs, session, task_id
                      ),
                      functools.partial(
                          bindings.get_TaskLogsFields,
                          session,
                          taskId=task_id,
                      ),
                  )
              except Exception as e:
                  result = e
  
          thread = threading.Thread(target=do_check_logs, daemon=True)
          thread.start()
          thread.join(timeout=5 * 60)
          if thread.is_alive():
              # The thread did not exit
              raise ValueError("do_check_logs timed out")
          elif isinstance(result, Exception):
              # There was a failure on the thread.
              raise result
      except Exception:
          print(
              "============= test_task_logs_failed, logs from task ============="
          )
          for log in task_logs():
              print(log.log, end="", file=sys.stderr)
          print("============= end of task logs =============")
          raise
  
      finally:
          if task_type == "tensorboard":
              bindings.post_KillTensorboard(
                  session, tensorboardId=task_id
              )
          elif task_type == "notebook":
              bindings.post_KillNotebook(session, notebookId=task_id)
          elif task_type == "command":
              bindings.post_KillCommand(session, commandId=task_id)
          elif task_type == "shell":
              bindings.post_KillShell(session, shellId=task_id)
  
file_path: determined/e2e_tests/tests/cluster/test_logging.py
imports: import functools
  import re
  import sys
  import threading
  from typing import Any, Callable, Dict, Iterable, Optional, Union
  import pytest
  from determined.common import api
  from determined.common.api import authentication, bindings, certs
  from tests import config
  from tests import experiment
label_desc: Use image node_id 2 for calling a global function with example usage: test_task_logs(task_type, task_config, log_regex) and returns: api, bindings