code: def test_trigger_strategy_lifetime_end_with_start_node_not_upstream_of_failure(
      self,
  ):
      self._setup_for_chore_pipeline()
  
      self.chore_c.execution_options.strategy = (
          pipeline_pb2.NodeExecutionOptions.LIFETIME_END_WHEN_SUBGRAPH_CANNOT_PROGRESS
      )
      self.chore_c.execution_options.resource_lifetime.lifetime_start = (
          "my_example_gen_1"
      )
  
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self.eg_1, 1, 1
      )
      test_utils.fake_example_gen_run(
          self._mlmd_connection, self.eg_2, 1, 1
      )
  
      [_, chore_d_task, _] = self._generate_and_test(
          False,
          num_initial_executions=2,
          num_tasks_generated=3,
          num_new_executions=3,
          num_active_executions=3,
          ignore_update_node_state_tasks=True,
      )
      self.assertEqual(
          task_lib.NodeUid.from_node(self._pipeline, self.chore_d),
          chore_d_task.node_uid,
      )
  
      # Fail chore_d execution
      with self._mlmd_connection as m:
          with mlmd_state.mlmd_execution_atomic_op(
              m, chore_d_task.execution_id
          ) as chore_d_exec:
              chore_d_exec.last_known_state = (
                  metadata_store_pb2.Execution.FAILED
              )
              data_types_utils.set_metadata_value(
                  chore_d_exec.custom_properties[
                      constants.EXECUTION_ERROR_CODE_KEY
                  ],
                  status_lib.Code.UNAVAILABLE,
              )
              data_types_utils.set_metadata_value(
                  chore_d_exec.custom_properties[
                      constants.EXECUTION_ERROR_MSG_KEY
                  ],
                  "foobar error",
              )
  
      self._run_next(False, expect_nodes=[self.chore_a, self.chore_e])
      self._run_next(False, expect_nodes=[self.chore_b])
  
      # chore_c should run as all of its subgraph ancestors succeeded, failed,
      # or became unrunnable.
      self._run_next(False, expect_nodes=[self.chore_c])
  
      # Pipeline should fail due to chore_d having failed.
      [finalize_task] = self._generate(False, True)
      self.assertEqual(
          status_lib.Code.UNAVAILABLE, finalize_task.status.code
      )
      self.assertEqual("foobar error", finalize_task.status.message)
  
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 33 for calling the SyncPipelineTaskGeneratorTest obj's underlying member method code with example usage: obj.test_trigger_strategy_lifetime_end_with_start_node_not_upstream_of_failure() without return types