test_rasa_layers.py

import pytest
import tensorflow as tf
import numpy as np

from typing import Text, Union, Any, Dict, List, Type

from rasa.shared.nlu.constants import TEXT, FEATURE_TYPE_SENTENCE, FEATURE_TYPE_SEQUENCE
from rasa.utils.tensorflow import layers
from rasa.utils.tensorflow.rasa_layers import (
    ConcatenateSparseDenseFeatures,
    RasaFeatureCombiningLayer,
    RasaSequenceLayer,
    RasaCustomLayer,
)
from rasa.utils.tensorflow.constants import (
    DENSE_INPUT_DROPOUT,
    SPARSE_INPUT_DROPOUT,
    DROP_RATE,
    DENSE_DIMENSION,
    REGULARIZATION_CONSTANT,
    CONCAT_DIMENSION,
    CONNECTION_DENSITY,
    DROP_RATE_ATTENTION,
    KEY_RELATIVE_ATTENTION,
    VALUE_RELATIVE_ATTENTION,
    MAX_RELATIVE_POSITION,
    UNIDIRECTIONAL_ENCODER,
    HIDDEN_LAYERS_SIZES,
    NUM_TRANSFORMER_LAYERS,
    TRANSFORMER_SIZE,
    NUM_HEADS,
    SENTENCE,
    SEQUENCE,
    MASKED_LM,
    LABEL,
)
from rasa.utils.tensorflow.exceptions import TFLayerConfigException
from rasa.utils.tensorflow.model_data import FeatureSignature


attribute_name = TEXT
units_1 = 2
units_2 = 3
units_sparse_to_dense = 10
units_concat = 7
units_hidden_layer = 11
units_transformer = 14
num_transformer_heads = 2
num_transformer_layers = 2
batch_size = 5
max_seq_length = 3


model_config_basic = {
    DENSE_INPUT_DROPOUT: False,
    SPARSE_INPUT_DROPOUT: False,
    DROP_RATE: 0.5,
    DENSE_DIMENSION: {attribute_name: units_sparse_to_dense},
    REGULARIZATION_CONSTANT: 0.001,
    CONCAT_DIMENSION: {attribute_name: units_concat},
    CONNECTION_DENSITY: 0.5,
    HIDDEN_LAYERS_SIZES: {attribute_name: [units_hidden_layer]},
    NUM_TRANSFORMER_LAYERS: 0,
    TRANSFORMER_SIZE: None,
    UNIDIRECTIONAL_ENCODER: None,
    MASKED_LM: False,
}

model_config_basic_no_hidden_layers = dict(
    model_config_basic, **{HIDDEN_LAYERS_SIZES: {attribute_name: []}}
)

model_config_transformer = dict(
    model_config_basic,
    **{
        DROP_RATE_ATTENTION: 0.5,
        KEY_RELATIVE_ATTENTION: True,
        VALUE_RELATIVE_ATTENTION: True,
        MAX_RELATIVE_POSITION: 10,
        UNIDIRECTIONAL_ENCODER: False,
        NUM_TRANSFORMER_LAYERS: {attribute_name: num_transformer_layers},
        TRANSFORMER_SIZE: {attribute_name: units_transformer},
        NUM_HEADS: num_transformer_heads,
    },
)

model_config_transformer_mlm = dict(model_config_transformer, **{MASKED_LM: True})


# Dummy feature signatures and features (full of 1s) for tests that don't check exact
# numerical outputs, only shapes

feature_signature_sparse_1 = FeatureSignature(
    is_sparse=True, units=units_1, number_of_dimensions=3
)
feature_sparse_seq_1 = tf.sparse.from_dense(
    tf.ones((batch_size, max_seq_length, units_1))
)
feature_sparse_sent_1 = tf.sparse.from_dense(tf.ones((batch_size, 1, units_1)))

feature_signature_dense_1 = FeatureSignature(
    is_sparse=False, units=units_1, number_of_dimensions=3
)
feature_dense_seq_1 = tf.ones((batch_size, max_seq_length, units_1))
feature_dense_sent_1 = tf.ones((batch_size, 1, units_1))

feature_signature_dense_2 = FeatureSignature(
    is_sparse=False, units=units_2, number_of_dimensions=3
)
feature_dense_seq_2 = tf.ones((batch_size, max_seq_length, units_2))
feature_dense_sent_2 = tf.ones((batch_size, 1, units_2))

sequence_lengths = tf.ones((batch_size,)) * max_seq_length
sequence_lengths_empty = tf.ones((batch_size,)) * 0

