code: def test_generate_resolved_info_with_dynamic_exec_prop(self):
      self._pipeline = (
          test_dynamic_exec_properties_pipeline.create_pipeline()
      )
      pipeline_runtime_spec = self._pipeline.runtime_spec
      pipeline_runtime_spec.pipeline_root.field_value.string_value = (
          self._pipeline_root
      )
      pipeline_runtime_spec.pipeline_run_id.field_value.string_value = (
          "test_run_dynamic_prop"
      )
  
      [upstream_node, dynamic_exec_properties_node] = [
          n.pipeline_node for n in self._pipeline.nodes
      ]
  
      self._set_pipeline_context(
          self._pipeline, "pipeline_run", "test_run_dynamic_prop"
      )
      for (
          input_spec
      ) in dynamic_exec_properties_node.inputs.inputs.values():
          for channel in input_spec.channels:
              for context_query in channel.context_queries:
                  if context_query.type.name == "pipeline_run":
                      context_query.name.field_value.string_value = (
                          "test_run_dynamic_prop"
                      )
  
      otu.fake_upstream_node_run(
          self._mlmd_connection,
          upstream_node,
          fake_result="Tflex rocks.",
          tmp_path=self.create_tempfile().full_path,
      )
      resolved_info = task_gen_utils.generate_resolved_info(
          self._mlmd_connection_manager,
          node_proto_view.get_view(dynamic_exec_properties_node),
          self._pipeline,
      )
  
      self.assertCountEqual(
          [
              "my_pipeline",
              "test_run_dynamic_prop",
              "my_pipeline.DownstreamComponent",
          ],
          [c.name for c in resolved_info.contexts],
      )
      self.assertLen(
          resolved_info.input_and_params[0].input_artifacts[
              "_UpstreamComponent.result"
          ],
          1,
      )
      self.assertEqual(
          "Tflex rocks. Especially the run with ID: test_run_dynamic_prop",
          resolved_info.input_and_params[0].exec_properties[
              "input_str"
          ],
      )
  
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 9 for calling the TaskGenUtilsTest obj's underlying member method code with example usage: obj.test_generate_resolved_info_with_dynamic_exec_prop() without return types