docstr: Tests that pipeline is finalized only after terminal nodes are successful.
  
  Args:
    use_task_queue: If task queue is enabled, new tasks are only generated if
      a task with the same task_id does not already exist in the queue.
      `use_task_queue=False` is useful to test the case of task generation
      when task queue is empty (for eg: due to orchestrator restart).
    fail_fast: If `True`, pipeline is aborted immediately if any node fails.
docstr_tok: ["Tests","that","pipeline","is","finalized","only","after","terminal","nodes","are","successful",".","Args",":","use_task_queue",":","If","task","queue","is","enabled",",","new","tasks","are","only","generated","if","a","task","with","the","same","task_id","does","not","already","exist","in","the","queue",".","`","use_task_queue=False","`","is","useful","to","test","the","case","of","task","generation","when","task","queue","is","empty","(","for","eg",":","due","to","orchestrator","restart",")",".","fail_fast",":","If","`","True","`",",","pipeline","is","aborted","immediately","if","any","node","fails","."]
code: def test_pipeline_succeeds_when_terminal_nodes_succeed(
      self, use_task_queue, fail_fast
  ):
      """Tests that pipeline is finalized only after terminal nodes are successful.
  
      Args:
        use_task_queue: If task queue is enabled, new tasks are only generated if
          a task with the same task_id does not already exist in the queue.
          `use_task_queue=False` is useful to test the case of task generation
          when task queue is empty (for eg: due to orchestrator restart).
        fail_fast: If `True`, pipeline is aborted immediately if any node fails.
      """
      # Check the expected terminal nodes.
      layers = sptg._topsorted_layers(self._pipeline)
      self.assertEqual(
          {
              self._example_validator.node_info.id,
              self._chore_b.node_info.id,
              # evaluator execution will be skipped as it is run conditionally and
              # the condition always evaluates to False in the current test.
              self._evaluator.node_info.id,
          },
          sptg._terminal_node_ids(layers, {}),
      )
  
      # Start executing the pipeline:
  
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
  
      self._run_next(use_task_queue, expect_nodes=[self._stats_gen])
      self._run_next(use_task_queue, expect_nodes=[self._schema_gen])
  
      # Both example-validator and transform are ready to execute.
      [example_validator_task, transform_task] = self._generate(
          use_task_queue, True, fail_fast=fail_fast
      )
      self.assertEqual(
          self._example_validator.node_info.id,
          example_validator_task.node_uid.node_id,
      )
      self.assertEqual(
          self._transform.node_info.id, transform_task.node_uid.node_id
      )
      # Start processing (but do not finish) example-validator.
      self._start_processing(use_task_queue, example_validator_task)
      # But finish transform which is in the same layer.
      self._finish_node_execution(use_task_queue, transform_task)
  
      # Readability note: below, example-validator task should continue to be
      # generated when not using task queue because the execution is active.
  
      # Trainer and downstream nodes can execute as transform is finished.
      self._run_next(
          use_task_queue,
          expect_nodes=[self._trainer]
          if use_task_queue
          else [self._example_validator, self._trainer],
          finish_nodes=[self._trainer],
          fail_fast=fail_fast,
      )
      self._run_next(
          use_task_queue,
          expect_nodes=[self._chore_a]
          if use_task_queue
          else [self._example_validator, self._chore_a],
          finish_nodes=[self._chore_a],
          fail_fast=fail_fast,
      )
      self._run_next(
          use_task_queue,
          expect_nodes=[self._chore_b]
          if use_task_queue
          else [self._example_validator, self._chore_b],
          finish_nodes=[self._chore_b],
          fail_fast=fail_fast,
      )
      self._run_next(
          use_task_queue,
          expect_nodes=[]
          if use_task_queue
          else [self._example_validator],
          finish_nodes=[],
          fail_fast=fail_fast,
      )
  
      # FinalizePipelineTask is generated only after example-validator finishes.
      test_utils.fake_execute_node(
          self._mlmd_connection, example_validator_task
      )
      self._finish_processing(use_task_queue, example_validator_task)
      [finalize_task] = self._generate(
          use_task_queue, True, fail_fast=fail_fast
      )
      self.assertIsInstance(
          finalize_task, task_lib.FinalizePipelineTask
      )
      self.assertEqual(status_lib.Code.OK, finalize_task.status.code)
  
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 10 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_pipeline_succeeds_when_terminal_nodes_succeed(use_task_queue, fail_fast) without return types