code: def test_interpret_status_from_failed_execution(self):
      execution = metadata_store_pb2.Execution(
          last_known_state=metadata_store_pb2.Execution.COMPLETE
      )
      with self.assertRaisesRegex(
          ValueError, "Must be called.*last_known_state = FAILED."
      ):
          task_gen_utils.interpret_status_from_failed_execution(
              execution
          )
  
      execution = metadata_store_pb2.Execution(
          last_known_state=metadata_store_pb2.Execution.FAILED
      )
      self.assertEqual(
          status_lib.Status(code=status_lib.Code.UNKNOWN),
          task_gen_utils.interpret_status_from_failed_execution(
              execution
          ),
      )
  
      # Status is created using special custom properties if they exist.
      execution.custom_properties[
          constants.EXECUTION_ERROR_MSG_KEY
      ].string_value = "permission denied"
      self.assertEqual(
          status_lib.Status(
              code=status_lib.Code.UNKNOWN, message="permission denied"
          ),
          task_gen_utils.interpret_status_from_failed_execution(
              execution
          ),
      )
      execution.custom_properties[
          constants.EXECUTION_ERROR_CODE_KEY
      ].int_value = status_lib.Code.PERMISSION_DENIED
      self.assertEqual(
          status_lib.Status(
              code=status_lib.Code.PERMISSION_DENIED,
              message="permission denied",
          ),
          task_gen_utils.interpret_status_from_failed_execution(
              execution
          ),
      )
  
      # ExecutionResult, if available, has the higher precedence in determining
      # Status as that indicates the most proximate cause.
      execution_result = execution_result_pb2.ExecutionResult(
          code=status_lib.Code.DEADLINE_EXCEEDED,
          result_message="deadline exceeded",
      )
      execution_lib.set_execution_result(execution_result, execution)
      self.assertEqual(
          status_lib.Status(
              code=status_lib.Code.DEADLINE_EXCEEDED,
              message="deadline exceeded",
          ),
          task_gen_utils.interpret_status_from_failed_execution(
              execution
          ),
      )
  
imports: import os
  import time
  from unittest import mock
  import uuid
  from absl.testing import parameterized
  import tensorflow
  from tfx import types
  from tfx import version
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.utils import status
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 19 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_interpret_status_from_failed_execution() without return types