attribute_signature_basic = {
    SEQUENCE: [feature_signature_dense_1, feature_signature_sparse_1],
    SENTENCE: [feature_signature_dense_1],
}
attribute_features_basic = (
    [feature_dense_seq_1, feature_sparse_seq_1],
    [feature_dense_sent_1],
    sequence_lengths,
)


@pytest.mark.parametrize(
    "layer_class, model_config, layer_args, expected_output_units",
    [
        # ConcatenateSparseDense layer with mixed features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [
                    feature_signature_sparse_1,
                    feature_signature_sparse_1,
                    feature_signature_dense_1,
                    feature_signature_dense_2,
                ],
            },
            2 * units_sparse_to_dense + units_1 + units_2,
        ),
        # ConcatenateSparseDense layer with only sparse features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [feature_signature_sparse_1],
            },
            units_sparse_to_dense,
        ),
        # ConcatenateSparseDense layer with only dense features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [feature_signature_dense_1],
            },
            units_1,
        ),
        # FeatureCombining layer with sequence- and sentence-level features, doing
        # dimension unifying
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {"attribute_signature": attribute_signature_basic},
            units_concat,
        ),
        # FeatureCombining layer with sequence- and sentence-level features, no
        # dimension unifying
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    SEQUENCE: [feature_signature_dense_1],
                    SENTENCE: [feature_signature_dense_1],
                }
            },
            units_1,
        ),
        # FeatureCombining layer with sentence-level features only
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    "sequence": [],
                    "sentence": [feature_signature_dense_1],
                }
            },
            units_1,
        ),
        # FeatureCombining layer with sequence-level features only
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    "sequence": [feature_signature_dense_1],
                    "sentence": [],
                }
            },
            units_1,
        ),
        # Sequence layer with mixed features, hidden layers and transformer
        (
            RasaSequenceLayer,
            model_config_transformer,
            {"attribute_signature": attribute_signature_basic},
            units_transformer,
        ),
        # Sequence layer with mixed features, hidden layers, no transformer
        (
            RasaSequenceLayer,
            model_config_basic,
            {"attribute_signature": attribute_signature_basic},
            units_hidden_layer,
        ),
        # Sequence layer with mixed features, no hidden layers, no transformer
        (
            RasaSequenceLayer,
            model_config_basic_no_hidden_layers,
            {"attribute_signature": attribute_signature_basic},
            units_concat,
        ),
    ],
)
def test_layer_gives_correct_output_units(
    layer_class: Type[tf.keras.layers.Layer],
    model_config: Dict[Text, Any],
    layer_args: Dict[Text, Any],
    expected_output_units: int,
) -> None:
    layer = layer_class(**layer_args, config=model_config, attribute=attribute_name)
    assert layer.output_units == expected_output_units


