code: def testBucketization(
      self,
      test_inputs,
      expected_boundaries,
      do_shuffle,
      epsilon,
      should_apply,
      is_manual_boundaries,
      input_dtype,
  ):
      test_inputs = list(test_inputs)
  
      # Shuffle the input to add randomness to input generated with
      # simple range().
      if do_shuffle:
          random.shuffle(test_inputs)
  
      def preprocessing_fn(inputs):
          x = tf.cast(inputs["x"], input_dtype)
          num_buckets = len(expected_boundaries) + 1
          if should_apply:
              if is_manual_boundaries:
                  bucket_boundaries = [expected_boundaries]
              else:
                  bucket_boundaries = tft.quantiles(
                      inputs["x"], num_buckets, epsilon
                  )
              result = tft.apply_buckets(x, bucket_boundaries)
          else:
              result = tft.bucketize(
                  x, num_buckets=num_buckets, epsilon=epsilon
              )
          return {"q_b": result}
  
      input_data = [{"x": [x]} for x in test_inputs]
  
      input_metadata = tft.DatasetMetadata.from_feature_spec(
          {
              "x": tf.io.FixedLenFeature(
                  [1], tft_unit.canonical_numeric_dtype(input_dtype)
              )
          }
      )
  
      # Sort the input based on value, index is used to create expected_data.
      indexed_input = enumerate(test_inputs)
      # We put all np.nans in the end of the list so that they get assigned to the
      # last bucket.
      sorted_list = sorted(
          indexed_input,
          key=lambda p: np.inf if np.isnan(p[1]) else p[1],
      )
  
      # Expected data has the same size as input, one bucket per input value.
      expected_data = [None] * len(test_inputs)
      bucket = 0
      for index, x in sorted_list:
          # Increment the bucket number when crossing the boundary
          if (
              bucket < len(expected_boundaries)
              and x >= expected_boundaries[bucket]
          ):
              bucket += 1
          expected_data[index] = {"q_b": [bucket]}
  
      expected_metadata = tft.DatasetMetadata.from_feature_spec(
          {
              "q_b": tf.io.FixedLenFeature([1], tf.int64),
          },
          {
              "q_b": schema_pb2.IntDomain(
                  min=0,
                  max=len(expected_boundaries),
                  is_categorical=True,
              ),
          },
      )
  
      @contextlib.contextmanager
      def no_assert():
          yield None
  
      assertion = no_assert()
      if input_dtype == tf.float16:
          assertion = self.assertRaisesRegexp(
              TypeError,
              ".*DataType float16 not in list of allowed values.*",
          )
  
      with assertion:
          self.assertAnalyzeAndTransformResults(
              input_data,
              input_metadata,
              preprocessing_fn,
              expected_data,
              expected_metadata,
              desired_batch_size=1000,
          )
  
imports: import contextlib
  import random
  import numpy
  import tensorflow
  import tensorflow_transform
  from tensorflow_transform import analyzers
  from tensorflow_transform.beam import impl
  from tensorflow_transform.beam import tft_unit
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 3 for calling the BucketizeIntegrationTest obj's underlying member method code with example usage: obj.testBucketization(test_inputs, expected_boundaries, do_shuffle, epsilon, should_apply, is_manual_boundaries, input_dtype) and returns: dict