class_docstr: Test class with extra transform stats generator related functionality.
docstr: Tests a slicing aware transform statistics generator.
  
  Args:
    examples: Input sliced examples.
    generator: A TransformStatsGenerator.
    expected_results: Expected statistics proto results.
    metrics_verify_fn: A callable which will be invoked on the resulting
      beam.metrics.metric.MetricResults object.
    add_default_slice_key_to_input: If True, adds the default slice key to
      the input examples.
    add_default_slice_key_to_output: If True, adds the default slice key to
      the result protos.
docstr_tok: ["Tests","a","slicing","aware","transform","statistics","generator",".","Args",":","examples",":","Input","sliced","examples",".","generator",":","A","TransformStatsGenerator",".","expected_results",":","Expected","statistics","proto","results",".","metrics_verify_fn",":","A","callable","which","will","be","invoked","on","the","resulting","beam.metrics.metric.MetricResults","object",".","add_default_slice_key_to_input",":","If","True",",","adds","the","default","slice","key","to","the","input","examples",".","add_default_slice_key_to_output",":","If","True",",","adds","the","default","slice","key","to","the","result","protos","."]
code: def assertSlicingAwareTransformOutputEqual(
      self,
      examples: List[Union[types.SlicedRecordBatch, pa.RecordBatch]],
      generator: stats_generator.TransformStatsGenerator,
      expected_results: List[
          Union[
              statistics_pb2.DatasetFeatureStatistics,
              Tuple[
                  types.SliceKey,
                  statistics_pb2.DatasetFeatureStatistics,
              ],
          ]
      ],
      metrics_verify_fn: Optional[
          Callable[[beam.metrics.metric.MetricResults], None]
      ] = None,
      add_default_slice_key_to_input: bool = False,
      add_default_slice_key_to_output: bool = False,
  ) -> None:
      """Tests a slicing aware transform statistics generator.
  
      Args:
        examples: Input sliced examples.
        generator: A TransformStatsGenerator.
        expected_results: Expected statistics proto results.
        metrics_verify_fn: A callable which will be invoked on the resulting
          beam.metrics.metric.MetricResults object.
        add_default_slice_key_to_input: If True, adds the default slice key to
          the input examples.
        add_default_slice_key_to_output: If True, adds the default slice key to
          the result protos.
      """
  
      def _make_result_matcher(
          test: absltest.TestCase,
          expected_results: List[
              Tuple[
                  types.SliceKey,
                  statistics_pb2.DatasetFeatureStatistics,
              ]
          ],
      ):
          """Makes matcher for a list of DatasetFeatureStatistics protos."""
  
          def _equal(
              actual_results: Iterable[
                  Tuple[
                      types.SliceKey,
                      statistics_pb2.DatasetFeatureStatistics,
                  ]
              ]
          ):
              """Matcher for comparing a list of DatasetFeatureStatistics protos."""
              actual_results = list(actual_results)
              if (
                  len(actual_results) == 1
                  and len(expected_results) == 1
              ):
                  # If appropriate use proto matcher for better errors
                  test.assertEqual(
                      expected_results[0][0], actual_results[0][0]
                  )
                  compare.assertProtoEqual(
                      test,
                      expected_results[0][1],
                      actual_results[0][1],
                      normalize_numbers=True,
                  )
              else:
                  test.assertCountEqual(
                      [
                          (
                              k,
                              _DatasetFeatureStatisticsComparatorWrapper(
                                  v
                              ),
                          )
                          for k, v in expected_results
                      ],
                      [
                          (
                              k,
                              _DatasetFeatureStatisticsComparatorWrapper(
                                  v
                              ),
                          )
                          for k, v in actual_results
                      ],
                  )
  
          return _equal
  
      if add_default_slice_key_to_input:
          examples = [(None, e) for e in examples]
      if add_default_slice_key_to_output:
          expected_results = [(None, p) for p in expected_results]
  
      options = beam.options.pipeline_options.PipelineOptions(
          runtime_type_check=True
      )
      with beam.Pipeline(options=options) as p:
          result = p | beam.Create(examples) | generator.ptransform
          util.assert_that(
              result, _make_result_matcher(self, expected_results)
          )
          pipeline_result = p.run()
          if metrics_verify_fn:
              metrics_verify_fn(pipeline_result.metrics())
  
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 2 for calling the TransformStatsGeneratorTest obj's underlying member method code with example usage: obj.assertSlicingAwareTransformOutputEqual(examples, generator, expected_results, metrics_verify_fn, add_default_slice_key_to_input, add_default_slice_key_to_output) and returns: _equal