code: def testPublishSuccessfulExecutionWithRuntimeResolvedUri(self):
      with metadata.Metadata(
          connection_config=self._connection_config
      ) as m:
          contexts = self._generate_contexts(m)
          execution_id = execution_publish_utils.register_execution(
              m, self._execution_type, contexts
          ).id
          output_key = "examples"
          output_example = standard_artifacts.Examples()
          output_example.uri = outputs_utils.RESOLVED_AT_RUNTIME
          output_example.is_external = True
          execution_lib.register_output_artifacts(
              m, execution_id, {output_key: [output_example]}
          )
          executor_output = execution_result_pb2.ExecutorOutput()
          # The executor output contains two artifacts compared to the original one.
          for output_artifact_uri in [
              "/examples_uri/1",
              "/examples_uri/2",
          ]:
              text_format.Parse(
                  f"""
              uri: '{output_artifact_uri}'
              custom_properties {{
                key: 'is_external'
                value {{int_value: 1}}
              }}
              """,
                  executor_output.output_artifacts[
                      output_key
                  ].artifacts.add(),
              )
  
          (
              output_dict,
              _,
          ) = execution_publish_utils.publish_succeeded_execution(
              m,
              execution_id,
              contexts,
              {output_key: [output_example]},
              executor_output,
          )
          self.assertLen(output_dict[output_key], 2)
          self.assertEqual(
              output_dict[output_key][0].uri, "/examples_uri/1"
          )
          self.assertEqual(
              output_dict[output_key][1].uri, "/examples_uri/2"
          )
          events = m.store.get_events_by_execution_ids([execution_id])
          self.assertLen(events, 3)
          for event, artifact_id, artifact_index, event_type in [
              (events[0], 1, 0, "PENDING_OUTPUT"),
              (events[1], 2, 0, "OUTPUT"),
              (events[2], 3, 1, "OUTPUT"),
          ]:
              self.assertProtoPartiallyEquals(
                  f"""
              artifact_id: {artifact_id}
              execution_id: 1
              path {{
                steps {{
                  key: 'examples'
                }}
                steps {{
                  index: {artifact_index}
                }}
              }}
              type: {event_type}
              """,
                  event,
                  ignored_fields=["milliseconds_since_epoch"],
              )
  
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 6 for calling the ExecutionPublisherTest obj's underlying member method code with example usage: obj.testPublishSuccessfulExecutionWithRuntimeResolvedUri() without return types