code: def test_register_execution_from_existing_execution(
      self,
      reset_stateful_working_dir,
      has_previous_stateful_working_dir_index,
  ):
      with self._mlmd_connection as m:
          # 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)
          # Add dynamic exec property to example gen
          ph_value = placeholder_pb2.PlaceholderExpression(
              value=data_types_utils.set_metadata_value(
                  metadata_store_pb2.Value(), "foo_value"
              )
          )
          dynamic_exec_property = (
              self._example_gen.parameters.parameters.get_or_create(
                  "ph_property"
              )
          )
          dynamic_exec_property.placeholder.CopyFrom(ph_value)
  
          # Put a failed execution.
          input_and_param = task_gen_utils.InputAndParam(
              input_artifacts={
                  "input_example": [standard_artifacts.Examples()]
              }
          )
          execution_type = metadata_store_pb2.ExecutionType(
              name="my_ex_type"
          )
          failed_execution = execution_lib.prepare_execution(
              m,
              execution_type,
              metadata_store_pb2.Execution.FAILED,
              input_and_param.exec_properties,
              execution_name=str(uuid.uuid4()),
          )
          failed_execution.custom_properties[
              task_gen_utils._EXTERNAL_EXECUTION_INDEX
          ].int_value = 1
          if has_previous_stateful_working_dir_index:
              failed_execution.custom_properties[
                  constants.STATEFUL_WORKING_DIR_INDEX
              ].string_value = "mocked-failed-index"
          failed_execution.custom_properties[
              "should_not_be_copied"
          ].int_value = 1
          failed_execution = execution_lib.put_execution(
              m,
              failed_execution,
              contexts,
              input_artifacts=input_and_param.input_artifacts,
          )
          # Create stateful working dir.
          mocked_node_dir = os.path.join(
              self.create_tempdir().full_path,
              self._example_gen.node_info.id,
          )
          self._example_gen.execution_options.reset_stateful_working_dir = (
              reset_stateful_working_dir
          )
          # Register a retry execution from a failed execution.
          mocked_new_uuid = "mocked-new-uuid"
          self.enter_context(
              mock.patch.object(
                  outputs_utils.uuid,
                  "uuid4",
                  return_value=mocked_new_uuid,
              )
          )
          self.enter_context(
              mock.patch.object(
                  outputs_utils,
                  "get_node_dir",
                  return_value=mocked_node_dir,
              )
          )
          [
              retry_execution
          ] = task_gen_utils.register_executions_from_existing_executions(
              m,
              self._pipeline,
              node_proto_view.get_view(self._example_gen),
              [failed_execution],
          )
  
          self.assertEqual(
              retry_execution.last_known_state,
              metadata_store_pb2.Execution.NEW,
          )
          self.assertEqual(
              retry_execution.custom_properties[
                  task_gen_utils._EXTERNAL_EXECUTION_INDEX
              ],
              failed_execution.custom_properties[
                  task_gen_utils._EXTERNAL_EXECUTION_INDEX
              ],
          )
          if (
              not reset_stateful_working_dir
              and has_previous_stateful_working_dir_index
          ):
              self.assertEqual(
                  retry_execution.custom_properties[
                      constants.STATEFUL_WORKING_DIR_INDEX
                  ],
                  failed_execution.custom_properties[
                      constants.STATEFUL_WORKING_DIR_INDEX
                  ],
              )
          else:
              self.assertEqual(
                  data_types_utils.get_metadata_value(
                      retry_execution.custom_properties[
                          constants.STATEFUL_WORKING_DIR_INDEX
                      ]
                  ),
                  mocked_new_uuid,
              )
          self.assertEqual(
              retry_execution.custom_properties[
                  "ph_property"
              ].string_value,
              "foo_value",
          )
          self.assertIsNone(
              retry_execution.custom_properties.get(
                  "should_not_be_copied"
              )
          )
          # Check all input artifacts are the same.
          retry_execution_inputs = execution_lib.get_input_artifacts(
              m, retry_execution.id
          )
          failed_execution_inputs = execution_lib.get_input_artifacts(
              m, failed_execution.id
          )
          self.assertEqual(
              retry_execution_inputs.keys(),
              failed_execution_inputs.keys(),
          )
          for key in retry_execution_inputs:
              retry_execution_artifacts_ids = sorted(
                  [a.id for a in retry_execution_inputs[key]]
              )
              failed_execution_artifacts_ids = sorted(
                  [a.id for a in failed_execution_inputs[key]]
              )
              self.assertEqual(
                  retry_execution_artifacts_ids,
                  failed_execution_artifacts_ids,
              )
  
          [context_1, context_2] = m.store.get_contexts()
          self.assertLen(
              m.store.get_executions_by_context(context_1.id), 2
          )
          self.assertLen(
              m.store.get_executions_by_context(context_2.id), 2
          )
  
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 14 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_register_execution_from_existing_execution(reset_stateful_working_dir, has_previous_stateful_working_dir_index) without return types