code: def testLatestPipelineRunOutputsOutputs_OneKey(self):
      with contextlib.nullcontext():
          node_context = self.put_context("node", "example-gen")
          end_node_context = self.put_context(
              "node", "pipeline-name.pipeline-name_end"
          )
  
          # Creates the first pipeline run and artifact.
          input_artifact_1 = self.put_artifact("Example")
          garbage_collected_artifact = self.put_artifact(
              "Example", state=metadata_store_pb2.Artifact.State.DELETED
          )
          pipeline_run_ctx_1 = self.put_context(
              "pipeline_run", "run-001"
          )
          self.put_execution(
              "ExampleGen",
              inputs={},
              # Although here the ExampleGen outputs a DELETED artifact, in reality
              # the artifact would be marked as LIVE first and then later marked
              # as DELETED when it is garbage collected.
              outputs={
                  "examples": [
                      input_artifact_1,
                      garbage_collected_artifact,
                  ]
              },
              contexts=[node_context, pipeline_run_ctx_1],
              output_event_type=metadata_store_pb2.Event.OUTPUT,
          )
          self.put_execution(
              "pipeline-name_end",
              inputs={},
              outputs={
                  "examples": [
                      input_artifact_1,
                      garbage_collected_artifact,
                  ]
              },
              contexts=[end_node_context, pipeline_run_ctx_1],
              output_event_type=metadata_store_pb2.Event.INTERNAL_OUTPUT,
          )
  
          # Tests LatestPipelineRunOutputs with the first artifact.
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
              output_keys=["examples"],
          )
          expected_result = {"examples": [input_artifact_1]}
          self.assertAllEqual(result.keys(), expected_result.keys())
          for key in result.keys():
              result_ids = [a.mlmd_artifact.id for a in result[key]]
              expected_ids = [a.id for a in expected_result[key]]
              self.assertAllEqual(result_ids, expected_ids)
  
          # Creates the second pipeline run and artifact.
          input_artifact_2 = self.put_artifact("Example")
          pipeline_run_ctx_2 = self.put_context(
              "pipeline_run", "run-002"
          )
          self.put_execution(
              "ExampleGen",
              inputs={},
              outputs={"examples": [input_artifact_2]},
              contexts=[
                  node_context,
                  pipeline_run_ctx_2,
                  end_node_context,
              ],
              output_event_type=metadata_store_pb2.Event.OUTPUT,
          )
          self.put_execution(
              "pipeline-name_end",
              inputs={},
              outputs={"examples": [input_artifact_2]},
              contexts=[end_node_context, pipeline_run_ctx_2],
              output_event_type=metadata_store_pb2.Event.INTERNAL_OUTPUT,
          )
  
          # Tests LatestPipelineRunOutputs with the first and second artifacts.
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
              output_keys=["examples"],
          )
          expected_result = {"examples": [input_artifact_2]}
          self.assertAllEqual(result.keys(), expected_result.keys())
          for key in result.keys():
              result_ids = [a.mlmd_artifact.id for a in result[key]]
              expected_ids = [a.id for a in expected_result[key]]
              self.assertAllEqual(result_ids, expected_ids)
  
file_path: tfx/tfx/dsl/input_resolution/ops/latest_pipeline_run_outputs_op_test.py
imports: import contextlib
  import tensorflow
  from tfx.dsl.input_resolution.ops import ops
  from tfx.dsl.input_resolution.ops import test_utils
  from tfx.orchestration.portable.input_resolution import exceptions
  from tfx.utils import test_case_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 4 for calling the LatestPipelineRunOutputsTest obj's underlying member method code with example usage: obj.testLatestPipelineRunOutputsOutputs_OneKey() without return types