@pytest.mark.parametrize(
    "layer_class, model_config, layer_args, layer_inputs, expected_output_shapes_train,"
    "expected_output_shapes_test",
    [
        # ConcatenateSparseDense layer with mixed features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [
                    feature_signature_sparse_1,
                    feature_signature_sparse_1,
                    feature_signature_dense_1,
                    feature_signature_dense_2,
                ],
            },
            (
                [
                    feature_sparse_seq_1,
                    feature_sparse_seq_1,
                    feature_dense_seq_1,
                    feature_dense_seq_2,
                ],
            ),
            [
                [
                    batch_size,
                    max_seq_length,
                    2 * units_sparse_to_dense + units_1 + units_2,
                ]
            ],
            "same_as_train",  # means that test-time shapes are same as train-time ones
        ),
        # ConcatenateSparseDense layer with only sparse features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [feature_signature_sparse_1],
            },
            ([feature_sparse_sent_1],),
            [[batch_size, 1, units_sparse_to_dense]],
            "same_as_train",
        ),
        # ConcatenateSparseDense layer with only dense features
        (
            ConcatenateSparseDenseFeatures,
            model_config_basic,
            {
                "feature_type": "arbitrary",
                "feature_type_signature": [feature_signature_dense_1],
            },
            ([feature_dense_sent_1],),
            [[batch_size, 1, units_1]],
            "same_as_train",
        ),
        # FeatureCombining layer with sequence- and sentence-level features, dimension
        # unifying
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {"attribute_signature": attribute_signature_basic},
            attribute_features_basic,
            [
                [batch_size, max_seq_length + 1, units_concat],
                [batch_size, max_seq_length + 1, 1],
            ],
            "same_as_train",
        ),
        # FeatureCombining layer with sequence- and sentence-level features, no
        # dimension unifying
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    SEQUENCE: [feature_signature_dense_1],
                    SENTENCE: [feature_signature_dense_1],
                }
            },
            ([feature_dense_seq_1], [feature_dense_sent_1], sequence_lengths),
            [
                [batch_size, max_seq_length + 1, units_1],
                [batch_size, max_seq_length + 1, 1],
            ],
            "same_as_train",
        ),
        # FeatureCombining layer with sentence-level features only
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    "sequence": [],
                    "sentence": [feature_signature_dense_1],
                }
            },
            ([], [feature_dense_sent_1], sequence_lengths_empty),
            [[batch_size, 1, units_1], [batch_size, 1, 1]],
            "same_as_train",
        ),
        # FeatureCombining layer with sequence-level features only
        (
            RasaFeatureCombiningLayer,
            model_config_basic,
            {
                "attribute_signature": {
                    "sequence": [feature_signature_dense_1],
                    "sentence": [],
                }
            },
            ([feature_dense_seq_1], [], sequence_lengths),
            [[batch_size, max_seq_length, units_1], [batch_size, max_seq_length, 1]],
            "same_as_train",
        ),
        # Sequence layer with mixed features, hidden layers and transformer, doing MLM
        (
            RasaSequenceLayer,
            model_config_transformer_mlm,
            {"attribute_signature": attribute_signature_basic},
            attribute_features_basic,
            [
                [batch_size, max_seq_length + 1, units_transformer],
                [batch_size, max_seq_length + 1, units_hidden_layer],
                [batch_size, max_seq_length + 1, 1],
                [batch_size, max_seq_length + 1, units_1],
                [batch_size, max_seq_length + 1, 1],
                [
                    batch_size,
                    num_transformer_layers,
                    num_transformer_heads,
                    max_seq_length + 1,
                    max_seq_length + 1,
                ],
            ],
            [
                [batch_size, max_seq_length + 1, units_transformer],
                [batch_size, max_seq_length + 1, units_hidden_layer],
                [batch_size, max_seq_length + 1, 1],
                [0],
                [0],
                [
                    batch_size,
                    num_transformer_layers,
                    num_transformer_heads,
                    max_seq_length + 1,
                    max_seq_length + 1,
                ],
            ],
        ),
        # Sequence layer with mixed features, hidden layers, no transformer, no MLM
        (
            RasaSequenceLayer,
            model_config_basic,
            {"attribute_signature": attribute_signature_basic},
            attribute_features_basic,
            [
                [batch_size, max_seq_length + 1, units_hidden_layer],
                [batch_size, max_seq_length + 1, units_hidden_layer],
                [batch_size, max_seq_length + 1, 1],
                [0],
                [0],
                [0],
            ],
            "same_as_train",
        ),
        # Sequence layer with mixed features, no hidden layers, no transformer, no MLM
        (
            RasaSequenceLayer,
            model_config_basic_no_hidden_layers,
            {"attribute_signature": attribute_signature_basic},
            attribute_features_basic,
            [
                [batch_size, max_seq_length + 1, units_concat],
                [batch_size, max_seq_length + 1, units_concat],
                [batch_size, max_seq_length + 1, 1],
                [0],
                [0],
                [0],
            ],
            "same_as_train",
        ),
        # Sequence layer with only sequence-level sparse features & MLM (to check the
        # shape of token_ids)
        (
            RasaSequenceLayer,
            model_config_transformer_mlm,
            {
                "attribute_signature": {
                    SEQUENCE: [feature_signature_sparse_1],
                    SENTENCE: [],
                }
            },
            ([feature_sparse_seq_1], [], sequence_lengths),
            [
                [batch_size, max_seq_length, units_transformer],
                [batch_size, max_seq_length, units_hidden_layer],
                [batch_size, max_seq_length, 1],
                [batch_size, max_seq_length, 2],
                [batch_size, max_seq_length, 1],
                [
                    batch_size,
                    num_transformer_layers,
                    num_transformer_heads,
                    max_seq_length,
                    max_seq_length,
                ],
            ],
            [
                [batch_size, max_seq_length, units_transformer],
                [batch_size, max_seq_length, units_hidden_layer],
                [batch_size, max_seq_length, 1],
                [0],
                [0],
                [
                    batch_size,
                    num_transformer_layers,
                    num_transformer_heads,
                    max_seq_length,
                    max_seq_length,
                ],
            ],
        ),
    ],
)
def test_correct_output_shape(
    layer_class: Type[tf.keras.layers.Layer],
    model_config: Dict[Text, Any],
    layer_args: Dict[Text, Any],
    layer_inputs: List[List[Union[tf.SparseTensor, tf.Tensor]]],
    expected_output_shapes_train: List[List[int]],
    expected_output_shapes_test: Union[Text, List[List[int]]],
) -> None:
    layer = layer_class(**layer_args, attribute=attribute_name, config=model_config)

    train_outputs = layer(layer_inputs, training=True)
    if not isinstance(train_outputs, tuple):
        train_outputs = [train_outputs]
    for i, expected_shape in enumerate(expected_output_shapes_train):
        assert train_outputs[i].shape == expected_shape

    if expected_output_shapes_test == "same_as_train":
        expected_output_shapes_test = expected_output_shapes_train
    test_outputs = layer(layer_inputs, training=False)
    if not isinstance(test_outputs, tuple):
        test_outputs = [test_outputs]
    for i, expected_shape in enumerate(expected_output_shapes_test):
        assert test_outputs[i].shape == expected_shape


