class_docstr: Base test class for testing tf-transform preprocessing functions.
class_docstr_tok: ["Base","test","class","for","testing","tf-transform","preprocessing","functions","."]
docstr: Assert that input data and metadata is transformed as expected.
  
  This methods asserts transformed data and transformed metadata match
  with expected_data and expected_metadata.
  
  Args:
    input_data: Input data formatted in one of three ways:
      * A sequence of dicts whose values are one of:
        strings, lists of strings, numeric types or a pair of those.
        Must have at least one key so that we can infer the batch size, or
      * A sequence of pa.RecordBatch.
      * A Beam source PTransform that produces either of the above.
    input_metadata: One of -
      * DatasetMetadata describing input_data if `input_data` are dicts.
      * TensorAdapterConfig otherwise.
    preprocessing_fn: A function taking a dict of tensors and returning
        a dict of tensors.
    expected_data: (optional) A dataset with the same type constraints as
        input_data, but representing the output after transformation.
        If supplied, transformed data is asserted to be equal.
    expected_metadata: (optional) DatasetMetadata describing the transformed
        data. If supplied, transformed metadata is asserted to be equal.
    expected_vocab_file_contents: (optional) A dictionary from vocab filenames
        to their expected content as a list of text lines or a list of tuples
        of frequency and text. Values should be the expected result of calling
        f.readlines() on the given asset files.
    test_data: (optional) If this is provided then instead of calling
        AnalyzeAndTransformDataset with input_data, this function will call
        AnalyzeDataset with input_data and TransformDataset with test_data.
        Note that this is the case even if input_data and test_data are equal.
        test_data should also conform to input_metadata.
    desired_batch_size: (optional) A batch size to batch elements by. If not
        provided, a batch size will be computed automatically.
    beam_pipeline: (optional) A Beam Pipeline to use in this test.
    temp_dir: If set, it is used as output directory, else a new unique
        directory is created.
    force_tf_compat_v1: A bool. If `True`, TFT's public APIs use Tensorflow
        in compat.v1 mode.
    output_record_batches: (optional) A bool. If `True`, `TransformDataset`
        and `AnalyzeAndTransformDataset` output `pyarrow.RecordBatch`es;
        otherwise, they output instance dicts.
  Raises:
    AssertionError: if the expected data does not match the results of
        transforming input_data according to preprocessing_fn, or
        (if provided) if the expected metadata does not match.
