code: def testAnomaliesGenerated(
      self,
      config,
      custom_validation_config,
      expected_anomalies,
      anomalies_blessed_value,
      expected_alerts,
  ):
      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"
      )
      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")
  
      validation_config = text_format.Parse(
          config,
          distribution_validator_pb2.DistributionValidatorConfig(),
      )
      if custom_validation_config is not None:
          custom_validation_config = text_format.Parse(
              custom_validation_config,
              custom_validation_config_pb2.CustomValidationConfig(),
          )
  
      input_dict = {
          standard_component_specs.STATISTICS_KEY: [stats_artifact],
          standard_component_specs.BASELINE_STATISTICS_KEY: [
              stats_artifact
          ],
      }
  
      # The analyzed splits are set for this test to get a single result proto.
      exec_properties = {
          # List needs to be serialized before being passed into Do function.
          standard_component_specs.INCLUDE_SPLIT_PAIRS_KEY: json_utils.dumps(
              [("train", "eval")]
          ),
          standard_component_specs.DISTRIBUTION_VALIDATOR_CONFIG_KEY: validation_config,
          standard_component_specs.CUSTOM_VALIDATION_CONFIG_KEY: custom_validation_config,
      }
  
      output_dict = {
          standard_component_specs.ANOMALIES_KEY: [validation_output],
      }
  
      distribution_validator_executor = executor.Executor()
      executor_output = distribution_validator_executor.Do(
          input_dict, output_dict, exec_properties
      )
  
      self.assertEqual(
          artifact_utils.encode_split_names(["train_eval"]),
          validation_output.split_names,
      )
  
      distribution_anomalies_path = os.path.join(
          validation_output.uri, "SplitPair-train_eval", "SchemaDiff.pb"
      )
      self.assertTrue(fileio.exists(distribution_anomalies_path))
      distribution_anomalies_bytes = io_utils.read_bytes_file(
          distribution_anomalies_path
      )
      distribution_anomalies = anomalies_pb2.Anomalies()
      distribution_anomalies.ParseFromString(
          distribution_anomalies_bytes
      )
      expected_anomalies = text_format.Parse(
          expected_anomalies, anomalies_pb2.Anomalies()
      )
  
      self.assertEqualExceptBaseline(
          expected_anomalies, distribution_anomalies
      )
      self.assertEqual(
          validation_output.get_json_value_custom_property(
              executor.ARTIFACT_PROPERTY_BLESSED_KEY
          ),
          {"train_eval": anomalies_blessed_value},
      )
      actual_alerts = (
          component_generated_alert_pb2.ComponentGeneratedAlertList()
      )
      executor_output.execution_properties[
          constants.COMPONENT_GENERATED_ALERTS_KEY
      ].proto_value.Unpack(actual_alerts)
      for alert in expected_alerts.component_generated_alert_list:
          self.assertIn(
              alert, actual_alerts.component_generated_alert_list
          )
  
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 4 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testAnomaliesGenerated(config, custom_validation_config, expected_anomalies, anomalies_blessed_value, expected_alerts) without return types