docstr: Tests node triggering strategies.
docstr_tok: ["Tests","node","triggering","strategies","."]
code: def test_node_triggering_strategies(
      self, node_id, node_execution_options
  ):
      """Tests node triggering strategies."""
      if node_id == "chore_a":
          # Set chore_a's node_success_optional bit to True.
          self._chore_a.execution_options.CopyFrom(
              node_execution_options
          )
      elif node_id == "chore_b":
          # Set chore_b's node triggering strategy to all upstream node succeeded.
          self._chore_b.execution_options.CopyFrom(
              node_execution_options
          )
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
      self._run_next(False, expect_nodes=[self._stats_gen])
      self._run_next(False, expect_nodes=[self._schema_gen])
      self._run_next(
          False, expect_nodes=[self._example_validator, self._transform]
      )
      self._run_next(False, expect_nodes=[self._trainer])
      [chore_a_task] = self._generate_and_test(
          False,
          num_initial_executions=6,
          num_tasks_generated=1,
          num_new_executions=1,
          num_active_executions=1,
          ignore_update_node_state_tasks=True,
          fail_fast=False,
      )
      with self._mlmd_connection as m:
          with mlmd_state.mlmd_execution_atomic_op(
              m, chore_a_task.execution_id
          ) as chore_a_exec:
              # Fail chore a execution.
              chore_a_exec.last_known_state = (
                  metadata_store_pb2.Execution.FAILED
              )
              data_types_utils.set_metadata_value(
                  chore_a_exec.custom_properties[
                      constants.EXECUTION_ERROR_MSG_KEY
                  ],
                  "foobar error",
              )
              data_types_utils.set_metadata_value(
                  chore_a_exec.custom_properties[
                      constants.EXECUTION_ERROR_CODE_KEY
                  ],
                  status_lib.Code.RESOURCE_EXHAUSTED,
              )
  
      # Despite upstream node failure, chore b proceeds because:
      # 1) It's failure strategy is ALL_UPSTREAM_NODES_COMPLETED, or
      # 2) chore a's `success_optional` bit is set to True.
      self._run_next(False, expect_nodes=[self._chore_b])
      # All runnable nodes executed, finalization task should be produced.
      [finalize_task] = self._generate(False, True)
      self.assertIsInstance(
          finalize_task, task_lib.FinalizePipelineTask
      )
  
      # Pipeline should only be ok if the failed node is optional.
      if node_execution_options.node_success_optional:
          self.assertEqual(
              status_lib.Code.OK, finalize_task.status.code
          )
      else:
          self.assertEqual(
              status_lib.Code.RESOURCE_EXHAUSTED,
              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 23 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_node_triggering_strategies(node_id, node_execution_options) without return types