class_docstr: Tests for NonStreamingCustomStatsGenerator.
class_docstr_tok: ["Tests","for","NonStreamingCustomStatsGenerator","."]
code: def test_ranklab_mi_with_slicing(self):
      sliced_record_batches = []
      for slice_key in ["slice1", "slice2"]:
          for record_batch in self.record_batches:
              sliced_record_batches.append((slice_key, record_batch))
  
      expected_result = [
          (
              "slice1",
              _get_test_stats_with_mi(
                  [
                      types.FeaturePath(["fa"]),
                      types.FeaturePath(["fb"]),
                      types.FeaturePath(["fc"]),
                      types.FeaturePath(["fd"]),
                  ]
              ),
          ),
          (
              "slice2",
              _get_test_stats_with_mi(
                  [
                      types.FeaturePath(["fa"]),
                      types.FeaturePath(["fb"]),
                      types.FeaturePath(["fc"]),
                      types.FeaturePath(["fd"]),
                  ]
              ),
          ),
      ]
      generator = (
          partitioned_stats_generator.NonStreamingCustomStatsGenerator(
              mutual_information.MutualInformation(
                  label_feature=types.FeaturePath(["label_key"]),
                  schema=self.schema,
                  max_encoding_length=TEST_MAX_ENCODING_LENGTH,
                  seed=TEST_SEED,
              ),
              num_partitions=2,
              min_partitions_stat_presence=2,
              seed=TEST_SEED,
              max_examples_per_partition=1000,
              batch_size=1,
              name="NonStreaming Mutual Information",
          )
      )
      self.assertSlicingAwareTransformOutputEqual(
          sliced_record_batches, generator, expected_result
      )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from absl.testing import absltest
  from absl.testing import parameterized
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import pyarrow
  from tensorflow_data_validation import types
  from tensorflow_data_validation.statistics.generators import mutual_information
  from tensorflow_data_validation.statistics.generators import partitioned_stats_generator
  from tensorflow_data_validation.utils import test_util
  from tfx_bsl.arrow import table_util
  from google.protobuf import text_format
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 4 for calling the NonStreamingCustomStatsGeneratorTest obj's underlying member method code with example usage: obj.test_ranklab_mi_with_slicing() without return types