<commit_before>import pytest

import numpy as np
import syft as sy
from syft import dependency_check

if dependency_check.tfe_available:  # pragma: no cover
    import tensorflow as tf
    import tf_encrypted as tfe


@pytest.mark.skipif(not dependency_check.tfe_available, reason="tf_encrypted not installed")
def test_instantiate_tfe_layer():  # pragma: no cover
    """tests tfe layer by running a constant 4*5 input matrix on a network
        with one constant tensor of weights, using tf.keras then tfe and comparing results"""

    from syft.frameworks.keras.model.sequential import _instantiate_tfe_layer

    hook = sy.KerasHook(tf.keras)

    # creates input and weights constant tensors
    input_shape = [4, 5]
    input_data = np.ones(input_shape)
    kernel = np.random.normal(size=[5, 5])
    initializer = tf.keras.initializers.Constant(kernel)

    # creates a network with 4*5 input shape, 5 output units, and 5*5 weights matrix
    d_tf = tf.keras.layers.Dense(
        5, kernel_initializer=initializer, batch_input_shape=input_shape, use_bias=True
    )

    # runs the model using tf.keras
    with tf.Session() as sess:
        x = tf.Variable(input_data, dtype=tf.float32)
        y = d_tf(x)
        sess.run(tf.global_variables_initializer())
        expected = sess.run(y)

    stored_keras_weights = {d_tf.name: d_tf.get_weights()}

    # runs the model using tfe
    with tf.Graph().as_default():
        p_x = tfe.define_private_variable(input_data)
        d_tfe = _instantiate_tfe_layer(d_tf, stored_keras_weights)

        out = d_tfe(p_x)

        with tfe.Session() as sess:
            sess.run(tf.global_variables_initializer())

            actual = sess.run(out.reveal())

    # compares results and raises error if not equal up to 0.001
    np.testing.assert_allclose(actual, expected, rtol=0.001)


@pytest.mark.skipif(not dependency_check.tfe_available, reason="tf_encrypted not installed")
def test_share():  # pragma: no cover
    """tests tfe federated learning functionality by running a constant input on same model using tf.keras
        then tfe on remote workers and comparing the outputs of both cases """
    from tensorflow.keras import Sequential
    from tensorflow.keras.layers import Dense

    hook = sy.KerasHook(tf.keras)

    # creates input and weights constant tensors
    input_shape = (4, 5)<commit_msg>Adjust comparison thresholds in `test_instantiate_tfe_layer` (#3358)<commit_after>import pytest

import numpy as np
import syft as sy
from syft import dependency_check

if dependency_check.tfe_available:  # pragma: no cover
    import tensorflow as tf
    import tf_encrypted as tfe


@pytest.mark.skipif(not dependency_check.tfe_available, reason="tf_encrypted not installed")
def test_instantiate_tfe_layer():  # pragma: no cover
    """tests tfe layer by running a constant 4*5 input matrix on a network
        with one constant tensor of weights, using tf.keras then tfe and comparing results"""

    from syft.frameworks.keras.model.sequential import _instantiate_tfe_layer

    hook = sy.KerasHook(tf.keras)

    # creates input and weights constant tensors
    input_shape = [4, 5]
    input_data = np.ones(input_shape)
    kernel = np.random.normal(size=[5, 5])
    initializer = tf.keras.initializers.Constant(kernel)

    # creates a network with 4*5 input shape, 5 output units, and 5*5 weights matrix
    d_tf = tf.keras.layers.Dense(
        5, kernel_initializer=initializer, batch_input_shape=input_shape, use_bias=True
    )

    # runs the model using tf.keras
    with tf.Session() as sess:
        x = tf.Variable(input_data, dtype=tf.float32)
        y = d_tf(x)
        sess.run(tf.global_variables_initializer())
        expected = sess.run(y)

    stored_keras_weights = {d_tf.name: d_tf.get_weights()}

    # runs the model using tfe
    with tf.Graph().as_default():
        p_x = tfe.define_private_variable(input_data)
        d_tfe = _instantiate_tfe_layer(d_tf, stored_keras_weights)

        out = d_tfe(p_x)

        with tfe.Session() as sess:
            sess.run(tf.global_variables_initializer())

            actual = sess.run(out.reveal())

    # compares results and raises error if not equal up to 0.001
    np.testing.assert_allclose(actual, expected, rtol=0.005, atol=0.001)


@pytest.mark.skipif(not dependency_check.tfe_available, reason="tf_encrypted not installed")
def test_share():  # pragma: no cover
    """tests tfe federated learning functionality by running a constant input on same model using tf.keras
        then tfe on remote workers and comparing the outputs of both cases """
    from tensorflow.keras import Sequential
    from tensorflow.keras.layers import Dense

    hook = sy.KerasHook(tf.keras)

    # creates input and weights constant tensors
    input_shape = (4, 5)
{'fraction_non_alphanumeric': 0.08328128253175099, 'fraction_numerical': 0.011659379554445138, 'mean_word_length': 3.49812734082397, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7033073', 'n_tokens_mistral': 1572, 'n_tokens_neox': 1493, 'n_words': 437}