@pytest.mark.parametrize(
    "layer_class, layer_args",
    [
        # ConcatenateSparseDense layer breaks on empty feature type signature
        (
            ConcatenateSparseDenseFeatures,
            {"feature_type": "arbitrary", "feature_type_signature": []},
        ),
        # FeatureCombining layer breaks on empty attribute signature
        (
            RasaFeatureCombiningLayer,
            {"attribute_signature": {"sequence": [], "sentence": []}},
        ),
        # Sequence layer breaks on no sequence-level features
        (
            RasaSequenceLayer,
            {
                "attribute_signature": {
                    "sequence": [],
                    "sentence": [feature_dense_sent_1],
                }
            },
        ),
    ],
)
def test_raises_exception_when_missing_features(
    layer_class: Type[tf.keras.layers.Layer], layer_args: Dict[Text, Any]
) -> None:
    with pytest.raises(TFLayerConfigException):
        layer_class(**layer_args, attribute=attribute_name, config=model_config_basic)


def test_concat_sparse_dense_raises_exception_when_inconsistent_sparse_features() -> None:  # noqa: E501
    with pytest.raises(TFLayerConfigException):
        ConcatenateSparseDenseFeatures(
            attribute=attribute_name,
            feature_type=SEQUENCE,
            feature_type_signature=[
                FeatureSignature(is_sparse=True, units=2, number_of_dimensions=3),
                FeatureSignature(is_sparse=True, units=1, number_of_dimensions=3),
            ],
            config=model_config_basic,
        )


# Realistic feature signatures and features for checking exact outputs

realistic_feature_signature_dense_1 = FeatureSignature(
    is_sparse=False, units=1, number_of_dimensions=3
)
realistic_feature_dense_seq_1 = tf.convert_to_tensor(
    [[[10.0], [20.0], [30.0]], [[40.0], [50.0], [0.0]]], dtype=tf.float32
)

realistic_feature_signature_dense_2 = FeatureSignature(
    is_sparse=False, units=2, number_of_dimensions=3
)
realistic_feature_dense_seq_2 = tf.convert_to_tensor(
    [[[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]], [[1.5, 2.5], [3.5, 4.5], [0.0, 0.0]]],
    dtype=tf.float32,
)

realistic_feature_signature_dense_3 = FeatureSignature(
    is_sparse=False, units=3, number_of_dimensions=3
)
realistic_feature_dense_sent_3 = tf.convert_to_tensor(
    [[[0.1, 0.2, 0.3]], [[0.4, 0.5, 0.6]]], dtype=tf.float32
)

realistic_sequence_lengths = tf.convert_to_tensor([3, 2], dtype=tf.int32)
realistic_sequence_lengths_empty = tf.convert_to_tensor([0, 0], dtype=tf.int32)


def test_concat_sparse_dense_correct_output_for_dense_input() -> None:
    layer = ConcatenateSparseDenseFeatures(
        attribute=attribute_name,
        feature_type=SEQUENCE,
        feature_type_signature=[
            realistic_feature_signature_dense_1,
            realistic_feature_signature_dense_2,
        ],
        config=dict(
            model_config_basic,
            # also activate all dropout to check that it has no effect on dense features
            **{SPARSE_INPUT_DROPOUT: True, DENSE_INPUT_DROPOUT: True},
        ),
    )
    outputs_expected = [
        [[10.0, 1.0, 2.0], [20.0, 3.0, 4.0], [30.0, 5.0, 6.0]],
        [[40.0, 1.5, 2.5], [50.0, 3.5, 4.5], [0.0, 0.0, 0.0]],
    ]
    inputs = ([realistic_feature_dense_seq_1, realistic_feature_dense_seq_2],)
    train_outputs = layer(inputs, training=True)
    assert (train_outputs.numpy() == outputs_expected).all()
    test_outputs = layer(inputs, training=False)
    assert (test_outputs.numpy() == outputs_expected).all()


