docstr: Runs a complete cycle of task generation and simulating their completion.
  
  Args:
    use_task_queue: Whether to use task queue.
    expect_nodes: List of nodes whose task generation is expected.
    finish_nodes: List of nodes whose completion should be simulated. If
      `None` (default), all of `expect_nodes` will be finished.
    artifact_custom_properties: A dict of custom properties to attach to the
      output artifacts.
    fail_fast: If `True`, pipeline is aborted immediately if any node fails.
docstr_tok: ["Runs","a","complete","cycle","of","task","generation","and","simulating","their","completion",".","Args",":","use_task_queue",":","Whether","to","use","task","queue",".","expect_nodes",":","List","of","nodes","whose","task","generation","is","expected",".","finish_nodes",":","List","of","nodes","whose","completion","should","be","simulated",".","If","`","None","`","(","default",")",",","all","of","`","expect_nodes","`","will","be","finished",".","artifact_custom_properties",":","A","dict","of","custom","properties","to","attach","to","the","output","artifacts",".","fail_fast",":","If","`","True","`",",","pipeline","is","aborted","immediately","if","any","node","fails","."]
code: def _run_next(
      self,
      use_task_queue,
      expect_nodes,
      finish_nodes=None,
      artifact_custom_properties=None,
      fail_fast=False,
  ):
      """Runs a complete cycle of task generation and simulating their completion.
  
      Args:
        use_task_queue: Whether to use task queue.
        expect_nodes: List of nodes whose task generation is expected.
        finish_nodes: List of nodes whose completion should be simulated. If
          `None` (default), all of `expect_nodes` will be finished.
        artifact_custom_properties: A dict of custom properties to attach to the
          output artifacts.
        fail_fast: If `True`, pipeline is aborted immediately if any node fails.
      """
      tasks = self._generate(use_task_queue, True, fail_fast=fail_fast)
      for task in tasks:
          self.assertIsInstance(task, task_lib.ExecNodeTask)
      expected_node_ids = [n.node_info.id for n in expect_nodes]
      task_node_ids = [task.node_uid.node_id for task in tasks]
      self.assertCountEqual(expected_node_ids, task_node_ids)
      finish_node_ids = set(
          [n.node_info.id for n in finish_nodes]
          if finish_nodes is not None
          else expected_node_ids
      )
      for task in tasks:
          if task.node_uid.node_id in finish_node_ids:
              self._finish_node_execution(
                  use_task_queue,
                  task,
                  artifact_custom_properties=artifact_custom_properties,
              )
  
imports: import itertools
  import os
  from typing import Literal
  import uuid
  from absl.testing import parameterized
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.dsl.compiler import constants
  from tfx.orchestration import data_types_utils
  from tfx.orchestration.experimental.core import constants
  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 sync_pipeline_task_gen
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_sync_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import runtime_parameter_utils
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import status
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 7 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj._run_next(use_task_queue, expect_nodes, finish_nodes, artifact_custom_properties, fail_fast) without return types