code: def testGetNodeExecutionsCapByArtifactCreateTime(self):
      pipeline_run_context_1 = self.put_context(
          "pipeline_run1", "run-20230413"
      )
      pipeline_run_context_2 = self.put_context(
          "pipeline_run1", "run-20230414"
      )
      node_context = self.put_context("node", "my-pipeline.my-node")
      type_id = self.put_artifact_type("test-artifact-type")
      a1 = metadata_store_pb2.Artifact(type_id=type_id, state="DELETED")
      a2 = metadata_store_pb2.Artifact(
          type_id=type_id, state="MARKED_FOR_DELETION"
      )
      a3 = metadata_store_pb2.Artifact(type_id=type_id, state="LIVE")
      a4 = metadata_store_pb2.Artifact(type_id=type_id, state="LIVE")
  
      # Insert 3 executions, sleep 1 second in between to ensure executions don't
      # have the same update time.
      self.put_execution(
          "test-execution-type",
          inputs={"input": [a1]},
          outputs={"output": [a2]},
          contexts=[pipeline_run_context_1, node_context],
      )
      time.sleep(1)
      e2 = self.put_execution(
          "test-execution-type",
          inputs={"input": [a2]},
          outputs={"output": [a3]},
          contexts=[pipeline_run_context_2, node_context],
      )
      time.sleep(1)
      e3 = self.put_execution(
          "test-execution-type",
          inputs={"input": [a3]},
          outputs={"output": [a4]},
          contexts=[pipeline_run_context_2, node_context],
      )
  
      with self.subTest(
          "With execution limited by live artifacts create time"
      ):
          artifacts = store_ext.get_live_output_artifacts_of_node(
              self.store, pipeline_id="my-pipeline", node_id="my-node"
          )
          min_artifact_create_time = min(
              [a.create_time_since_epoch for a in artifacts], default=0
          )
          result = store_ext.get_node_executions(
              self.store,
              pipeline_id="my-pipeline",
              node_id="my-node",
              min_last_update_time_since_epoch=min_artifact_create_time,
          )
          self.assertEqual(_ids(result), _ids([e2, e3]))
  
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 3 for calling the StoreExtTest obj's underlying member method code with example usage: obj.testGetNodeExecutionsCapByArtifactCreateTime() without return types