def test_concat_sparse_dense_applies_dropout_to_sparse_input() -> None:
    layer_dropout_for_sparse = ConcatenateSparseDenseFeatures(
        attribute=attribute_name,
        feature_type=SEQUENCE,
        feature_type_signature=[feature_signature_sparse_1, feature_signature_sparse_1],
        config=dict(model_config_basic, **{SPARSE_INPUT_DROPOUT: True, DROP_RATE: 1.0}),
    )

    inputs = ([feature_sparse_seq_1, feature_sparse_seq_1],)
    expected_outputs_train = tf.zeros(
        (batch_size, max_seq_length, units_sparse_to_dense * 2)
    )

    train_outputs = layer_dropout_for_sparse(inputs, training=True)
    assert np.allclose(train_outputs.numpy(), expected_outputs_train.numpy())

    # We can't check exact output contents for sparse inputs but during test-time no
    # dropout should be applied, hence the outputs should not be all zeros in this case
    # (unlike at training time).
    test_outputs = layer_dropout_for_sparse(inputs, training=False)
    assert not np.allclose(test_outputs.numpy(), expected_outputs_train.numpy())


def test_concat_sparse_dense_applies_dropout_to_sparse_densified_input() -> None:
    layer_dropout_for_sparse_densified = ConcatenateSparseDenseFeatures(
        attribute=attribute_name,
        feature_type=SEQUENCE,
        feature_type_signature=[feature_signature_sparse_1, feature_signature_sparse_1],
        config=dict(
            model_config_basic, **{DENSE_INPUT_DROPOUT: True, DROP_RATE: 0.99999999}
        ),  # keras dropout doesn't accept velues >= 1.0
    )

    inputs = ([feature_sparse_seq_1, feature_sparse_seq_1],)
    expected_outputs_train = tf.zeros(
        (batch_size, max_seq_length, units_sparse_to_dense * 2)
    )

    train_outputs = layer_dropout_for_sparse_densified(inputs, training=True)
    assert np.allclose(train_outputs.numpy(), expected_outputs_train.numpy())

    # We can't check exact output contents for sparse inputs but during test-time no
    # dropout should be applied, hence the outputs should not be all zeros in this case
    # (unlike at training time).
    test_outputs = layer_dropout_for_sparse_densified(inputs, training=False)
    assert not np.allclose(test_outputs.numpy(), expected_outputs_train.numpy())


@pytest.mark.parametrize(
    "attribute_signature, inputs, expected_outputs_train, expected_outputs_test",
    [
        # Both sequence- and sentence-level features, not unifying dimensions before
        # concatenation
        (
            {
                SEQUENCE: [
                    realistic_feature_signature_dense_1,
                    realistic_feature_signature_dense_2,
                ],
                SENTENCE: [realistic_feature_signature_dense_3],
            },
            (
                [realistic_feature_dense_seq_1, realistic_feature_dense_seq_2],
                [realistic_feature_dense_sent_3],
                realistic_sequence_lengths,
            ),
            (
                np.array(
                    [
                        [
                            [10.0, 1.0, 2.0],
                            [20.0, 3.0, 4.0],
                            [30.0, 5.0, 6.0],
                            [0.1, 0.2, 0.3],
                        ],
                        [
                            [40.0, 1.5, 2.5],
                            [50.0, 3.5, 4.5],
                            [0.4, 0.5, 0.6],
                            [0.0, 0.0, 0.0],
                        ],
                    ],
                    dtype=np.float32,
                ),
                np.array([[[1.0], [1.0], [1.0], [1.0]], [[1.0], [1.0], [1.0], [0.0]]]),
            ),
            "same_as_train",
        ),
        # Sequence-level features only
        (
            {
                SEQUENCE: [
                    realistic_feature_signature_dense_1,
                    realistic_feature_signature_dense_2,
                ],
                SENTENCE: [],
            },
            (
                [realistic_feature_dense_seq_1, realistic_feature_dense_seq_2],
                [],
                realistic_sequence_lengths,
            ),
            (
                np.array(
                    [
                        [[10.0, 1.0, 2.0], [20.0, 3.0, 4.0], [30.0, 5.0, 6.0]],
                        [[40.0, 1.5, 2.5], [50.0, 3.5, 4.5], [0.0, 0.0, 0.0]],
                    ],
                    dtype=np.float32,
                ),
                np.array([[[1.0], [1.0], [1.0]], [[1.0], [1.0], [0.0]]]),
            ),
            "same_as_train",
        ),
        # Sentence-level features only
        (
            {SEQUENCE: [], SENTENCE: [realistic_feature_signature_dense_3]},
            ([], [realistic_feature_dense_sent_3], realistic_sequence_lengths_empty),
            (realistic_feature_dense_sent_3.numpy(), [[[1.0]], [[1.0]]]),
            "same_as_train",
        ),
    ],
)
def test_feature_combining_correct_output(
    attribute_signature: Dict[Text, List[FeatureSignature]],
    inputs: List[List[Union[tf.SparseTensor, tf.Tensor]]],
    expected_outputs_train: List[np.ndarray],
    expected_outputs_test: Union[Text, List[np.ndarray]],
) -> None:
    layer = RasaFeatureCombiningLayer(
        attribute=attribute_name,
        config=model_config_basic,
        attribute_signature=attribute_signature,
    )
    if expected_outputs_test == "same_as_train":
        expected_outputs_test = expected_outputs_train

    train_outputs, train_mask_seq_sent = layer(inputs, training=True)
    assert (train_outputs.numpy() == expected_outputs_train[0]).all()
    assert (train_mask_seq_sent.numpy() == expected_outputs_train[1]).all()

    test_outputs, test_mask_seq_sent = layer(inputs, training=False)
    assert (test_outputs.numpy() == expected_outputs_test[0]).all()
    assert (test_mask_seq_sent.numpy() == expected_outputs_test[1]).all()


