code: def test_get_executions(self):
      with self._mlmd_connection as m:
          for node in [n.pipeline_node for n in self._pipeline.nodes]:
              self.assertEmpty(task_gen_utils.get_executions(m, node))
  
      # Create executions for the same nodes under different pipeline contexts.
      self._set_pipeline_context(
          self._pipeline, "pipeline", "my_pipeline1"
      )
      otu.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 1, 1
      )
      otu.fake_example_gen_run(
          self._mlmd_connection, self._example_gen, 2, 1
      )
      otu.fake_component_output(self._mlmd_connection, self._transform)
  
      # Get all executions across all pipeline contexts.
      with self._mlmd_connection as m:
          all_eg_execs = sorted(
              m.store.get_executions_by_type(
                  self._example_gen.node_info.type.name
              ),
              key=lambda e: e.id,
          )
          all_transform_execs = sorted(
              m.store.get_executions_by_type(
                  self._transform.node_info.type.name
              ),
              key=lambda e: e.id,
          )
  
      # Check that correct executions are returned for each node in each pipeline.
      self._set_pipeline_context(
          self._pipeline, "pipeline", "my_pipeline1"
      )
      with self._mlmd_connection as m:
          self.assertCountEqual(
              all_eg_execs[0:2],
              task_gen_utils.get_executions(m, self._example_gen),
          )
          self.assertCountEqual(
              all_transform_execs[0:1],
              task_gen_utils.get_executions(m, self._transform),
          )
          self.assertEmpty(
              task_gen_utils.get_executions(m, self._trainer)
          )
  
          self.assertLen(
              task_gen_utils.get_executions(
                  m, self._example_gen, limit=1
              ),
              1,
          )
          self.assertLen(
              task_gen_utils.get_executions(
                  m, self._example_gen, limit=2
              ),
              2,
          )
  
          all_eg_execs = sorted(
              m.store.get_executions_by_type(
                  self._example_gen.node_info.type.name
              ),
              key=lambda e: e.create_time_since_epoch,
          )
          last_2_executions = task_gen_utils.get_executions(
              m, self._example_gen, limit=2
          )
          self.assertEqual(all_eg_execs[-1].id, last_2_executions[0].id)
          self.assertEqual(all_eg_execs[-2].id, last_2_executions[1].id)
  
          # Fake a FAILED execution. Then, there should be 2 COMPLETED executions
          # and 1 FAILED execution.
          otu.fake_example_gen_execution_with_state(
              self._mlmd_connection,
              self._example_gen,
              metadata_store_pb2.Execution.State.FAILED,
          )
          self.assertLen(
              task_gen_utils.get_executions(m, self._example_gen), 3
          )
  
imports: import os
  import time
  from unittest import mock
  import uuid
  from absl.testing import parameterized
  import tensorflow
  from tfx import types
  from tfx import version
  from tfx.orchestration import data_types_utils
  from tfx.orchestration import node_proto_view
  from tfx.orchestration.experimental.core import constants
  from tfx.orchestration.experimental.core import pipeline_state
  from tfx.orchestration.experimental.core import task
  from tfx.orchestration.experimental.core import task_gen_utils
  from tfx.orchestration.experimental.core import test_utils
  from tfx.orchestration.experimental.core.testing import test_async_pipeline
  from tfx.orchestration.experimental.core.testing import test_dynamic_exec_properties_pipeline
  from tfx.orchestration import mlmd_connection_manager
  from tfx.orchestration.portable import outputs_utils
  from tfx.orchestration.portable.mlmd import execution_lib
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.proto.orchestration import placeholder_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.utils import status
  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 TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_get_executions() without return types