code: def test_generate_task_from_active_execution(self):
      with self._mlmd_connection as m:
          # No tasks generated without running execution.
          executions = task_gen_utils.get_executions(m, self._trainer)
          self.assertIsNone(
              task_gen_utils.generate_cancel_task_from_running_execution(
                  m,
                  self._pipeline,
                  self._trainer,
                  executions,
                  task_lib.NodeCancelType.CANCEL_EXEC,
              )
          )
  
      # Next, ensure an active execution for trainer.
      exec_properties = {"int_arg": 24, "list_bool_arg": [True, False]}
      otu.fake_component_output(
          self._mlmd_connection,
          self._trainer,
          exec_properties=exec_properties,
      )
      with self._mlmd_connection as m:
          execution = m.store.get_executions()[0]
          execution.last_known_state = (
              metadata_store_pb2.Execution.RUNNING
          )
          m.store.put_executions([execution])
  
          # Check that task can be generated.
          executions = task_gen_utils.get_executions(m, self._trainer)
          task = task_gen_utils.generate_cancel_task_from_running_execution(
              m,
              self._pipeline,
              self._trainer,
              executions,
              task_lib.NodeCancelType.CANCEL_EXEC,
          )
          self.assertEqual(execution.id, task.execution_id)
          self.assertEqual(exec_properties, task.exec_properties)
  
          # Mark execution complete. No tasks should be generated.
          execution = m.store.get_executions()[0]
          execution.last_known_state = (
              metadata_store_pb2.Execution.COMPLETE
          )
          m.store.put_executions([execution])
          executions = task_gen_utils.get_executions(m, self._trainer)
          self.assertIsNone(
              task_gen_utils.generate_cancel_task_from_running_execution(
                  m,
                  self._pipeline,
                  self._trainer,
                  executions,
                  task_lib.NodeCancelType.CANCEL_EXEC,
              )
          )
  
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 7 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_generate_task_from_active_execution() without return types