code: def testSplitPairs(
      self,
      split_pairs,
      expected_split_pair_names,
      expected_error_regex=None,
      exclude_artifact_splits=False,
  ):
      source_data_dir = os.path.join(
          os.path.dirname(os.path.dirname(__file__)), "testdata"
      )
  
      stats_artifact = standard_artifacts.ExampleStatistics()
      stats_artifact.uri = os.path.join(
          source_data_dir, "statistics_gen"
      )
      if not exclude_artifact_splits:
          stats_artifact.split_names = (
              artifact_utils.encode_split_names(["train", "eval"])
          )
  
      output_data_dir = os.path.join(
          os.environ.get(
              "TEST_UNDECLARED_OUTPUTS_DIR", self.get_temp_dir()
          ),
          self._testMethodName,
      )
  
      validation_output = standard_artifacts.ExampleAnomalies()
      validation_output.uri = os.path.join(output_data_dir, "output")
  
      input_dict = {
          standard_component_specs.STATISTICS_KEY: [stats_artifact],
          standard_component_specs.BASELINE_STATISTICS_KEY: [
              stats_artifact
          ],
      }
  
      exec_properties = {
          # List needs to be serialized before being passed into Do function.
          standard_component_specs.INCLUDE_SPLIT_PAIRS_KEY: json_utils.dumps(
              split_pairs
          ),
          standard_component_specs.DISTRIBUTION_VALIDATOR_CONFIG_KEY: distribution_validator_pb2.DistributionValidatorConfig(),
      }
  
      output_dict = {
          standard_component_specs.ANOMALIES_KEY: [validation_output],
      }
  
      distribution_validator_executor = executor.Executor()
      if expected_error_regex:
          with self.assertRaisesRegex(ValueError, expected_error_regex):
              distribution_validator_executor.Do(
                  input_dict, output_dict, exec_properties
              )
      else:
          distribution_validator_executor.Do(
              input_dict, output_dict, exec_properties
          )
  
          for split_pair_name in expected_split_pair_names:
              output_path = os.path.join(
                  validation_output.uri,
                  "SplitPair-" + split_pair_name,
                  "SchemaDiff.pb",
              )
              self.assertTrue(fileio.exists(output_path))
  
          # Confirm that no unexpected result files exist.
          all_outputs = fileio.glob(
              os.path.join(validation_output.uri, "SplitPair-*")
          )
          for output in all_outputs:
              split_pair = output.split("SplitPair-")[1]
              self.assertIn(split_pair, expected_split_pair_names)
  
imports: import os
  import tempfile
  from absl import flags
  from absl.testing import absltest
  from absl.testing import parameterized
  from tensorflow_data_validation.anomalies.proto import custom_validation_config_pb2
  from tfx.components.distribution_validator import executor
  from tfx.dsl.io import fileio
  from tfx.orchestration.experimental.core import component_generated_alert_pb2
  from tfx.orchestration.experimental.core import constants
  from tfx.proto import distribution_validator_pb2
  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 json_utils
  from tfx.utils import test_case_utils
  from google.protobuf import text_format
  from tensorflow_metadata.proto.v0 import anomalies_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 3 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testSplitPairs(split_pairs, expected_split_pair_names, expected_error_regex, exclude_artifact_splits) without return types