# Copyright 2019 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Tests for Keras text vectorization preprocessing layer."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import tensorflow as tf

import os

from absl.testing import parameterized
import numpy as np

import keras
from keras import backend
from keras import keras_parameterized
from keras import testing_utils
from keras.layers import convolutional
from keras.layers import core
from keras.layers import embeddings
from keras.layers.preprocessing import preprocessing_test_utils
from keras.layers.preprocessing import text_vectorization
from keras.layers.preprocessing import text_vectorization_v1
from keras.utils import generic_utils


def get_layer_class():
  if tf.executing_eagerly():
    return text_vectorization.TextVectorization
  else:
    return text_vectorization_v1.TextVectorization


def _get_end_to_end_test_cases():
  test_cases = (
      {
          "testcase_name":
              "test_simple_tokens_int_mode",
          # Create an array where 'earth' is the most frequent term, followed by
          # 'wind', then 'and', then 'fire'. This ensures that the vocab
          # is sorting by frequency.
          "vocab_data":
              np.array([["fire"], ["earth"], ["earth"], ["earth"], ["earth"],
                        ["wind"], ["wind"], ["wind"], ["and"], ["and"]]),
          "input_data":
              np.array([["earth"], ["wind"], ["and"], ["fire"], ["fire"],
                        ["and"], ["earth"], ["michigan"]]),
          "kwargs": {
              "max_tokens": None,
              "standardize": None,
              "split": None,
              "output_mode": text_vectorization.INT
          },
          "expected_output": [[2], [3], [4], [5], [5], [4], [2], [1]],
      },
      {
          "testcase_name":
              "test_simple_tokens_int_mode_hard_cap",
          # Create an array where 'earth' is the most frequent term, followed by
          # 'wind', then 'and', then 'fire'. This ensures that the vocab
          # is sorting by frequency.
          "vocab_data":
              np.array([["fire"], ["earth"], ["earth"], ["earth"], ["earth"],
                        ["wind"], ["wind"], ["wind"], ["and"], ["and"]]),
          "input_data":
              np.array([["earth"], ["wind"], ["and"], ["fire"], ["fire"],
                        ["and"], ["earth"], ["michigan"]]),
          "kwargs": {
              "max_tokens": 6,
              "standardize": None,
              "split": None,
              "output_mode": text_vectorization.INT
          },
          "expected_output": [[2], [3], [4], [5], [5], [4], [2], [1]],
      },
      {
          "testcase_name":
              "test_documents_int_mode",
          "vocab_data":
              np.array([["fire earth earth"], ["earth earth"], ["wind wind"],
                        ["and wind and"]]),
          "input_data":
              np.array([["earth wind and"], ["fire fire"], ["and earth"],
                        ["michigan"]]),
          "kwargs": {
              "max_tokens": None,
              "standardize": None,
              "split": text_vectorization.SPLIT_ON_WHITESPACE,
              "output_mode": text_vectorization.INT
          },
          "expected_output": [[2, 3, 4], [5, 5, 0], [4, 2, 0], [1, 0, 0]],
      },
      {
          "testcase_name":
              "test_documents_1d_input_int_mode",
          "vocab_data":
              np.array([
                  "fire earth earth", "earth earth", "wind wind", "and wind and"
              ]),
          "input_data":
              np.array([["earth wind and"], ["fire fire"], ["and earth"],
                        ["michigan"]]),
          "kwargs": {
              "max_tokens": None,
              "standardize": None,
              "split": text_vectorization.SPLIT_ON_WHITESPACE,
              "output_mode": text_vectorization.INT
          },
          "expected_output": [[2, 3, 4], [5, 5, 0], [4, 2, 0], [1, 0, 0]],
      },
      {
          "testcase_name":
              "test_simple_tokens_binary_mode",
          "vocab_data":
              np.array([["fire"], ["earth"], ["earth"], ["earth"], ["earth"],
                        ["wind"], ["wind"], ["wind"], ["and"], ["and"]]),
          "input_data":
              np.array([["earth"], ["wind"], ["and"], ["fire"], ["fire"],
                        ["and"], ["earth"], ["michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": None,
              "output_mode": text_vectorization.BINARY
          },
          "expected_output": [[0, 1, 0, 0, 0], [0, 0, 1, 0, 0], [0, 0, 0, 1, 0],
                              [0, 0, 0, 0, 1], [0, 0, 0, 0, 1], [0, 0, 0, 1, 0],
                              [0, 1, 0, 0, 0], [1, 0, 0, 0, 0]],
      },
      {
          "testcase_name":
              "test_documents_binary_mode",
          "vocab_data":
              np.array([["fire earth earth"], ["earth earth"], ["wind wind"],
                        ["and wind and"]]),
          "input_data":
              np.array([["earth wind"], ["and"], ["fire fire"],
                        ["earth michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": text_vectorization.SPLIT_ON_WHITESPACE,
              "output_mode": text_vectorization.BINARY
          },
          "expected_output": [[0, 1, 1, 0, 0], [0, 0, 0, 1, 0], [0, 0, 0, 0, 1],
                              [1, 1, 0, 0, 0]],
      },
      {
          "testcase_name":
              "test_simple_tokens_count_mode",
          "vocab_data":
              np.array([["fire"], ["earth"], ["earth"], ["earth"], ["earth"],
                        ["wind"], ["wind"], ["wind"], ["and"], ["and"]]),
          "input_data":
              np.array([["earth"], ["wind"], ["and"], ["fire"], ["fire"],
                        ["and"], ["earth"], ["michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": None,
              "output_mode": text_vectorization.COUNT
          },
          "expected_output": [[0, 1, 0, 0, 0], [0, 0, 1, 0, 0], [0, 0, 0, 1, 0],
                              [0, 0, 0, 0, 1], [0, 0, 0, 0, 1], [0, 0, 0, 1, 0],
                              [0, 1, 0, 0, 0], [1, 0, 0, 0, 0]],
      },
      {
          "testcase_name":
              "test_documents_count_mode",
          "vocab_data":
              np.array([["fire earth earth"], ["earth earth"], ["wind wind"],
                        ["and wind and"]]),
          "input_data":
              np.array([["earth wind"], ["and"], ["fire fire"],
                        ["earth michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": text_vectorization.SPLIT_ON_WHITESPACE,
              "output_mode": text_vectorization.COUNT
          },
          "expected_output": [[0, 1, 1, 0, 0], [0, 0, 0, 1, 0], [0, 0, 0, 0, 2],
                              [1, 1, 0, 0, 0]],
      },
      {
          "testcase_name":
              "test_tokens_idf_mode",
          "vocab_data":
              np.array([["fire"], ["earth"], ["earth"], ["earth"], ["earth"],
                        ["wind"], ["wind"], ["wind"], ["and"], ["and"]]),
          "input_data":
              np.array([["earth"], ["wind"], ["and"], ["fire"], ["fire"],
                        ["and"], ["earth"], ["michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": None,
              "output_mode": text_vectorization.TFIDF
          },
          "expected_output": [[0, 1.098612, 0, 0, 0], [0, 0, 1.252763, 0, 0],
                              [0, 0, 0, 1.466337, 0], [0, 0, 0, 0, 1.7917595],
                              [0, 0, 0, 0, 1.7917595], [0, 0, 0, 1.4663371, 0],
                              [0, 1.098612, 0, 0, 0], [2.3978953, 0, 0, 0, 0]],
      },
      {
          "testcase_name":
              "test_documents_idf_mode",
          "vocab_data":
              np.array([["fire earth earth"], ["earth earth"], ["wind wind"],
                        ["and wind and"]]),
          "input_data":
              np.array([["earth wind"], ["and"], ["fire fire"],
                        ["earth michigan"]]),
          "kwargs": {
              "max_tokens": 5,
              "standardize": None,
              "split": text_vectorization.SPLIT_ON_WHITESPACE,
              "output_mode": text_vectorization.TFIDF
          },
          "expected_output": [[0., 0.847298, 0.847298, 0., 0.],
                              [0., 0., 0., 1.098612, 0.],
                              [0., 0., 0., 0., 2.197225],
                              [1.609438, 0.847298, 0., 0., 0.]],
      },
  )

  crossed_test_cases = []
  # Cross above test cases with use_dataset in (True, False)
  for use_dataset in (True, False):
    for case in test_cases:
      case = case.copy()
      if use_dataset:
        case["testcase_name"] = case["testcase_name"] + "_with_dataset"
      case["use_dataset"] = use_dataset
      crossed_test_cases.append(case)

  return crossed_test_cases


@keras_parameterized.run_all_keras_modes
class TextVectorizationLayerTest(keras_parameterized.TestCase,
                                 preprocessing_test_utils.PreprocessingLayerTest
                                ):

  @parameterized.named_parameters(*_get_end_to_end_test_cases())
  def test_layer_end_to_end_with_adapt(self, vocab_data, input_data, kwargs,
                                       use_dataset, expected_output):
    cls = get_layer_class()
    if kwargs.get("output_mode") == text_vectorization.INT:
      expected_output_dtype = tf.int64
    else:
      expected_output_dtype = tf.float32
    input_shape = input_data.shape

    if use_dataset:
      # Keras APIs expect batched datasets.
      # TODO(rachelim): `model.predict` predicts the result on each
      # dataset batch separately, then tries to concatenate the results
      # together. When the results have different shapes on the non-concat
      # axis (which can happen in the output_mode = INT case for
      # TextVectorization), the concatenation fails. In real use cases, this may
      # not be an issue because users are likely to pipe the preprocessing layer
      # into other keras layers instead of predicting it directly. A workaround
      # for these unit tests is to have the dataset only contain one batch, so
      # no concatenation needs to happen with the result. For consistency with
      # numpy input, we should make `predict` join differently shaped results
      # together sensibly, with 0 padding.
      input_data = tf.data.Dataset.from_tensor_slices(input_data).batch(
          input_shape[0])
      vocab_data = tf.data.Dataset.from_tensor_slices(vocab_data).batch(
          input_shape[0])

    output_data = testing_utils.layer_test(
        cls,
        kwargs=kwargs,
        input_shape=input_shape,
        input_data=input_data,
        input_dtype=tf.string,
        expected_output_dtype=expected_output_dtype,
        validate_training=False,
        adapt_data=vocab_data)
    self.assertAllClose(expected_output, output_data)

  def test_list_inputs_1d(self):
    vocab_data = ["two two two", "two three three", "three four four five"]
    input_data = ["two three", "four five"]
    layer = get_layer_class()()
    layer.adapt(vocab_data)
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])
    layer.set_vocabulary(["two", "three", "four", "five"])
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])

  def test_tensor_inputs(self):
    vocab_data = tf.constant(
        ["two two two", "two three three", "three four four five"])
    input_data = tf.constant(["two three", "four five"])
    layer = get_layer_class()()
    layer.adapt(vocab_data)
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])
    layer.set_vocabulary(["two", "three", "four", "five"])
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])

  def test_list_inputs_2d(self):
    vocab_data = [
        ["two two two"], ["two three three"], ["three four four five"]]
    input_data = [["two three"], ["four five"]]
    layer = get_layer_class()()
    layer.adapt(vocab_data)
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])
    layer.set_vocabulary(["two", "three", "four", "five"])
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])

  def test_dataset_of_single_strings(self):
    vocab_data = ["two two two", "two three three", "three four four five"]
    input_data = ["two three", "four five"]
    vocab_ds = tf.data.Dataset.from_tensor_slices(vocab_data)  # unbatched
    layer = get_layer_class()()
    layer.adapt(vocab_ds)
    out = layer(input_data)
    if tf.executing_eagerly():
      self.assertAllClose(out.numpy(), [[2, 3], [4, 5]])

  @parameterized.named_parameters(
      {
          "testcase_name": "1d",
          "data": ["0", "a", "b", "c", "d", "e", "a", "b", "c", "d", "f"],
          "expected": [1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 1]
      },
      {
          "testcase_name": "2d",
          "data": [["0", "a", "b", "c", "d"], ["e", "a", "b", "c", "d"], ["f"]],
          "expected": [[1, 2, 3, 4, 5], [1, 2, 3, 4, 5], [1, 0, 0, 0, 0]]
      },
      {
          "testcase_name":
              "3d",
          "data": [[["0", "a", "b"], ["c", "d"]], [["e", "a"], ["b", "c", "d"]],
                   [["f"]]],
          "expected": [[[1, 2, 3], [4, 5, 0]], [[1, 2, 0], [3, 4, 5]],
                       [[1, 0, 0], [0, 0, 0]]]
      },
  )
  def test_layer_dimensionality_handling(self, data, expected):
    vocab = ["a", "b", "c", "d"]
    vectorization = get_layer_class()(
        max_tokens=None, standardize=None, split=None, pad_to_max_tokens=False)
    vectorization.set_vocabulary(vocab)
    output = vectorization(tf.ragged.constant(data))
    self.assertAllEqual(expected, output)

  @parameterized.named_parameters(
      {
          "testcase_name": "1d",
          "data": ["0 a b c d e a b c d f"],
          "expected": [[1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 1]]
      },
      {
          "testcase_name":
              "3d",
          "data": [[["0 a b"], ["c d"]], [["e a"], ["b c d"]], [["f"]]],
          "expected": [[[1, 2, 3], [4, 5, 0]], [[1, 2, 0], [3, 4, 5]],
                       [[1, 0, 0], [0, 0, 0]]]
      },
  )
  def test_layer_dimensionality_handling_with_split(self, data, expected):
    vocab = ["a", "b", "c", "d"]
    vectorization = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        pad_to_max_tokens=False)
    vectorization.set_vocabulary(vocab)
    output = vectorization(tf.ragged.constant(data, inner_shape=(1,)))
    self.assertAllEqual(expected, output)


