code: def testLatestPipelineRunOutputs_TwoKeys(self):
      with contextlib.nullcontext():
          example_gen_node_context = self.put_context(
              "node", "example-gen"
          )
          statistics_gen_node_context = self.put_context(
              "node", "statistics-gen"
          )
          end_node_context = self.put_context(
              "node", "pipeline-name.pipeline-name_end"
          )
  
          # First pipeline run generates two artifacts
          example_artifact_1 = self.put_artifact("Example")
          statistics_artifact_1 = self.put_artifact("Statistics")
          pipeline_run_ctx_1 = self.put_context(
              "pipeline_run", "run-001"
          )
          self.put_execution(
              "ExampleGen",
              inputs={},
              outputs={"examples": [example_artifact_1]},
              contexts=[example_gen_node_context, pipeline_run_ctx_1],
              output_event_type=metadata_store_pb2.Event.OUTPUT,
          )
          self.put_execution(
              "StatisticsGen",
              inputs={},
              outputs={"statistics": [statistics_artifact_1]},
              contexts=[
                  statistics_gen_node_context,
                  pipeline_run_ctx_1,
              ],
              output_event_type=metadata_store_pb2.Event.OUTPUT,
          )
          self.put_execution(
              "pipeline-name_end",
              inputs={},
              outputs={
                  "statistics": [statistics_artifact_1],
                  "examples": [example_artifact_1],
              },
              contexts=[end_node_context, pipeline_run_ctx_1],
              output_event_type=metadata_store_pb2.Event.INTERNAL_OUTPUT,
          )
  
          # Second pipeline run generates one artifact
          example_artifact_2 = self.put_artifact("Example")
          pipeline_run_ctx_2 = self.put_context(
              "pipeline_run", "run-002"
          )
          self.put_execution(
              "ExampleGen",
              inputs={},
              outputs={"examples": [example_artifact_2]},
              contexts=[example_gen_node_context, pipeline_run_ctx_2],
              output_event_type=metadata_store_pb2.Event.OUTPUT,
          )
          self.put_execution(
              "pipeline-name_end",
              inputs={},
              outputs={"examples": [example_artifact_2]},
              contexts=[end_node_context, pipeline_run_ctx_2],
              output_event_type=metadata_store_pb2.Event.INTERNAL_OUTPUT,
          )
  
          # Only Examples is output_key
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
              output_keys=["examples"],
          )
          expected_result = {"examples": [example_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)
  
          # Only Statistics is output_key
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
              output_keys=["statistics"],
          )
          self.assertEmpty(result)
  
          # Both Examples and Statistics are output_key
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
              output_keys=["examples", "statistics"],
          )
          expected_result = {"examples": [example_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)
  
          # If output_keys is not provided, use all the keys from producer pipeline.
          result = self._latest_pipeline_run(
              pipeline_name="pipeline-name",
          )
          expected_result = {"examples": [example_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 5 for calling the LatestPipelineRunOutputsTest obj's underlying member method code with example usage: obj.testLatestPipelineRunOutputs_TwoKeys() without return types