docstr: Test the entrypoint with dynamic execution properties.
docstr_tok: ["Test","the","entrypoint","with","dynamic","execution","properties","."]
code: def testDynamicExecutionProperties(self):
      """Test the entrypoint with dynamic execution properties."""
  
      test_value_artifact_float_dir = os.path.join(
          self.tmp_dir, "test_value_artifact_float"
      )
      test_value_artifact_string_dir = os.path.join(
          self.tmp_dir, "test_value_artifact_string"
      )
      test_value_artifact_boolean_dir = os.path.join(
          self.tmp_dir, "test_value_artifact_boolean"
      )
      test_value_artifact_integer_dir = os.path.join(
          self.tmp_dir, "test_value_artifact_integer"
      )
      test_executor_output_dir = os.path.join(
          self.tmp_dir, "test_executor_output"
      )
      self._metadata_json_dynamic_execution["outputs"]["artifacts"][
          "test_value_artifact_float"
      ]["artifacts"][0]["uri"] = test_value_artifact_float_dir
      self._metadata_json_dynamic_execution["outputs"]["artifacts"][
          "test_value_artifact_string"
      ]["artifacts"][0]["uri"] = test_value_artifact_string_dir
      self._metadata_json_dynamic_execution["outputs"]["artifacts"][
          "test_value_artifact_boolean"
      ]["artifacts"][0]["uri"] = test_value_artifact_boolean_dir
      self._metadata_json_dynamic_execution["outputs"]["artifacts"][
          "test_value_artifact_integer"
      ]["artifacts"][0]["uri"] = test_value_artifact_integer_dir
      self._metadata_json_dynamic_execution["outputs"][
          "output_file"
      ] = test_executor_output_dir
      serialized_metadata_dynamic_execution = json.dumps(
          self._metadata_json_dynamic_execution
      )
  
      with _ArgsCapture() as args_capture:
          args = [
              "--executor_class_path",
              name_utils.get_full_name(_FakeExecutor),
              "--json_serialized_invocation_args",
              serialized_metadata_dynamic_execution,
          ]
          kubeflow_v2_run_executor.main(
              kubeflow_v2_run_executor._parse_flags(args)
          )
  
          self.assertEqual(
              set(args_capture.output_dict.keys()),
              set(
                  [
                      "output",
                      "test_value_artifact_float",
                      "test_value_artifact_string",
                      "test_value_artifact_boolean",
                      "test_value_artifact_integer",
                  ]
              ),
          )
          executor_output = pipeline_spec_pb2.ExecutorOutput()
          with fileio.open(test_executor_output_dir, "rb") as f:
              json_format.Parse(
                  f.read(), executor_output, ignore_unknown_fields=True
              )
          self.assertDictEqual(
              dict(executor_output.parameter_values),
              {
                  "test_value_artifact_boolean": struct_pb2.Value(
                      bool_value=True
                  ),
                  "test_value_artifact_float": struct_pb2.Value(
                      number_value=1.1
                  ),
                  "test_value_artifact_integer": struct_pb2.Value(
                      number_value=1
                  ),
                  "test_value_artifact_string": struct_pb2.Value(
                      string_value="String ValueArtifact"
                  ),
              },
          )
          self.assertEqual(
              io_utils.read_string_file(test_value_artifact_float_dir),
              "1.1",
          )
          self.assertEqual(
              io_utils.read_string_file(test_value_artifact_string_dir),
              "String ValueArtifact",
          )
          self.assertEqual(
              io_utils.read_string_file(
                  test_value_artifact_boolean_dir
              ),
              "1",
          )
          self.assertEqual(
              io_utils.read_string_file(
                  test_value_artifact_integer_dir
              ),
              "1",
          )
  
imports: import json
  import os
  from typing import Any, Mapping, Sequence
  from unittest import mock
  from kfp.pipeline_spec import pipeline_spec_pb2
  import tensorflow
  from tfx import version
  from tfx.components.evaluator import constants
  from tfx.components.evaluator import executor
  from tfx.dsl.io import fileio
  from tfx.orchestration.kubeflow.v2.container import kubeflow_v2_run_executor
  from tfx.types import artifact
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import name_utils
  from tfx.utils import test_case_utils
  from google.protobuf import struct_pb2
  from google.protobuf import json_format
label_desc: Use image node_id 4 for calling the KubeflowV2RunExecutorTest obj's underlying member method code with example usage: obj.testDynamicExecutionProperties() without return types