@keras_parameterized.run_all_keras_modes
class TextVectorizationPreprocessingTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def _write_to_temp_file(self, file_name, vocab_list):
    vocab_path = os.path.join(self.get_temp_dir(), file_name + ".txt")
    with tf.io.gfile.GFile(vocab_path, "w") as writer:
      for vocab in vocab_list:
        writer.write(vocab + "\n")
      writer.flush()
      writer.close()
    return vocab_path

  def test_summary_before_adapt(self):
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=text_vectorization.LOWER_AND_STRIP_PUNCTUATION,
        split=None,
        ngrams=None,
        output_mode=text_vectorization.TFIDF)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    # We are testing that model.summary() can be called without erroring out.
    # (b/145726907)
    model.summary()

  def test_normalization(self):
    input_array = np.array([["Earth", "wInD", "aNd", "firE"],
                            ["fire|", "an<>d", "{earth}", "michigan@%$"]])
    expected_output = np.array([[b"earth", b"wind", b"and", b"fire"],
                                [b"fire", b"and", b"earth", b"michigan"]])

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=text_vectorization.LOWER_AND_STRIP_PUNCTUATION,
        split=None,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_normalization_ragged_inputs(self):
    input_array = tf.ragged.constant([["Earth", "wInD", "aNd", "firE"],
                                               ["fire|", "an<>d", "{earth}"]])
    expected_output = [[b"earth", b"wind", b"and", b"fire"],
                       [b"fire", b"and", b"earth"]]

    input_data = keras.Input(shape=(None,), ragged=True, dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=text_vectorization.LOWER_AND_STRIP_PUNCTUATION,
        split=None,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_custom_normalization(self):
    input_array = np.array([["Earth", "wInD", "aNd", "firE"],
                            ["fire|", "an<>d", "{earth}", "michigan@%$"]])
    expected_output = np.array(
        [[b"earth", b"wind", b"and", b"fire"],
         [b"fire|", b"an<>d", b"{earth}", b"michigan@%$"]])

    custom_standardization = tf.strings.lower
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=custom_standardization,
        split=None,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_string_splitting(self):
    input_array = np.array([["earth wind and fire"],
                            ["\tfire\tand\nearth    michigan  "]])
    expected_output = [[b"earth", b"wind", b"and", b"fire"],
                       [b"fire", b"and", b"earth", b"michigan"]]

    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_custom_string_splitting(self):
    input_array = np.array([["earth>wind>and fire"],
                            ["\tfire>and\nearth>michigan"]])
    expected_output = [[b"earth", b"wind", b"and fire"],
                       [b"\tfire", b"and\nearth", b"michigan"]]

    custom_split = lambda x: tf.strings.split(x, sep=">")
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=custom_split,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_single_ngram_value_ragged_inputs(self):
    input_array = tf.ragged.constant([["earth", "wind", "and", "fire"],
                                               ["fire", "and", "earth"]])
    # pyformat: disable
    expected_output = [[b"earth", b"wind", b"and", b"fire",
                        b"earth wind", b"wind and", b"and fire",
                        b"earth wind and", b"wind and fire"],
                       [b"fire", b"and", b"earth",
                        b"fire and", b"and earth",
                        b"fire and earth"]]
    # pyformat: enable

    input_data = keras.Input(shape=(None,), ragged=True, dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        ngrams=3,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_single_ngram_value(self):
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    # pyformat: disable
    expected_output = [[b"earth", b"wind", b"and", b"fire",
                        b"earth wind", b"wind and", b"and fire",
                        b"earth wind and", b"wind and fire"],
                       [b"fire", b"and", b"earth", b"michigan",
                        b"fire and", b"and earth", b"earth michigan",
                        b"fire and earth", b"and earth michigan"]]
    # pyformat: enable

    input_data = keras.Input(shape=(4,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        ngrams=3,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_multiple_ngram_values(self):
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    # pyformat: disable
    expected_output = [[b"earth wind", b"wind and", b"and fire",
                        b"earth wind and", b"wind and fire"],
                       [b"fire and", b"and earth", b"earth michigan",
                        b"fire and earth", b"and earth michigan"]]
    # pyformat: enable

    input_data = keras.Input(shape=(4,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        ngrams=(2, 3),
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_string_multiple_preprocessing_steps(self):
    input_array = np.array([["earth wInD and firE"],
                            ["\tfire\tand\nearth!!    michig@n  "]])
    expected_output = [[
        b"earth",
        b"wind",
        b"and",
        b"fire",
        b"earth wind",
        b"wind and",
        b"and fire",
    ],
                       [
                           b"fire",
                           b"and",
                           b"earth",
                           b"michign",
                           b"fire and",
                           b"and earth",
                           b"earth michign",
                       ]]

    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=text_vectorization.LOWER_AND_STRIP_PUNCTUATION,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        ngrams=2,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_string_splitting_with_non_1d_array_fails(self):
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=None)
    with self.assertRaisesRegex(RuntimeError,
                                ".*tokenize strings, the innermost dime.*"):
      _ = layer(input_data)

  def test_string_splitting_with_non_1d_raggedarray_fails(self):
    input_data = keras.Input(shape=(None,), ragged=True, dtype=tf.string)
    layer = get_layer_class()(
        vocabulary=["a"],
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=None)
    with self.assertRaisesRegex(RuntimeError,
                                ".*tokenize strings, the innermost dime.*"):
      _ = layer(input_data)

  def test_standardization_with_invalid_standardize_arg(self):
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(vocabulary=["a"])
    layer._standardize = "unsupported"
    with self.assertRaisesRegex(ValueError,
                                ".*is not a supported standardization.*"):
      _ = layer(input_data)

  def test_splitting_with_invalid_split_arg(self):
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(vocabulary=["a"])
    layer._split = "unsupported"
    with self.assertRaisesRegex(ValueError, ".*is not a supported splitting.*"):
      _ = layer(input_data)

  def test_vocab_setting_via_init(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT,
        vocabulary=vocab_data)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)

    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_vocab_setting_via_init_file(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    vocab_path = self._write_to_temp_file("vocab_file", vocab_data)
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT,
        vocabulary=vocab_path)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)

    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_vocab_setting_via_setter(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    vocab_path = self._write_to_temp_file("vocab_file", vocab_data)
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT)
    layer.set_vocabulary(vocab_path)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)

    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)


@keras_parameterized.run_all_keras_modes
class TextVectorizationDistributionTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_distribution_strategy_output(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    strategy = tf.distribute.OneDeviceStrategy("/cpu:0")
    with strategy.scope():
      input_data = keras.Input(shape=(None,), dtype=tf.string)
      layer = get_layer_class()(
          max_tokens=None,
          standardize=None,
          split=None,
          output_mode=text_vectorization.INT)
      layer.set_vocabulary(vocab_data)
      int_data = layer(input_data)
      model = keras.Model(inputs=input_data, outputs=int_data)

    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)


@keras_parameterized.run_all_keras_modes
class TextVectorizationOutputTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_int_output(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_int_output_densifies_with_zeros(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    # Create an input array that has 5 elements in the first example and 4 in
    # the second. This should output a 2x5 tensor with a padding value in the
    # second example.
    input_array = np.array([["earth wind and also fire"],
                            ["fire and earth michigan"]])
    expected_output = [[2, 3, 4, 1, 5], [5, 4, 2, 1, 0]]

    # This test doesn't explicitly set an output shape, so the 2nd dimension
    # should stay 'None'.
    expected_output_shape = [None, None]

    # The input shape here is explicitly 1 because we're tokenizing.
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=text_vectorization.INT)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_int_output_densifies_with_zeros_and_pads(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    # Create an input array that has 5 elements in the first example and 4 in
    # the second. This should output a 2x6 tensor with a padding value in the
    # second example, since output_sequence_length is set to 6.
    input_array = np.array([["earth wind and also fire"],
                            ["fire and earth michigan"]])
    expected_output = [[2, 3, 4, 1, 5, 0], [5, 4, 2, 1, 0, 0]]

    output_sequence_length = 6
    expected_output_shape = [None, output_sequence_length]

    # The input shape here is explicitly 1 because we're tokenizing.
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=text_vectorization.INT,
        output_sequence_length=output_sequence_length)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_int_output_densifies_with_zeros_and_strips(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    # Create an input array that has 5 elements in the first example and 4 in
    # the second. This should output a 2x3 tensor with a padding value in the
    # second example, since output_sequence_length is set to 3.
    input_array = np.array([["earth wind and also fire"],
                            ["fire and earth michigan"]])
    expected_output = [[2, 3, 4], [5, 4, 2]]
    output_sequence_length = 3
    expected_output_shape = [None, output_sequence_length]

    # The input shape here is explicitly 1 because we're tokenizing.
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=text_vectorization.INT,
        output_sequence_length=output_sequence_length)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_int_output_dynamically_strips_and_pads(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    # Create an input array that has 5 elements in the first example and 4 in
    # the second. This should output a 2x3 tensor with a padding value in the
    # second example, since output_sequence_length is set to 3.
    input_array = np.array([["earth wind and also fire"],
                            ["fire and earth michigan"]])
    expected_output = [[2, 3, 4], [5, 4, 2]]
    output_sequence_length = 3
    expected_output_shape = [None, output_sequence_length]

    # The input shape here is explicitly 1 because we're tokenizing.
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=text_vectorization.INT,
        output_sequence_length=output_sequence_length)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

    # Create an input array that has 1 element in the first example and 2 in
    # the second. This should output a 2x3 tensor with a padding value in the
    # second example, since output_sequence_length is set to 3.
    input_array_2 = np.array([["wind"], ["fire and"]])
    expected_output_2 = [[3, 0, 0], [5, 4, 0]]
    output_dataset = model.predict(input_array_2)
    self.assertAllEqual(expected_output_2, output_dataset)

  def test_binary_output_hard_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0, 0],
                       [1, 1, 0, 1, 0, 0]]
    # pyformat: enable
    max_tokens = 6
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=max_tokens,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=True)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_binary_output_soft_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0],
                       [1, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=False)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_bag_output_hard_maximum_set_vocabulary_after_build(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0],
                       [1, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=max_tokens,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=True)
    int_data = layer(input_data)
    layer.set_vocabulary(vocab_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_bag_output_hard_maximum_adapt_after_build(self):
    vocab_data = np.array([
        "earth", "earth", "earth", "earth", "wind", "wind", "wind", "and",
        "and", "fire"
    ])
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0],
                       [1, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=max_tokens,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=True)
    int_data = layer(input_data)
    layer.adapt(vocab_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_bag_output_hard_maximum_set_state_variables_after_build(self):
    state_variables = {
        text_vectorization._VOCAB_NAME: ["earth", "wind", "and", "fire"]
    }
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0],
                       [1, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=max_tokens,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=True)
    int_data = layer(input_data)
    layer._set_state_variables(state_variables)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_bag_output_soft_maximum_set_state_after_build(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 1, 1, 1, 0],
                       [1, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=False)
    layer.build(input_data.shape)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_bag_output_soft_maximum_set_vocabulary_after_call_fails(self):
    vocab_data = ["earth", "wind", "and", "fire"]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=False)
    layer.adapt(vocab_data)
    _ = layer(input_data)
    with self.assertRaisesRegex(RuntimeError, "vocabulary cannot be changed"):
      layer.set_vocabulary(vocab_data)

  def test_bag_output_soft_maximum_set_state_variables_after_call_fails(self):
    state_variables = {
        text_vectorization._VOCAB_NAME: ["earth", "wind", "and", "fire"]
    }

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY,
        pad_to_max_tokens=False)
    layer.adapt(["earth", "wind"])
    _ = layer(input_data)
    with self.assertRaisesRegex(RuntimeError, "vocabulary cannot be changed"):
      layer._set_state_variables(state_variables)

  def test_count_output_hard_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 2, 1, 1, 0, 0],
                       [2, 1, 0, 1, 0, 0]]
    # pyformat: enable
    max_tokens = 6
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=6,
        standardize=None,
        split=None,
        output_mode=text_vectorization.COUNT)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_count_output_soft_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[0, 2, 1, 1, 0],
                       [2, 1, 0, 1, 0]]
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=text_vectorization.COUNT,
        pad_to_max_tokens=False)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

  def test_tfidf_output_hard_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    tfidf_data = [.5, .25, .2, .125]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "fire", "earth", "michigan"]])

    # pyformat: disable
    # pylint: disable=bad-whitespace
    expected_output = [[ 0,  1, .25, .2,    0, 0],
                       [.1, .5,   0,  0, .125, 0]]
    # pylint: enable=bad-whitespace
    # pyformat: enable
    max_tokens = 6
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=6,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF,
        pad_to_max_tokens=True)
    layer.set_vocabulary(
        vocab_data,
        df_data=tfidf_data,
        oov_df_value=.05)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllClose(expected_output, output_dataset)

  def test_tfidf_output_soft_maximum(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    tfidf_data = [.5, .25, .2, .125]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "fire", "earth", "michigan"]])

    # pyformat: disable
    # pylint: disable=bad-whitespace
    expected_output = [[ 0,  1, .25, .2,    0],
                       [.1, .5,   0,  0, .125]]
    # pylint: enable=bad-whitespace
    # pyformat: enable
    max_tokens = 5
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF,
        pad_to_max_tokens=False)
    layer.set_vocabulary(vocab_data, df_data=tfidf_data, oov_df_value=.05)
    int_data = layer(input_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())

    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllClose(expected_output, output_dataset)

  def test_accept_1D_input(self):
    input_array = np.array(["earth wind and fire",
                            "fire and earth michigan"])
    layer = get_layer_class()(
        standardize=None,
        split=None,
        output_mode="int")
    layer.adapt(input_array)
    _ = layer(input_array)


