docstr: Tests pipeline failure strategies.
docstr_tok: ["Tests","pipeline","failure","strategies","."]
code: def test_pipeline_failure_strategies(self, fail_fast):
      """Tests pipeline failure strategies."""
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
  
      self._run_next(
          False, expect_nodes=[self._stats_gen], fail_fast=fail_fast
      )
      self._run_next(
          False, expect_nodes=[self._schema_gen], fail_fast=fail_fast
      )
  
      # Both example-validator and transform are ready to execute.
      [example_validator_task, transform_task] = self._generate(
          False, 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
      )
  
      # Simulate Transform success.
      self._finish_node_execution(False, transform_task)
  
      # But fail example-validator.
      with self._mlmd_connection as m:
          with mlmd_state.mlmd_execution_atomic_op(
              m, example_validator_task.execution_id
          ) as ev_exec:
              # Fail stats-gen execution.
              ev_exec.last_known_state = (
                  metadata_store_pb2.Execution.FAILED
              )
              data_types_utils.set_metadata_value(
                  ev_exec.custom_properties[
                      constants.EXECUTION_ERROR_CODE_KEY
                  ],
                  status_lib.Code.PERMISSION_DENIED,
              )
              data_types_utils.set_metadata_value(
                  ev_exec.custom_properties[
                      constants.EXECUTION_ERROR_MSG_KEY
                  ],
                  "example-validator error",
              )
  
      if fail_fast:
          # Pipeline run should immediately fail because example-validator failed.
          [finalize_task] = self._generate(
              False, True, fail_fast=fail_fast
          )
          self.assertIsInstance(
              finalize_task, task_lib.FinalizePipelineTask
          )
          self.assertEqual(
              status_lib.Code.PERMISSION_DENIED,
              finalize_task.status.code,
          )
      else:
          # Trainer and downstream nodes can execute as transform has finished.
          # example-validator failure does not impact them as it is not upstream.
          # Pipeline run will still fail but when no more progress can be made.
          self._run_next(
              False, expect_nodes=[self._trainer], fail_fast=fail_fast
          )
          self._run_next(
              False, expect_nodes=[self._chore_a], fail_fast=fail_fast
          )
          self._run_next(
              False, expect_nodes=[self._chore_b], fail_fast=fail_fast
          )
          [finalize_task] = self._generate(
              False, True, fail_fast=fail_fast
          )
          self.assertIsInstance(
              finalize_task, task_lib.FinalizePipelineTask
          )
          self.assertEqual(
              status_lib.Code.PERMISSION_DENIED,
              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 22 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_pipeline_failure_strategies(fail_fast) without return types