code: def testPublishCachedExecution(self):
      with metadata.Metadata(
          connection_config=self._connection_config
      ) as m:
          contexts = self._generate_contexts(m)
          execution = execution_publish_utils.register_execution(
              m, self._execution_type, contexts
          )
          output_example = standard_artifacts.Examples()
          execution_publish_utils.publish_cached_executions(
              m,
              contexts,
              [execution],
              output_artifacts_maps=[{"examples": [output_example]}],
          )
          self.assertEqual(
              output_example.get_string_custom_property(
                  artifact_utils.ARTIFACT_TFX_VERSION_CUSTOM_PROPERTY_KEY
              ),
              version.__version__,
          )
          [execution] = m.store.get_executions()
          self.assertProtoPartiallyEquals(
              """
            id: 1
            last_known_state: CACHED
            """,
              execution,
              ignored_fields=[
                  "type_id",
                  "type",
                  "create_time_since_epoch",
                  "last_update_time_since_epoch",
                  "name",
              ],
          )
          [event] = m.store.get_events_by_execution_ids([execution.id])
          self.assertProtoPartiallyEquals(
              """
            artifact_id: 1
            execution_id: 1
            path {
              steps {
                key: 'examples'
              }
              steps {
                index: 0
              }
            }
            type: OUTPUT
            """,
              event,
              ignored_fields=["milliseconds_since_epoch"],
          )
          # Verifies the context-execution edges are set up.
          self.assertCountEqual(
              [c.id for c in contexts],
              [
                  c.id
                  for c in m.store.get_contexts_by_execution(
                      execution.id
                  )
              ],
          )
          self.assertCountEqual(
              [c.id for c in contexts],
              [
                  c.id
                  for c in m.store.get_contexts_by_artifact(
                      output_example.id
                  )
              ],
          )
  
imports: import copy
  from absl.testing import parameterized
  import tensorflow
  from tfx import version
  from tfx.orchestration import metadata
  from tfx.orchestration.portable import execution_publish_utils
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import context_lib
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.types import artifact
  from tfx.types import artifact_utils
  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 4 for calling the ExecutionPublisherTest obj's underlying member method code with example usage: obj.testPublishCachedExecution() without return types