@keras_parameterized.run_all_keras_modes
class TextVectorizationModelBuildingTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  @parameterized.named_parameters(
      {
          "testcase_name": "count_hard_max",
          "pad_to_max_tokens": True,
          "output_mode": text_vectorization.COUNT
      }, {
          "testcase_name": "count_soft_max",
          "pad_to_max_tokens": False,
          "output_mode": text_vectorization.COUNT
      }, {
          "testcase_name": "binary_hard_max",
          "pad_to_max_tokens": True,
          "output_mode": text_vectorization.BINARY
      }, {
          "testcase_name": "binary_soft_max",
          "pad_to_max_tokens": False,
          "output_mode": text_vectorization.BINARY
      }, {
          "testcase_name": "tfidf_hard_max",
          "pad_to_max_tokens": True,
          "output_mode": text_vectorization.TFIDF
      }, {
          "testcase_name": "tfidf_soft_max",
          "pad_to_max_tokens": False,
          "output_mode": text_vectorization.TFIDF
      })
  def test_end_to_end_bagged_modeling(self, output_mode, pad_to_max_tokens):
    vocab_data = ["earth", "wind", "and", "fire"]
    tfidf_data = [.5, .25, .2, .125]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=output_mode,
        pad_to_max_tokens=pad_to_max_tokens)
    if output_mode == text_vectorization.TFIDF:
      layer.set_vocabulary(vocab_data, df_data=tfidf_data, oov_df_value=.05)
    else:
      layer.set_vocabulary(vocab_data)

    int_data = layer(input_data)
    float_data = backend.cast(int_data, dtype="float32")
    output_data = core.Dense(64)(float_data)
    model = keras.Model(inputs=input_data, outputs=output_data)
    _ = model.predict(input_array)

  def test_end_to_end_vocab_modeling(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth wind and also fire"],
                            ["fire and earth michigan"]])
    output_sequence_length = 6
    max_tokens = 5

    # The input shape here is explicitly 1 because we're tokenizing.
    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=text_vectorization.SPLIT_ON_WHITESPACE,
        output_mode=text_vectorization.INT,
        output_sequence_length=output_sequence_length)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    embedded_data = embeddings.Embedding(
        input_dim=max_tokens + 1, output_dim=32)(
            int_data)
    output_data = convolutional.Conv1D(
        250, 3, padding="valid", activation="relu", strides=1)(
            embedded_data)

    model = keras.Model(inputs=input_data, outputs=output_data)
    _ = model.predict(input_array)


