code: def run_generator(
      mlmd_connection_manager: mlmd_cm.MLMDConnectionManager,
      generator_class,
      pipeline,
      task_queue,
      use_task_queue,
      service_job_manager,
      ignore_update_node_state_tasks=False,
      fail_fast=None,
  ):
      """Generates tasks for testing."""
      with mlmd_connection_manager:
          pipeline_state = get_or_create_pipeline_state(
              mlmd_connection_manager.primary_mlmd_handle, pipeline
          )
          generator_params = dict(
              mlmd_connection_manager=mlmd_connection_manager,
              is_task_id_tracked_fn=task_queue.contains_task_id,
              service_job_manager=service_job_manager,
          )
          if fail_fast is not None:
              generator_params["fail_fast"] = fail_fast
          task_gen = generator_class(**generator_params)
          with mock.patch.object(
              outputs_utils,
              "get_stateful_working_dir_index",
              autospec=True,
          ) as mocked_get_stateful_working_dir_index:
              mocked_get_stateful_working_dir_index.return_value = (
                  _MOCKED_STATEFUL_WORKING_DIR_INDEX
              )
              tasks = task_gen.generate(pipeline_state)
          if use_task_queue:
              for task in tasks:
                  if isinstance(task, task_lib.ExecNodeTask):
                      task_queue.enqueue(task)
          for task in tasks:
              if isinstance(task, task_lib.UpdateNodeStateTask):
                  with pipeline_state:
                      with pipeline_state.node_state_update_context(
                          task.node_uid
                      ) as node_state:
                          node_state.update(
                              task.state,
                              task.status,
                              task.backfill_token,
                          )
      if ignore_update_node_state_tasks:
          tasks = [
              t
              for t in tasks
              if not isinstance(t, task_lib.UpdateNodeStateTask)
          ]
      return tasks
  
imports: import os
  from typing import Dict, Optional
  import uuid
  from absl.testing.absltest import mock
  from tfx import types
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import env
  from tfx.orchestration.experimental.core import mlmd_state
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import cache_utils
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from tfx.utils import typing_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 15 for calling a global function with example usage: run_generator(mlmd_connection_manager, generator_class, pipeline, task_queue, use_task_queue, service_job_manager, ignore_update_node_state_tasks, fail_fast) and returns: tasks