code: def testDo(self, sharded_output: bool, custom_split_uri: bool):
      source_data_dir = os.path.join(
          os.path.dirname(os.path.dirname(__file__)), "testdata"
      )
      output_data_dir = os.path.join(
          os.environ.get(
              "TEST_UNDECLARED_OUTPUTS_DIR", self.get_temp_dir()
          ),
          self._testMethodName,
      )
      fileio.makedirs(output_data_dir)
  
      # Create input dict.
      examples = standard_artifacts.Examples()
      examples.uri = os.path.join(source_data_dir, "csv_example_gen")
  
      if custom_split_uri:
          (
              k,
              v,
          ) = examples_utils.get_custom_split_patterns_key_and_property(
              {
                  "train": "Split-train/*",
                  "eval": "Split-eval/*",
                  "test": "Split-test/*",
              },
          )
          examples.set_string_custom_property(k, v)
      else:
          examples.split_names = artifact_utils.encode_split_names(
              ["train", "eval", "test"]
          )
      examples.span = _TEST_SPAN_NUMBER
  
      input_dict = {
          standard_component_specs.EXAMPLES_KEY: [examples],
      }
  
      exec_properties = {
          # List needs to be serialized before being passed into Do function.
          standard_component_specs.EXCLUDE_SPLITS_KEY: json_utils.dumps(
              ["test"]
          ),
          standard_component_specs.SHARDED_STATS_OUTPUT_KEY: sharded_output,
      }
  
      # Create output dict.
      stats = standard_artifacts.ExampleStatistics()
      stats.uri = output_data_dir
      output_dict = {
          standard_component_specs.STATISTICS_KEY: [stats],
      }
  
      # Run executor.
      stats_gen_executor = executor.Executor()
      stats_gen_executor.Do(input_dict, output_dict, exec_properties)
  
      self.assertEqual(
          artifact_utils.encode_split_names(["train", "eval"]),
          stats.split_names,
      )
      self.assertEqual(
          stats.get_string_custom_property(
              executor.STATS_DASHBOARD_LINK
          ),
          "",
      )
      self.assertEqual(stats.span, _TEST_SPAN_NUMBER)
  
      # Check statistics_gen outputs.
      self._validate_stats_output(
          os.path.join(stats.uri, "Split-train", "FeatureStats.pb")
      )
      self._validate_stats_output(
          os.path.join(stats.uri, "Split-eval", "FeatureStats.pb")
      )
      if sharded_output:
          self._validate_sharded_stats_output(
              os.path.join(
                  stats.uri,
                  "Split-train",
                  "FeatureStats" + tfdv.default_sharded_output_suffix(),
              )
          )
          self._validate_sharded_stats_output(
              os.path.join(
                  stats.uri,
                  "Split-eval",
                  "FeatureStats" + tfdv.default_sharded_output_suffix(),
              )
          )
      else:
          # We want to verify that attempting to load sharded stats produces an
          # error.
          with self.assertRaisesRegex(
              ValueError, "No input paths found.*"
          ):
              self._validate_sharded_stats_output(
                  os.path.join(
                      stats.uri,
                      "Split-train",
                      "FeatureStats"
                      + tfdv.default_sharded_output_suffix(),
                  )
              )
          with self.assertRaisesRegex(
              ValueError, "No input paths found.*"
          ):
              self._validate_sharded_stats_output(
                  os.path.join(
                      stats.uri,
                      "Split-eval",
                      "FeatureStats"
                      + tfdv.default_sharded_output_suffix(),
                  )
              )
  
      # Assert 'test' split is excluded.
      self.assertFalse(
          fileio.exists(
              os.path.join(stats.uri, "test", "FeatureStats.pb")
          )
      )
  
imports: import os
  import tempfile
  from absl.testing import absltest
  from absl.testing import parameterized
  import tensorflow_data_validation
  from tfx.components.statistics_gen import executor
  from tfx.components.util import examples_utils
  from tfx.dsl.io import fileio
  from tfx.types import artifact_utils
  from tfx.types import standard_artifacts
  from tfx.types import standard_component_specs
  from tfx.utils import json_utils
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 5 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testDo(sharded_output, custom_split_uri) without return types