@keras_parameterized.run_all_keras_modes(always_skip_eager=True)
class TextVectorizationSaveableTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_ops_are_not_added_with_multiple_saves(self):
    vocab_data = ["earth", "wind", "and", "fire"]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=10,
        standardize=None,
        split=None,
        output_mode=text_vectorization.COUNT,
        pad_to_max_tokens=False)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    weights = model.get_weights()
    model.set_weights(weights)
    keras.backend.get_session().graph.finalize()
    weights = model.get_weights()
    model.set_weights(weights)


@keras_parameterized.run_all_keras_modes
class TextVectorizationErrorTest(keras_parameterized.TestCase,
                                 preprocessing_test_utils.PreprocessingLayerTest
                                ):

  def test_too_long_vocab_fails_in_single_setting(self):
    vocab_data = ["earth", "wind", "and", "fire"]

    layer = get_layer_class()(
        max_tokens=4,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT)
    with self.assertRaisesRegex(ValueError,
                                "vocabulary larger than the maximum vocab.*"):
      layer.set_vocabulary(vocab_data)

  def test_setting_vocab_without_tfidf_data_fails_in_tfidf_mode(self):
    vocab_data = ["earth", "wind", "and", "fire"]

    layer = get_layer_class()(
        max_tokens=5,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF)
    with self.assertRaisesRegex(ValueError,
                                "df_data must be set if output_mode is TFIDF"):
      layer.set_vocabulary(vocab_data)

  def test_tfidf_data_length_mismatch_fails(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    df_data = [1, 2, 3]
    layer = get_layer_class()(
        max_tokens=5,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF)
    with self.assertRaisesRegex(ValueError,
                                "df_data must be the same length as vocab.*"):
      layer.set_vocabulary(vocab_data, df_data)

  def test_tfidf_set_vocab_with_no_oov_fails(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    df_data = [1, 2, 3, 4]
    layer = get_layer_class()(
        max_tokens=5,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF)
    with self.assertRaisesRegex(ValueError,
                                "You must pass an oov_df_value.*"):
      layer.set_vocabulary(vocab_data, df_data)

  def test_set_tfidf_in_non_tfidf_fails(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    df_data = [1, 2, 3, 4]
    layer = get_layer_class()(
        max_tokens=5,
        standardize=None,
        split=None,
        output_mode=text_vectorization.BINARY)
    with self.assertRaisesRegex(ValueError,
                                ".*df_data should only be set if.*"):
      layer.set_vocabulary(vocab_data, df_data)

  def test_zero_max_tokens_fails(self):
    with self.assertRaisesRegex(ValueError, ".*max_tokens.*"):
      _ = get_layer_class()(max_tokens=0)

  def test_non_string_dtype_fails(self):
    with self.assertRaisesRegex(ValueError, ".*dtype of string.*"):
      _ = get_layer_class()(dtype=tf.int64)

  def test_unknown_standardize_arg_fails(self):
    with self.assertRaisesRegex(ValueError,
                                ".*standardize arg.*unsupported_value.*"):
      _ = get_layer_class()(standardize="unsupported_value")

  def test_unknown_split_arg_fails(self):
    with self.assertRaisesRegex(ValueError, ".*split arg.*unsupported_value.*"):
      _ = get_layer_class()(split="unsupported_value")

  def test_unknown_output_mode_arg_fails(self):
    with self.assertRaisesRegex(ValueError,
                                ".*output_mode arg.*unsupported_value.*"):
      _ = get_layer_class()(output_mode="unsupported_value")

  def test_unknown_ngrams_arg_fails(self):
    with self.assertRaisesRegex(ValueError, ".*ngrams.*unsupported_value.*"):
      _ = get_layer_class()(ngrams="unsupported_value")

  def test_float_ngrams_arg_fails(self):
    with self.assertRaisesRegex(ValueError, ".*ngrams.*2.9.*"):
      _ = get_layer_class()(ngrams=2.9)

  def test_float_tuple_ngrams_arg_fails(self):
    with self.assertRaisesRegex(ValueError, ".*ngrams.*(1.3, 2.9).*"):
      _ = get_layer_class()(ngrams=(1.3, 2.9))

  def test_non_int_output_sequence_length_dtype_fails(self):
    with self.assertRaisesRegex(ValueError, ".*output_sequence_length.*2.0.*"):
      _ = get_layer_class()(output_mode="int", output_sequence_length=2.0)

  def test_non_none_output_sequence_length_fails_if_output_type_not_int(self):
    with self.assertRaisesRegex(ValueError,
                                ".*`output_sequence_length` must not be set.*"):
      _ = get_layer_class()(output_mode="count", output_sequence_length=2)


# Custom functions for the custom callable serialization test. Declared here
# to avoid multiple registrations from run_all_keras_modes().
@generic_utils.register_keras_serializable(package="Test")
def custom_standardize_fn(x):
  return tf.strings.lower(x)


@generic_utils.register_keras_serializable(package="Test")
def custom_split_fn(x):
  return tf.strings.split(x, sep=">")


@keras_parameterized.run_all_keras_modes
class TextVectorizationSavingTest(
    keras_parameterized.TestCase,
    preprocessing_test_utils.PreprocessingLayerTest):

  def test_saving(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    # Build and validate a golden model.
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)

    # Save the model to disk.
    output_path = os.path.join(self.get_temp_dir(), "tf_keras_saved_model")

    model.save(output_path, save_format="tf")

    # Delete the session and graph to ensure that the loaded model is generated
    # from scratch.
    # TODO(b/149526183): Can't clear session when TF2 is disabled.
    if tf.__internal__.tf2.enabled():
      keras.backend.clear_session()

    loaded_model = keras.models.load_model(output_path)
    self.assertAllEqual(loaded_model.predict(input_array), expected_output)

  def test_saving_when_nested(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    input_array = np.array([["earth", "wind", "and", "fire"],
                            ["fire", "and", "earth", "michigan"]])
    expected_output = [[2, 3, 4, 5], [5, 4, 2, 1]]

    # Build and validate a golden model.
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=None,
        split=None,
        output_mode=text_vectorization.INT)
    layer.set_vocabulary(vocab_data)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)

    outer_input = keras.Input(shape=(None,), dtype=tf.string)
    outer_output = model(outer_input)
    outer_model = keras.Model(inputs=outer_input, outputs=outer_output)

    # Save the model to disk.
    output_path = os.path.join(self.get_temp_dir(), "tf_keras_saved_model")
    outer_model.save(output_path, save_format="tf")

    # Delete the session and graph to ensure that the loaded model is generated
    # from scratch.
    # TODO(b/149526183): Can't clear session when TF2 is disabled.
    if tf.__internal__.tf2.enabled():
      keras.backend.clear_session()

    loaded_model = keras.models.load_model(output_path)
    self.assertAllEqual(loaded_model.predict(input_array), expected_output)

  def test_saving_with_tfidf(self):
    vocab_data = ["earth", "wind", "and", "fire"]
    tfidf_data = [.5, .25, .2, .125]
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "fire", "earth", "michigan"]])

    # pyformat: disable
    # pylint: disable=bad-whitespace
    expected_output = [[ 0,  1, .25, .2,    0],
                       [.1, .5,   0,  0, .125]]
    # pylint: enable=bad-whitespace
    # pyformat: enable

    # Build and validate a golden model.
    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=5,
        standardize=None,
        split=None,
        output_mode=text_vectorization.TFIDF)
    layer.set_vocabulary(vocab_data, df_data=tfidf_data, oov_df_value=.05)

    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllClose(output_dataset, expected_output)

    # Save the model to disk.
    output_path = os.path.join(self.get_temp_dir(), "tf_keras_saved_model")
    model.save(output_path, save_format="tf")
    loaded_model = keras.models.load_model(output_path)

    # Ensure that the loaded model is unique (so that the save/load is real)
    self.assertIsNot(model, loaded_model)

    # Validate correctness of the new model.
    new_output_dataset = loaded_model.predict(input_array)
    self.assertAllClose(new_output_dataset, expected_output)

  def test_serialization_with_custom_callables(self):
    input_array = np.array([["earth>wind>and Fire"],
                            ["\tfire>And\nearth>michigan"]])
    expected_output = [[b"earth", b"wind", b"and fire"],
                       [b"\tfire", b"and\nearth", b"michigan"]]

    input_data = keras.Input(shape=(1,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=None,
        standardize=custom_standardize_fn,
        split=custom_split_fn,
        ngrams=None,
        output_mode=None)
    int_data = layer(input_data)
    model = keras.Model(inputs=input_data, outputs=int_data)
    output_dataset = model.predict(input_array)
    self.assertAllEqual(expected_output, output_dataset)

    serialized_model_data = model.get_config()
    new_model = keras.Model.from_config(serialized_model_data)
    new_output_dataset = new_model.predict(input_array)
    self.assertAllEqual(expected_output, new_output_dataset)


@keras_parameterized.run_all_keras_modes
class TextVectorizationE2ETest(keras_parameterized.TestCase,
                               preprocessing_test_utils.PreprocessingLayerTest):

  def test_keras_vocab_trimming_example(self):
    vocab_data = np.array([
        "earth", "earth", "earth", "earth", "wind", "wind", "wind", "and",
        "and", "fire"
    ])
    input_array = np.array([["earth", "wind", "and", "earth"],
                            ["ohio", "and", "earth", "michigan"]])

    # pyformat: disable
    expected_output = [[1, 2, 1],
                       [3, 1, 0]]
    # pyformat: enable
    max_tokens = 3
    expected_output_shape = [None, max_tokens]

    input_data = keras.Input(shape=(None,), dtype=tf.string)
    layer = get_layer_class()(
        max_tokens=max_tokens,
        standardize=None,
        split=None,
        output_mode=text_vectorization.COUNT,
        pad_to_max_tokens=True)
    int_data = layer(input_data)
    layer.adapt(vocab_data)
    self.assertAllEqual(expected_output_shape, int_data.shape.as_list())
    model = keras.Model(input_data, int_data)
    output = model.predict(input_array)
    self.assertAllEqual(expected_output, output)


if __name__ == "__main__":
  tf.test.main()
