code: def testGetNodeExecutions(self):
      c = self.put_context("node", "my-pipeline.my-node")
      e1 = self.put_execution(
          "E", last_known_state="UNKNOWN", contexts=[c]
      )
      e2 = self.put_execution("E", last_known_state="NEW", contexts=[c])
      e3 = self.put_execution(
          "E", last_known_state="RUNNING", contexts=[c]
      )
      e4 = self.put_execution(
          "E", last_known_state="COMPLETE", contexts=[c]
      )
      e5 = self.put_execution(
          "E", last_known_state="FAILED", contexts=[c]
      )
      e6 = self.put_execution(
          "E", last_known_state="CACHED", contexts=[c]
      )
      e7 = self.put_execution(
          "E", last_known_state="CANCELED", contexts=[c]
      )
  
      with self.subTest("With execution_states unspecified"):
          result = store_ext.get_node_executions(
              self.store, pipeline_id="my-pipeline", node_id="my-node"
          )
          self.assertEqual(
              _ids(result), _ids([e1, e2, e3, e4, e5, e6, e7])
          )
  
      with self.subTest("Bad pipeline_id"):
          result = store_ext.get_node_executions(
              self.store, pipeline_id="not-exist", node_id="my-node"
          )
          self.assertEmpty(result)
  
      with self.subTest("Bad node_id"):
          result = store_ext.get_node_executions(
              self.store, pipeline_id="my-pipeline", node_id="not-exist"
          )
          self.assertEmpty(result)
  
      with self.subTest("Multiple execution states"):
          result = store_ext.get_node_executions(
              self.store,
              pipeline_id="my-pipeline",
              node_id="my-node",
              execution_states=[
                  metadata_store_pb2.Execution.RUNNING,
                  metadata_store_pb2.Execution.COMPLETE,
                  metadata_store_pb2.Execution.FAILED,
                  metadata_store_pb2.Execution.CACHED,
                  metadata_store_pb2.Execution.CANCELED,
              ],
          )
          self.assertEqual(_ids(result), _ids([e3, e4, e5, e6, e7]))
  
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 2 for calling the StoreExtTest obj's underlying member method code with example usage: obj.testGetNodeExecutions() without return types