code: def test_generate_tasks_from_one_input(self):
      with self._mlmd_connection as m:
          # Fake one output for _example_gen, so there is 1 input for _transform.
          otu.fake_upstream_node_run(
              m,
              self._example_gen,
              fake_result="Tflex rocks.",
              tmp_path=self.create_tempfile().full_path,
          )
          artifact_types = m.store.get_artifact_types()
          artifacts = artifact_utils.deserialize_artifacts(
              artifact_types[0], m.store.get_artifacts()
          )
          input_and_param = task_gen_utils.InputAndParam(
              input_artifacts={"examples": artifacts}
          )
  
          # Put contexts.
          context_type = metadata_store_pb2.ContextType(
              name="my_ctx_type"
          )
          context_type_id = m.store.put_context_type(context_type)
          contexts = [
              metadata_store_pb2.Context(
                  name="context-1", type_id=context_type_id
              ),
              metadata_store_pb2.Context(
                  name="context-2", type_id=context_type_id
              ),
          ]
          m.store.put_contexts(contexts)
          executions = task_gen_utils.register_executions(
              metadata_handle=m,
              execution_type=self._transform.node_info.type,
              contexts=contexts,
              input_and_params=[input_and_param],
          )
          tasks = task_gen_utils.generate_tasks_from_one_input(
              metadata_handle=m,
              node=self._transform,
              execution=executions[0],
              input_and_param=input_and_param,
              contexts=contexts,
              pipeline=self._pipeline,
              execution_node_state=pstate.NodeState.RUNNING,
          )
  
          self.assertLen(tasks, 2)
          [update_task, exec_task] = tasks
          self.assertIsInstance(
              update_task, task_lib.UpdateNodeStateTask
          )
          self.assertEqual(
              update_task,
              task_lib.UpdateNodeStateTask(
                  task_lib.NodeUid.from_node(
                      self._pipeline, self._transform
                  ),
                  state=pstate.NodeState.RUNNING,
              ),
          )
          self.assertIsInstance(exec_task, task_lib.ExecNodeTask)
  
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 24 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_generate_tasks_from_one_input() without return types