code: def testGetLiveOutputArtifactsOfNodeByOutputKeySync(self):
      c1 = self.put_context("pipeline_run", "run-20230413")
      c2 = self.put_context("node", "my-pipeline.my-node")
  
      with self.subTest("With no LIVE node artifacts"):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="run-20230413",
              )
          )
          self.assertEmpty(result)
  
      x1 = self.put_artifact("X")
      x2 = self.put_artifact("X")
      x3 = self.put_artifact("X")
      y1 = self.put_artifact("Y")
      y2 = self.put_artifact("Y", state="DELETED")
      y3 = self.put_artifact("Y")
      y4 = self.put_artifact("Y")
      y5 = self.put_artifact("Y")
      z1 = self.put_artifact("Z")
      z2 = self.put_artifact("Z")
      z3 = self.put_artifact("Z", state="ABANDONED")
  
      self.put_execution(
          "E",
          inputs={"x": [x1]},
          outputs={"y": [y1], "z": [z1]},
          contexts=[c1, c2],
      )
      self.put_execution(
          "E",
          inputs={"x": [x2]},
          outputs={"y": [y2, y3, y4], "z": [z2]},
          contexts=[c1, c2],
      )
      self.put_execution(
          "E",
          inputs={"x": [x3]},
          outputs={"y": [y5], "z": [z3]},
          contexts=[c1, c2],
      )
  
      with self.subTest("With execution_states unspecified"):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="run-20230413",
              )
          )
          self.assertDictEqual(
              result,
              {"y": [[y5], [y3, y4], [y1]], "z": [[], [z2], [z1]]},
          )
  
      with self.subTest("With execution_states=[COMPLETE, CACHED]"):
          result = (
              store_ext.get_live_output_artifacts_of_node_by_output_key(
                  self.store,
                  pipeline_id="my-pipeline",
                  node_id="my-node",
                  pipeline_run_id="",
                  execution_states=[
                      metadata_store_pb2.Execution.COMPLETE,
                      metadata_store_pb2.Execution.CACHED,
                  ],
              )
          )
          self.assertDictEqual(
              result,
              {"y": [[y5], [y3, y4], [y1]], "z": [[], [z2], [z1]]},
          )
  
file_path: tfx/tfx/orchestration/portable/mlmd/store_ext_test.py
imports: import time
  import tensorflow
  from tfx.orchestration.portable.mlmd import store_ext
  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 StoreExtTest obj's underlying member method code with example usage: obj.testGetLiveOutputArtifactsOfNodeByOutputKeySync() without return types