class_docstr: Test class with extra combiner stats generator related functionality.
docstr: Tests a combiner statistics generator.
  
  This runs the generator twice to cover different behavior. There must be at
  least two input batches in order to test the generator's merging behavior.
  
  Args:
    batches: A list of batches of test data.
    generator: The CombinerStatsGenerator to test.
    expected_feature_stats: Dict mapping feature name to FeatureNameStatistics
      proto that it is expected the generator will return for the feature.
    expected_cross_feature_stats: Dict mapping feature cross to
      CrossFeatureStatistics proto that it is expected the generator will
      return for the feature cross.
    only_match_expected_feature_stats: if True, will only compare features
      that appear in `expected_feature_stats`.
docstr_tok: ["Tests","a","combiner","statistics","generator",".","This","runs","the","generator","twice","to","cover","different","behavior",".","There","must","be","at","least","two","input","batches","in","order","to","test","the","generator","'s","merging","behavior",".","Args",":","batches",":","A","list","of","batches","of","test","data",".","generator",":","The","CombinerStatsGenerator","to","test",".","expected_feature_stats",":","Dict","mapping","feature","name","to","FeatureNameStatistics","proto","that","it","is","expected","the","generator","will","return","for","the","feature",".","expected_cross_feature_stats",":","Dict","mapping","feature","cross","to","CrossFeatureStatistics","proto","that","it","is","expected","the","generator","will","return","for","the","feature","cross",".","only_match_expected_feature_stats",":","if","True",",","will","only","compare","features","that","appear","in","`","expected_feature_stats","`","."]
code: def assertCombinerOutputEqual(
      self,
      batches: List[pa.RecordBatch],
      generator: stats_generator.CombinerStatsGenerator,
      expected_feature_stats: Dict[
          types.FeaturePath, statistics_pb2.FeatureNameStatistics
      ],
      expected_cross_feature_stats: Optional[
          Dict[
              types.FeatureCross, statistics_pb2.CrossFeatureStatistics
          ]
      ] = None,
      only_match_expected_feature_stats: bool = False,
  ) -> None:
      """Tests a combiner statistics generator.
  
      This runs the generator twice to cover different behavior. There must be at
      least two input batches in order to test the generator's merging behavior.
  
      Args:
        batches: A list of batches of test data.
        generator: The CombinerStatsGenerator to test.
        expected_feature_stats: Dict mapping feature name to FeatureNameStatistics
          proto that it is expected the generator will return for the feature.
        expected_cross_feature_stats: Dict mapping feature cross to
          CrossFeatureStatistics proto that it is expected the generator will
          return for the feature cross.
        only_match_expected_feature_stats: if True, will only compare features
          that appear in `expected_feature_stats`.
      """
      generator.setup()
  
      if expected_cross_feature_stats is None:
          expected_cross_feature_stats = {}
  
      def _verify(output):
          """Verifies that the output meeds the expectations."""
          if only_match_expected_feature_stats:
              features_in_stats = set(
                  [
                      types.FeaturePath.from_proto(f.path)
                      for f in output.features
                  ]
              )
              self.assertTrue(
                  set(expected_feature_stats.keys()).issubset(
                      features_in_stats
                  )
              )
          else:
              self.assertEqual(  # pylint: disable=g-generic-assert
                  len(output.features),
                  len(expected_feature_stats),
                  "{}, {}".format(output, expected_feature_stats),
              )
          for actual_feature_stats in output.features:
              actual_path = types.FeaturePath.from_proto(
                  actual_feature_stats.path
              )
              expected_stats = expected_feature_stats.get(actual_path)
              if (
                  only_match_expected_feature_stats
                  and expected_stats is None
              ):
                  continue
              compare.assertProtoEqual(
                  self,
                  expected_stats,
                  actual_feature_stats,
                  normalize_numbers=True,
              )
  
          self.assertEqual(  # pylint: disable=g-generic-assert
              len(result.cross_features),
              len(expected_cross_feature_stats),
              "{}, {}".format(result, expected_cross_feature_stats),
          )
          for actual_cross_feature_stats in result.cross_features:
              cross = (
                  actual_cross_feature_stats.path_x.step[0],
                  actual_cross_feature_stats.path_y.step[0],
              )
              compare.assertProtoEqual(
                  self,
                  expected_cross_feature_stats[cross],
                  actual_cross_feature_stats,
                  normalize_numbers=True,
              )
  
      # Run generator to check that merge_accumulators() works correctly.
      accumulators = [
          generator.add_input(generator.create_accumulator(), batch)
          for batch in batches
      ]
      result = generator.extract_output(
          generator.merge_accumulators(accumulators)
      )
      _verify(result)
  
      # Run generator to check that compact() works correctly after
      # merging accumulators.
      accumulators = [
          generator.add_input(generator.create_accumulator(), batch)
          for batch in batches
      ]
      result = generator.extract_output(
          generator.compact(generator.merge_accumulators(accumulators))
      )
      _verify(result)
  
      # Run generator to check that add_input() works correctly when adding
      # inputs to a non-empty accumulator.
      accumulator = generator.create_accumulator()
  
      for batch in batches:
          accumulator = generator.add_input(accumulator, batch)
  
      result = generator.extract_output(accumulator)
      _verify(result)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import traceback
  from typing import Callable, Dict, Iterable, List, Optional, Tuple, Union
  from absl.testing import absltest
  import apache_beam
  from apache_beam.testing import util
  import pyarrow
  from tensorflow_data_validation import types
  from tensorflow_data_validation.statistics.generators import stats_generator
  from tensorflow.python.util.protobuf import compare
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 1 for calling the CombinerStatsGeneratorTest obj's underlying member method code with example usage: obj.assertCombinerOutputEqual(batches, generator, expected_feature_stats, expected_cross_feature_stats, only_match_expected_feature_stats) without return types