code: def testGetCachedOutputArtifacts(self, mock_verify_artifacts):
      # Output artifacts that will be used by the first execution with the same
      # cache key.
      output_model_one = standard_artifacts.Model()
      output_model_one.uri = "model_one"
      output_model_two = standard_artifacts.Model()
      output_model_two.uri = "model_two"
      output_example_one = standard_artifacts.Examples()
      output_example_one.uri = "example_one"
      # Output artifacts that will be used by the second execution with the same
      # cache key.
      output_model_three = standard_artifacts.Model()
      output_model_three.uri = "model_three"
      output_model_four = standard_artifacts.Model()
      output_model_four.uri = "model_four"
      output_example_two = standard_artifacts.Examples()
      output_example_two.uri = "example_two"
      output_models_key = "output_models"
      output_examples_key = "output_examples"
      with metadata.Metadata(
          connection_config=self._connection_config
      ) as m:
          cache_context = context_lib.register_context_if_not_exists(
              m, context_lib.CONTEXT_TYPE_EXECUTION_CACHE, "cache_key"
          )
  
          # No succeed execution is associate with this context yet, so the cached
          # output is None
          cached_output = cache_utils.get_cached_outputs(
              m, cache_context
          )
          self.assertIsNone(cached_output)
  
          execution_one = execution_publish_utils.register_execution(
              m,
              metadata_store_pb2.ExecutionType(name="my_type"),
              [cache_context],
          )
          execution_publish_utils.publish_succeeded_execution(
              m,
              execution_one.id,
              [cache_context],
              output_artifacts={
                  output_models_key: [
                      output_model_one,
                      output_model_two,
                  ],
                  output_examples_key: [output_example_one],
              },
          )
          execution_two = execution_publish_utils.register_execution(
              m,
              metadata_store_pb2.ExecutionType(name="my_type"),
              [cache_context],
          )
          (
              output_artifacts,
              _,
          ) = execution_publish_utils.publish_succeeded_execution(
              m,
              execution_two.id,
              [cache_context],
              output_artifacts={
                  output_models_key: [
                      output_model_three,
                      output_model_four,
                  ],
                  output_examples_key: [output_example_two],
              },
          )
  
          # The cached output got should be the artifacts produced by the most
          # recent execution under the given cache context.
          cached_output = cache_utils.get_cached_outputs(
              m, cache_context
          )
          self.assertLen(cached_output, 2)
          self.assertLen(cached_output[output_models_key], 2)
          self.assertLen(cached_output[output_examples_key], 1)
          self.assertProtoPartiallyEquals(
              cached_output[output_models_key][0].mlmd_artifact,
              output_artifacts[output_models_key][0].mlmd_artifact,
              ignored_fields=[
                  "type",
                  "create_time_since_epoch",
                  "last_update_time_since_epoch",
              ],
          )
          self.assertProtoPartiallyEquals(
              cached_output[output_models_key][1].mlmd_artifact,
              output_artifacts[output_models_key][1].mlmd_artifact,
              ignored_fields=[
                  "type",
                  "create_time_since_epoch",
                  "last_update_time_since_epoch",
              ],
          )
          self.assertProtoPartiallyEquals(
              cached_output[output_examples_key][0].mlmd_artifact,
              output_artifacts[output_examples_key][0].mlmd_artifact,
              ignored_fields=[
                  "type",
                  "create_time_since_epoch",
                  "last_update_time_since_epoch",
              ],
          )
  
          # There should again be no cached outputs if the artifacts cannot be
          # verified as still existing
          mock_verify_artifacts.side_effect = RuntimeError()
          cached_output = cache_utils.get_cached_outputs(
              m, cache_context
          )
          self.assertIsNone(cached_output)
  
imports: import os
  from unittest import mock
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import cache_utils
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.proto.orchestration import executable_spec_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import standard_artifacts
  from tfx.utils import test_case_utils
  from google.protobuf import text_format
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 13 for calling the CacheUtilsTest obj's underlying member method code with example usage: obj.testGetCachedOutputArtifacts(mock_verify_artifacts) without return types