code: def run_generator_and_test(
      test_case,
      mlmd_connection_manager,
      generator_class,
      pipeline,
      task_queue,
      use_task_queue,
      service_job_manager,
      num_initial_executions,
      num_tasks_generated,
      num_new_executions,
      num_active_executions,
      expected_exec_nodes=None,
      ignore_update_node_state_tasks=False,
      fail_fast=None,
      expected_context_names=None,
  ):
      """Runs generator.generate() and tests the effects."""
      if service_job_manager is None:
          service_job_manager = service_jobs.DummyServiceJobManager()
      with mlmd_connection_manager:
          executions = get_non_orchestrator_executions(
              mlmd_connection_manager.primary_mlmd_handle
          )
          test_case.assertLen(
              executions,
              num_initial_executions,
              f"Expected {num_initial_executions} execution(s) in MLMD.",
          )
      tasks = run_generator(
          mlmd_connection_manager,
          generator_class,
          pipeline,
          task_queue,
          use_task_queue,
          service_job_manager,
          ignore_update_node_state_tasks=ignore_update_node_state_tasks,
          fail_fast=fail_fast,
      )
      with mlmd_connection_manager:
          test_case.assertLen(
              tasks,
              num_tasks_generated,
              f"Expected {num_tasks_generated} task(s) to be generated.",
          )
          executions = get_non_orchestrator_executions(
              mlmd_connection_manager.primary_mlmd_handle
          )
          num_total_executions = (
              num_initial_executions + num_new_executions
          )
          test_case.assertLen(
              executions,
              num_total_executions,
              f"Expected {num_total_executions} execution(s) in MLMD.",
          )
          active_executions = [
              e
              for e in executions
              if execution_lib.is_execution_active(e)
          ]
          test_case.assertLen(
              active_executions,
              num_active_executions,
              f"Expected {num_active_executions} active execution(s) in MLMD.",
          )
          if expected_exec_nodes:
              for i, task in enumerate(
                  t
                  for t in tasks
                  if isinstance(t, task_lib.ExecNodeTask)
              ):
                  _verify_exec_node_task(
                      test_case,
                      pipeline,
                      expected_exec_nodes[i],
                      active_executions[i].id,
                      task,
                      expected_context_names,
                  )
          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 18 for calling a global function with example usage: run_generator_and_test(test_case, mlmd_connection_manager, generator_class, pipeline, task_queue, use_task_queue, service_job_manager, num_initial_executions, num_tasks_generated, num_new_executions, num_active_executions, expected_exec_nodes, ignore_update_node_state_tasks, fail_fast, expected_context_names) and returns: tasks