code: def test_post_execution_exceptions_are_surfaced(
      self, mock_fail_exec, mock_publish, mock_record_state_change_time
  ):
      def _publish(**kwargs):
          task = kwargs["task"]
          assert isinstance(task, task_lib.ExecNodeTask)
          if task.node_uid.node_id == "Transform":
              raise ValueError("test error 1")
          return mock.DEFAULT
  
      def _fail_execution(*args, **kwargs):
          raise ValueError("test error 2")
  
      mock_publish.side_effect = _publish
      mock_fail_exec.side_effect = _fail_execution
  
      collector = _Collector()
  
      # Register a fake task scheduler.
      ts.TaskSchedulerRegistry.register(
          self._type_url,
          functools.partial(
              _FakeTaskScheduler, block_nodes={}, collector=collector
          ),
      )
  
      task_queue = tq.TaskQueue()
  
      with self._task_manager(task_queue) as task_manager:
          transform_task = _test_exec_node_task(
              "Transform", "test-pipeline", pipeline=self._pipeline
          )
          trainer_task = _test_exec_node_task(
              "Trainer", "test-pipeline", pipeline=self._pipeline
          )
          task_queue.enqueue(transform_task)
          task_queue.enqueue(trainer_task)
  
      self.assertTrue(task_manager.done())
      exception = task_manager.exception()
      self.assertIsNotNone(exception)
      self.assertIsInstance(exception, tm.TasksProcessingError)
      self.assertLen(exception.errors, 1)
      self.assertEqual("test error 2", str(exception.errors[0]))
  
      self.assertCountEqual(
          [transform_task, trainer_task], collector.scheduled_tasks
      )
      result_ok = ts.TaskSchedulerResult(
          status=status_lib.Status(
              code=status_lib.Code.OK,
              message="_FakeTaskScheduler result",
          )
      )
      mock_publish.assert_has_calls(
          [
              mock.call(
                  mlmd_handle=mock.ANY,
                  task=transform_task,
                  result=result_ok,
              ),
              mock.call(
                  mlmd_handle=mock.ANY,
                  task=trainer_task,
                  result=result_ok,
              ),
          ],
          any_order=True,
      )
      mock_fail_exec.assert_called_once()
      self.assertLen(mock_publish.mock_calls, 2)
      self.assertLen(mock_record_state_change_time.mock_calls, 1)
  
imports: import contextlib
  import functools
  import os
  import threading
  import time
  from absl import logging
  from absl.testing.absltest import mock
  import tensorflow
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import metadata
  from tfx.orchestration.experimental.core import async_pipeline_task_gen
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import post_execution_utils
  from tfx.orchestration.experimental.core import service_jobs
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_manager
  from tfx.orchestration.experimental.core import task_queue
  from tfx.orchestration.experimental.core import task_scheduler
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.proto.orchestration import execution_result_pb2
  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 4 for calling the TaskManagerTest obj's underlying member method code with example usage: obj.test_post_execution_exceptions_are_surfaced(mock_fail_exec, mock_publish, mock_record_state_change_time) and returns: mock