@pytest.mark.parametrize(
    "attribute_signature, inputs, expected_outputs_train",
    [
        # Both sequence- and sentence-level features
        (
            {
                SEQUENCE: [
                    realistic_feature_signature_dense_1,
                    realistic_feature_signature_dense_2,
                ],
                SENTENCE: [realistic_feature_signature_dense_3],
            },
            (
                [realistic_feature_dense_seq_1, realistic_feature_dense_seq_2],
                [realistic_feature_dense_sent_3],
                realistic_sequence_lengths,
            ),
            (
                np.array(
                    [
                        [
                            [10.0, 1.0, 2.0],
                            [20.0, 3.0, 4.0],
                            [30.0, 5.0, 6.0],
                            [0.1, 0.2, 0.3],
                        ],
                        [
                            [40.0, 1.5, 2.5],
                            [50.0, 3.5, 4.5],
                            [0.4, 0.5, 0.6],
                            [0.0, 0.0, 0.0],
                        ],
                    ],
                    dtype=np.float32,
                ),
                np.array([[[1.0], [1.0], [1.0], [1.0]], [[1.0], [1.0], [1.0], [0.0]]]),
                np.concatenate(
                    (realistic_feature_dense_seq_1, [[[0.0]], [[0.0]]]), axis=1
                ),
            ),
        ),
        # Only sequence-level features
        (
            {
                SEQUENCE: [
                    realistic_feature_signature_dense_1,
                    realistic_feature_signature_dense_2,
                ],
                SENTENCE: [],
            },
            (
                [realistic_feature_dense_seq_1, realistic_feature_dense_seq_2],
                [],
                realistic_sequence_lengths,
            ),
            (
                np.array(
                    [
                        [[10.0, 1.0, 2.0], [20.0, 3.0, 4.0], [30.0, 5.0, 6.0]],
                        [[40.0, 1.5, 2.5], [50.0, 3.5, 4.5], [0.0, 0.0, 0.0]],
                    ],
                    dtype=np.float32,
                ),
                np.array([[[1.0], [1.0], [1.0]], [[1.0], [1.0], [0.0]]]),
                realistic_feature_dense_seq_1.numpy(),
            ),
        ),
    ],
)
def test_sequence_layer_correct_output(
    attribute_signature: Dict[Text, List[FeatureSignature]],
    inputs: List[Union[tf.Tensor, List[Union[tf.SparseTensor, tf.Tensor]]]],
    expected_outputs_train: List[np.ndarray],
) -> None:
    layer = RasaSequenceLayer(
        attribute=attribute_name,
        # Use MLM but no transformer and no hidden layers.
        config=dict(model_config_basic_no_hidden_layers, **{MASKED_LM: True}),
        attribute_signature=attribute_signature,
    )

    # Training-time check
    (
        seq_sent_features_expected,
        mask_seq_sent_expected,
        token_ids_expected,
    ) = expected_outputs_train
    (_, seq_sent_features, mask_seq_sent, token_ids, mlm_boolean_mask, _) = layer(
        inputs, training=True
    )
    assert (seq_sent_features.numpy() == seq_sent_features_expected).all()
    assert (mask_seq_sent.numpy() == mask_seq_sent_expected).all()
    assert (token_ids.numpy() == token_ids_expected).all()
    assert mlm_boolean_mask.dtype == bool
    # no masking at the padded position found in the shorter sequence
    assert not mlm_boolean_mask[-1][-1][0]
    # when sentence-level features are present, also ensure that no masking is done at
    # sentence-level feature positions (determined by sequence lengths)
    if len(attribute_signature[SENTENCE]) > 0:
        assert not mlm_boolean_mask.numpy()[0][realistic_sequence_lengths.numpy()][0]

    # Test-time check
    (seq_sent_features_expected, mask_seq_sent_expected, _) = expected_outputs_train
    (
        transformer_outputs,
        seq_sent_features,
        mask_seq_sent,
        token_ids,
        mlm_boolean_mask,
        _,
    ) = layer(inputs, training=False)
    # Check that transformer outputs match the combined features, i.e. that MLM wasn't
    # applied
    assert (transformer_outputs.numpy() == seq_sent_features_expected).all()
    assert (seq_sent_features.numpy() == seq_sent_features_expected).all()
    assert (mask_seq_sent.numpy() == mask_seq_sent_expected).all()
    assert token_ids.numpy().size == 0
    assert mlm_boolean_mask.numpy().size == 0