docstr_tok: ["Assert","that","input","data","and","metadata","is","transformed","as","expected",".","This","methods","asserts","transformed","data","and","transformed","metadata","match","with","expected_data","and","expected_metadata",".","Args",":","input_data",":","Input","data","formatted","in","one","of","three","ways",":","*","A","sequence","of","dicts","whose","values","are","one","of",":","strings",",","lists","of","strings",",","numeric","types","or","a","pair","of","those",".","Must","have","at","least","one","key","so","that","we","can","infer","the","batch","size",",","or","*","A","sequence","of","pa.RecordBatch",".","*","A","Beam","source","PTransform","that","produces","either","of","the","above",".","input_metadata",":","One","of","-","*","DatasetMetadata","describing","input_data","if","`","input_data","`","are","dicts",".","*","TensorAdapterConfig","otherwise",".","preprocessing_fn",":","A","function","taking","a","dict","of","tensors","and","returning","a","dict","of","tensors",".","expected_data",":","(","optional",")","A","dataset","with","the","same","type","constraints","as","input_data",",","but","representing","the","output","after","transformation",".","If","supplied",",","transformed","data","is","asserted","to","be","equal",".","expected_metadata",":","(","optional",")","DatasetMetadata","describing","the","transformed","data",".","If","supplied",",","transformed","metadata","is","asserted","to","be","equal",".","expected_vocab_file_contents",":","(","optional",")","A","dictionary","from","vocab","filenames","to","their","expected","content","as","a","list","of","text","lines","or","a","list","of","tuples","of","frequency","and","text",".","Values","should","be","the","expected","result","of","calling","f.readlines","(",")","on","the","given","asset","files",".","test_data",":","(","optional",")","If","this","is","provided","then","instead","of","calling","AnalyzeAndTransformDataset","with","input_data",",","this","function","will","call","AnalyzeDataset","with","input_data","and","TransformDataset","with","test_data",".","Note","that","this","is","the","case","even","if","input_data","and","test_data","are","equal",".","test_data","should","also","conform","to","input_metadata",".","desired_batch_size",":","(","optional",")","A","batch","size","to","batch","elements","by",".","If","not","provided",",","a","batch","size","will","be","computed","automatically",".","beam_pipeline",":","(","optional",")","A","Beam","Pipeline","to","use","in","this","test",".","temp_dir",":","If","set",",","it","is","used","as","output","directory",",","else","a","new","unique","directory","is","created",".","force_tf_compat_v1",":","A","bool",".","If","`","True","`",",","TFT","'s","public","APIs","use","Tensorflow","in","compat.v1","mode",".","output_record_batches",":","(","optional",")","A","bool",".","If","`","True","`",",","`","TransformDataset","`","and","`","AnalyzeAndTransformDataset","`","output","`","pyarrow.RecordBatch","`","es",";","otherwise",",","they","output","instance","dicts",".","Raises",":","AssertionError",":","if","the","expected","data","does","not","match","the","results","of","transforming","input_data","according","to","preprocessing_fn",",","or","(","if","provided",")","if","the","expected","metadata","does","not","match","."]
code: def assertAnalyzeAndTransformResults(
      self,
      input_data,
      input_metadata,
      preprocessing_fn,
      expected_data=None,
      expected_metadata=None,
      expected_vocab_file_contents=None,
      test_data=None,
      desired_batch_size=None,
      beam_pipeline=None,
      temp_dir=None,
      force_tf_compat_v1=False,
      output_record_batches=False,
  ):
      """Assert that input data and metadata is transformed as expected.
  
      This methods asserts transformed data and transformed metadata match
      with expected_data and expected_metadata.
  
      Args:
        input_data: Input data formatted in one of three ways:
          * A sequence of dicts whose values are one of:
            strings, lists of strings, numeric types or a pair of those.
            Must have at least one key so that we can infer the batch size, or
          * A sequence of pa.RecordBatch.
          * A Beam source PTransform that produces either of the above.
        input_metadata: One of -
          * DatasetMetadata describing input_data if `input_data` are dicts.
          * TensorAdapterConfig otherwise.
        preprocessing_fn: A function taking a dict of tensors and returning
            a dict of tensors.
        expected_data: (optional) A dataset with the same type constraints as
            input_data, but representing the output after transformation.
            If supplied, transformed data is asserted to be equal.
        expected_metadata: (optional) DatasetMetadata describing the transformed
            data. If supplied, transformed metadata is asserted to be equal.
        expected_vocab_file_contents: (optional) A dictionary from vocab filenames
            to their expected content as a list of text lines or a list of tuples
            of frequency and text. Values should be the expected result of calling
            f.readlines() on the given asset files.
        test_data: (optional) If this is provided then instead of calling
            AnalyzeAndTransformDataset with input_data, this function will call
            AnalyzeDataset with input_data and TransformDataset with test_data.
            Note that this is the case even if input_data and test_data are equal.
            test_data should also conform to input_metadata.
        desired_batch_size: (optional) A batch size to batch elements by. If not
            provided, a batch size will be computed automatically.
        beam_pipeline: (optional) A Beam Pipeline to use in this test.
        temp_dir: If set, it is used as output directory, else a new unique
            directory is created.
        force_tf_compat_v1: A bool. If `True`, TFT's public APIs use Tensorflow
            in compat.v1 mode.
        output_record_batches: (optional) A bool. If `True`, `TransformDataset`
            and `AnalyzeAndTransformDataset` output `pyarrow.RecordBatch`es;
            otherwise, they output instance dicts.
      Raises:
        AssertionError: if the expected data does not match the results of
            transforming input_data according to preprocessing_fn, or
            (if provided) if the expected metadata does not match.
      """
  
      expected_vocab_file_contents = expected_vocab_file_contents or {}
  
      # Note: we don't separately test AnalyzeDataset and TransformDataset as
      # AnalyzeAndTransformDataset currently simply composes these two
      # transforms.  If in future versions of the code, the implementation
      # differs, we should also run AnalyzeDataset and TransformDataset composed.
      temp_dir = temp_dir or tempfile.mkdtemp(
          prefix=self._testMethodName, dir=self.get_temp_dir()
      )
      with beam_pipeline or self._makeTestPipeline() as pipeline:
          with tft_beam.Context(
              temp_dir=temp_dir,
              desired_batch_size=desired_batch_size,
              force_tf_compat_v1=force_tf_compat_v1,
          ):
              source_ptransform = (
                  input_data
                  if isinstance(input_data, beam.PTransform)
                  else beam.Create(input_data, reshuffle=False)
              )
              input_data = pipeline | "CreateInput" >> source_ptransform
              if test_data is None:
                  (
                      transformed_data,
                      transformed_metadata,
                  ), transform_fn = (
                      input_data,
                      input_metadata,
                  ) | tft_beam.AnalyzeAndTransformDataset(
                      preprocessing_fn,
                      output_record_batches=output_record_batches,
                  )
              else:
                  transform_fn = (
                      input_data,
                      input_metadata,
                  ) | tft_beam.AnalyzeDataset(preprocessing_fn)
                  test_data = pipeline | "CreateTest" >> beam.Create(
                      test_data
                  )
                  transformed_data, transformed_metadata = (
                      (test_data, input_metadata),
                      transform_fn,
                  ) | tft_beam.TransformDataset(
                      output_record_batches=output_record_batches
                  )
  
              # Write transform_fn so we can test its assets
              _ = transform_fn | transform_fn_io.WriteTransformFn(
                  temp_dir
              )
  
              transformed_data_path = os.path.join(
                  temp_dir, "transformed_data"
              )
              if expected_data is not None:
                  _ = (
                      (transformed_data, transformed_metadata)
                      | "Encode" >> tft_beam.EncodeTransformedDataset()
                      | "Write"
                      >> beam.io.tfrecordio.WriteToTFRecord(
                          transformed_data_path, shard_name_template=""
                      )
                  )
  
      # TODO(ebreck) Log transformed_data somewhere.
      tf_transform_output = tft.TFTransformOutput(temp_dir)
      if expected_data is not None:
          examples = tf.compat.v1.python_io.tf_record_iterator(
              path=transformed_data_path
          )
          shapes = {
              f.name: [s.size for s in f.shape.dim]
              if f.HasField("shape")
              else [-1]
              for f in tf_transform_output.transformed_metadata.schema.feature
          }
          transformed_data = [
              _format_example_as_numpy_dict(e, shapes) for e in examples
          ]
          self.assertDataCloseOrEqual(expected_data, transformed_data)
  
      if expected_metadata:
          # Make a copy with no annotations.
          transformed_schema = schema_pb2.Schema()
          transformed_schema.CopyFrom(
              tf_transform_output.transformed_metadata.schema
          )
          transformed_schema.ClearField("annotation")
          for feature in transformed_schema.feature:
              feature.ClearField("annotation")
  
          # assertProtoEqual has a size limit on the length of the
          # serialized as text strings. Therefore, we first try to use
          # assertProtoEqual, if that fails we try to use assertEqual, if that fails
          # as well then we raise the exception from assertProtoEqual.
          try:
              compare.assertProtoEqual(
                  self, expected_metadata.schema, transformed_schema
              )
          except AssertionError as compare_exception:
              try:
                  self.assertEqual(
                      expected_metadata.schema, transformed_schema
                  )
              except AssertionError:
                  raise compare_exception
  
      for (
          filename,
          file_contents,
      ) in expected_vocab_file_contents.items():
          full_filename = tf_transform_output.vocabulary_file_by_name(
              filename
          )
          self.AssertVocabularyContents(full_filename, file_contents)
  
imports: import os
  import tempfile
  from typing import Dict, Iterable, List, Optional, Tuple
  from absl import logging
  import apache_beam
  import pyarrow
  import tensorflow
  import tensorflow_transform
  import tensorflow_transform.beam
  from tensorflow_transform.beam.tft_beam_io import transform_fn_io
  from tensorflow_transform import test_case
  from tensorflow_transform.beam import test_helpers
  from tensorflow_transform.tf_metadata import dataset_metadata
  from tfx_bsl.coders import example_coder
  import unittest
  from tensorflow.python.util.protobuf import compare
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 6 for calling the TransformTestCase obj's underlying member method code with example usage: obj.assertAnalyzeAndTransformResults(input_data, input_metadata, preprocessing_fn, expected_data, expected_metadata, expected_vocab_file_contents, test_data, desired_batch_size, beam_pipeline, temp_dir, force_tf_compat_v1, output_record_batches) without return types