code: def testDo(
      self,
      custom_validation_config,
      expected_anomalies,
      expected_blessing,
  ):
      source_data_dir = os.path.join(
          os.path.dirname(os.path.dirname(__file__)), "testdata"
      )
  
      eval_stats_artifact = standard_artifacts.ExampleStatistics()
      eval_stats_artifact.uri = os.path.join(
          source_data_dir, "statistics_gen"
      )
      eval_stats_artifact.split_names = (
          artifact_utils.encode_split_names(["train", "eval", "test"])
      )
  
      schema_artifact = standard_artifacts.Schema()
      schema_artifact.uri = os.path.join(source_data_dir, "schema_gen")
  
      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: [
              eval_stats_artifact
          ],
          standard_component_specs.SCHEMA_KEY: [schema_artifact],
      }
  
      if custom_validation_config is not None:
          custom_validation_config = text_format.Parse(
              custom_validation_config,
              custom_validation_config_pb2.CustomValidationConfig(),
          )
      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.CUSTOM_VALIDATION_CONFIG_KEY: custom_validation_config,
      }
  
      output_dict = {
          standard_component_specs.ANOMALIES_KEY: [validation_output],
      }
  
      example_validator_executor = executor.Executor()
      executor_output = example_validator_executor.Do(
          input_dict, output_dict, exec_properties
      )
  
      self.assertEqual(
          artifact_utils.encode_split_names(["train", "eval"]),
          validation_output.split_names,
      )
  
      # Check example_validator outputs.
      train_anomalies_path = os.path.join(
          validation_output.uri, "Split-train", "SchemaDiff.pb"
      )
      eval_anomalies_path = os.path.join(
          validation_output.uri, "Split-eval", "SchemaDiff.pb"
      )
      self.assertTrue(fileio.exists(train_anomalies_path))
      self.assertTrue(fileio.exists(eval_anomalies_path))
      train_anomalies_bytes = io_utils.read_bytes_file(
          train_anomalies_path
      )
      train_anomalies = anomalies_pb2.Anomalies()
      train_anomalies.ParseFromString(train_anomalies_bytes)
      eval_anomalies_bytes = io_utils.read_bytes_file(
          eval_anomalies_path
      )
      eval_anomalies = anomalies_pb2.Anomalies()
      eval_anomalies.ParseFromString(eval_anomalies_bytes)
  
      self._assert_equal_anomalies(train_anomalies, expected_anomalies)
      self._assert_equal_anomalies(eval_anomalies, expected_anomalies)
  
      # Assert 'test' split is excluded.
      train_file_path = os.path.join(
          validation_output.uri, "Split-test", "SchemaDiff.pb"
      )
      self.assertFalse(fileio.exists(train_file_path))
      # TODO(zhitaoli): Add comparison to expected anomolies.
  
      self.assertEqual(
          validation_output.get_json_value_custom_property(
              executor.ARTIFACT_PROPERTY_BLESSED_KEY
          ),
          expected_blessing,
      )
  
      if expected_anomalies:
          alerts = (
              component_generated_alert_pb2.ComponentGeneratedAlertList()
          )
          alerts.component_generated_alert_list.append(
              component_generated_alert_pb2.ComponentGeneratedAlertInfo(
                  alert_name="Feature does not have enough values.",
                  alert_body=(
                      "Custom validation triggered anomaly. Query:"
                      " feature.string_stats.common_stats.min_num_values > 5 Test"
                      " dataset: default slice for feature company in split train."
                  ),
              )
          )
          alerts.component_generated_alert_list.append(
              component_generated_alert_pb2.ComponentGeneratedAlertInfo(
                  alert_name="Feature does not have enough values.",
                  alert_body=(
                      "Custom validation triggered anomaly. Query:"
                      " feature.string_stats.common_stats.min_num_values > 5 Test"
                      " dataset: default slice for feature company in split eval."
                  ),
              )
          )
          alerts_any_proto = any_pb2.Any()
          alerts_any_proto.Pack(alerts)
          self.assertEqual(
              executor_output,
              execution_result_pb2.ExecutorOutput(
                  execution_properties={
                      constants.COMPONENT_GENERATED_ALERTS_KEY: (
                          metadata_store_pb2.Value(
                              proto_value=alerts_any_proto
                          )
                      )
                  },
                  output_artifacts={
                      standard_component_specs.ANOMALIES_KEY: (
                          execution_result_pb2.ExecutorOutput.ArtifactList(
                              artifacts=[
                                  validation_output.mlmd_artifact
                              ]
                          )
                      )
                  },
              ),
          )
      else:
          self.assertEqual(
              executor_output,
              execution_result_pb2.ExecutorOutput(
                  output_artifacts={
                      standard_component_specs.ANOMALIES_KEY: (
                          execution_result_pb2.ExecutorOutput.ArtifactList(
                              artifacts=[
                                  validation_output.mlmd_artifact
                              ]
                          )
                      )
                  },
              ),
          )
  
imports: import os
  import tempfile
  from absl.testing import absltest
  from absl.testing import parameterized
  from tensorflow_data_validation.anomalies.proto import custom_validation_config_pb2
  from tfx.components.example_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.orchestration import execution_result_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 google.protobuf import any_pb2
  from google.protobuf import text_format
  from ml_metadata.proto import metadata_store_pb2
  from tensorflow_metadata.proto.v0 import anomalies_pb2
label_desc: Use image node_id 3 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testDo(custom_validation_config, expected_anomalies, expected_blessing) without return types