@pytest.mark.parametrize(
    "new_sparse_feature_sizes, old_sparse_feature_sizes, feature_type, use_bias",
    [
        ([10, 10, 10], [3, 2, 3], FEATURE_TYPE_SENTENCE, True),
        ([10, 10, 10], [1, 5, 2], FEATURE_TYPE_SEQUENCE, False),
        ([3, 3, 3], [3, 3, 3], FEATURE_TYPE_SEQUENCE, True),
        ([8], [3], FEATURE_TYPE_SENTENCE, False),
    ],
)
def test_replace_dense_for_sparse_layers(
    new_sparse_feature_sizes: List[int],
    old_sparse_feature_sizes: List[int],
    feature_type: Text,
    use_bias: bool,
):
    """Tests if `DenseForSparse` layers are adjusted correctly."""
    output_units = 10
    kernel_initializer = tf.constant_initializer(
        np.random.random((sum(old_sparse_feature_sizes), output_units))
    )
    layer = layers.DenseForSparse(
        units=output_units, kernel_initializer=kernel_initializer, use_bias=use_bias
    )
    layer.build(input_shape=sum(old_sparse_feature_sizes))

    new_layer = RasaCustomLayer._replace_dense_for_sparse_layer(
        layer_to_replace=layer,
        new_sparse_feature_sizes=new_sparse_feature_sizes,
        old_sparse_feature_sizes=old_sparse_feature_sizes,
        attribute=TEXT,
        feature_type=feature_type,
        reg_lambda=0.02,
    )
    new_layer.build(input_shape=sum(new_sparse_feature_sizes))

    # check dimensions
    assert new_layer.get_kernel().shape[0] == sum(new_sparse_feature_sizes)

    # check if bias tensor was preserved correctly
    if use_bias:
        assert np.array_equal(layer.get_bias().numpy(), new_layer.get_bias().numpy())
    else:
        assert new_layer.get_bias() is None

    # check if the existing weights were preserved
    chunk_index, new_chunk_index = 0, 0
    kernel, new_kernel = layer.get_kernel().numpy(), new_layer.get_kernel().numpy()
    for old_size, new_size in zip(old_sparse_feature_sizes, new_sparse_feature_sizes):
        chunk = kernel[chunk_index : chunk_index + old_size, :]
        new_chunk = new_kernel[new_chunk_index : new_chunk_index + old_size, :]
        assert np.array_equal(chunk, new_chunk)
        chunk_index += old_size
        new_chunk_index += new_size


