code: def testEmptyData(self, stats_train, stats_eval, expected_anomalies):
      source_data_dir = FLAGS.test_tmpdir
      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"]
      )
  
      validation_config = text_format.Parse(
          """
          default_slice_config: {
            feature: {
                path: {
                    step: 'first_feature'
                }
                distribution_comparator: {
                  jensen_shannon_divergence: {
                      threshold: 0.0
                  }
                }
            }
          }
          """,
          distribution_validator_pb2.DistributionValidatorConfig(),
      )
  
      train_stats = text_format.Parse(
          stats_train, statistics_pb2.DatasetFeatureStatisticsList()
      )
      eval_stats = text_format.Parse(
          stats_eval, statistics_pb2.DatasetFeatureStatisticsList()
      )
  
      # Create stats artifacts with a struct feature.
      for split_dir in ["Split-eval", "Split-train"]:
          full_split_dir = os.path.join(stats_artifact.uri, split_dir)
          fileio.makedirs(full_split_dir)
          stats_path = os.path.join(full_split_dir, "FeatureStats.pb")
          if split_dir == "Split-eval":
              io_utils.write_bytes_file(
                  stats_path, eval_stats.SerializeToString()
              )
          else:
              io_utils.write_bytes_file(
                  stats_path, train_stats.SerializeToString()
              )
  
      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
          ],
      }
  
      # 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,
      }
  
      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,
      )
  
      expected_anomalies = text_format.Parse(
          expected_anomalies, anomalies_pb2.Anomalies()
      )
      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
      )
      self.assertEqualExceptBaseline(
          expected_anomalies, distribution_anomalies
      )
  
      expected_alerts = component_generated_alert_pb2.ComponentGeneratedAlertList(
          component_generated_alert_list=[
              component_generated_alert_pb2.ComponentGeneratedAlertInfo(
                  alert_name=(
                      "[train_eval] Comparison could not be done."
                  ),
                  alert_body=(
                      "[train_eval] Validation could not be done, which could be "
                      "due to missing data, use of a comparator that is not "
                      "suitable for the feature type, or some other reason."
                  ),
              ),
          ]
      )
      actual_alerts = (
          component_generated_alert_pb2.ComponentGeneratedAlertList()
      )
      executor_output.execution_properties[
          constants.COMPONENT_GENERATED_ALERTS_KEY
      ].proto_value.Unpack(actual_alerts)
      self.assertEqual(actual_alerts, expected_alerts)
  
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 7 for calling the ExecutorTest obj's underlying member method code with example usage: obj.testEmptyData(stats_train, stats_eval, expected_anomalies) without return types