@pytest.mark.parametrize(
    "new_sparse_feature_sizes, old_sparse_feature_sizes",
    [
        (
            {
                TEXT: {
                    FEATURE_TYPE_SENTENCE: [10, 5],
                    FEATURE_TYPE_SEQUENCE: [5, 10, 15],
                },
                LABEL: {FEATURE_TYPE_SEQUENCE: [5], FEATURE_TYPE_SENTENCE: []},
            },
            {
                TEXT: {
                    FEATURE_TYPE_SENTENCE: [5, 2],
                    FEATURE_TYPE_SEQUENCE: [3, 5, 10],
                },
                LABEL: {FEATURE_TYPE_SEQUENCE: [2], FEATURE_TYPE_SENTENCE: []},
            },
        ),
        (
            {
                TEXT: {
                    FEATURE_TYPE_SENTENCE: [5, 2],
                    FEATURE_TYPE_SEQUENCE: [3, 5, 10],
                },
                LABEL: {FEATURE_TYPE_SEQUENCE: [2], FEATURE_TYPE_SENTENCE: []},
            },
            {
                TEXT: {
                    FEATURE_TYPE_SENTENCE: [5, 2],
                    FEATURE_TYPE_SEQUENCE: [3, 5, 10],
                },
                LABEL: {FEATURE_TYPE_SEQUENCE: [2], FEATURE_TYPE_SENTENCE: []},
            },
        ),
    ],
)
def test_adjust_sparse_layers_for_incremental_training(
    new_sparse_feature_sizes: Dict[Text, Dict[Text, List[int]]],
    old_sparse_feature_sizes: Dict[Text, Dict[Text, List[int]]],
):
    """Tests if `adjust_sparse_layers_for_incremental_training` finds and updates
    every `DenseForSparse` layer that has its sparse feature sizes increased."""

    def init_sparse_to_dense_layer(
        attribute, feature_type, input_size, output_size, reg_lambda
    ):
        kernel_initializer = tf.constant_initializer(
            np.random.random((input_size, output_size))
        )
        layer = layers.DenseForSparse(
            name=f"sparse_to_dense.{attribute}_{feature_type}",
            kernel_initializer=kernel_initializer,
            reg_lambda=reg_lambda,
            units=output_size,
        )
        layer.build(input_shape=input_size)
        return layer

    units, reg_lambda = 10, 0.02
    bottom_custom_layer = RasaCustomLayer()
    layer_text_sequence = init_sparse_to_dense_layer(
        attribute=TEXT,
        feature_type=FEATURE_TYPE_SEQUENCE,
        input_size=sum(old_sparse_feature_sizes[TEXT][FEATURE_TYPE_SEQUENCE]),
        output_size=units,
        reg_lambda=reg_lambda,
    )
    bottom_tf_layers = {"other_layer": object, "sparse_to_dense": layer_text_sequence}
    bottom_custom_layer._tf_layers = bottom_tf_layers
    middle_custom_layer = RasaCustomLayer()
    layer_label_sequence = init_sparse_to_dense_layer(
        attribute=LABEL,
        feature_type=FEATURE_TYPE_SEQUENCE,
        input_size=sum(old_sparse_feature_sizes[LABEL][FEATURE_TYPE_SEQUENCE]),
        output_size=units,
        reg_lambda=reg_lambda,
    )
    middle_tf_layers = {
        "other_layer": object,
        "sparse_to_dense": layer_label_sequence,
        "another_custom_layer": bottom_custom_layer,
    }
    middle_custom_layer._tf_layers = middle_tf_layers
    top_custom_layer = RasaCustomLayer()
    layer_text_sentence = init_sparse_to_dense_layer(
        attribute=TEXT,
        feature_type=FEATURE_TYPE_SENTENCE,
        input_size=sum(old_sparse_feature_sizes[TEXT][FEATURE_TYPE_SENTENCE]),
        output_size=units,
        reg_lambda=reg_lambda,
    )
    top_tf_layers = {
        "other_layer": object,
        "sparse_to_dense": layer_text_sentence,
        "another_custom_layer": middle_custom_layer,
    }
    top_custom_layer._tf_layers = top_tf_layers

    top_custom_layer.adjust_sparse_layers_for_incremental_training(
        new_sparse_feature_sizes=new_sparse_feature_sizes,
        old_sparse_feature_sizes=old_sparse_feature_sizes,
        reg_lambda=reg_lambda,
    )
    custom_layers = [bottom_custom_layer, middle_custom_layer, top_custom_layer]
    for custom_layer in custom_layers:
        dense_layer = custom_layer._tf_layers["sparse_to_dense"]
        layer_attribute = dense_layer.get_attribute()
        layer_feature_type = dense_layer.get_feature_type()
        layer_expected_size = sum(
            new_sparse_feature_sizes[layer_attribute][layer_feature_type]
        )
        dense_layer.build(input_shape=layer_expected_size)
        layer_final_size = dense_layer.get_kernel().shape[0]
        assert layer_attribute and layer_feature_type
        assert layer_final_size == layer_expected_size