import collections

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
from caffe2.python import core, dyndep, workspace
from caffe2.quantization.server.dnnlowp_test_utils import check_quantized_results_close
from hypothesis import given


            

Reported by Pylint.

              import numpy as np
from caffe2.python import core, dyndep, workspace
from caffe2.quantization.server.dnnlowp_test_utils import check_quantized_results_close
from hypothesis import given


dyndep.InitOpsLibrary("//caffe2/caffe2/quantization/server:dnnlowp_ops")
workspace.GlobalInit(["caffe2", "--caffe2_omp_num_threads=11"])


            

Reported by Pylint.

                      **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0
        # All inputs have scale 1, so exactly represented after quantization
        min_ = -100

            

Reported by Pylint.

              

import collections

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
from caffe2.python import core, dyndep, workspace
from caffe2.quantization.server.dnnlowp_test_utils import check_quantized_results_close

            

Reported by Pylint.

              workspace.GlobalInit(["caffe2", "--caffe2_omp_num_threads=11"])


class DNNLowPElementwiseLinearOpTest(hu.HypothesisTestCase):
    @given(
        N=st.integers(32, 256),
        D=st.integers(32, 256),
        empty_batch=st.booleans(),
        in_quantized=st.booleans(),

            

Reported by Pylint.

                      in_quantized=st.booleans(),
        out_quantized=st.booleans(),
        **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0

            

Reported by Pylint.

                      in_quantized=st.booleans(),
        out_quantized=st.booleans(),
        **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0

            

Reported by Pylint.

                      in_quantized=st.booleans(),
        out_quantized=st.booleans(),
        **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0

            

Reported by Pylint.

                      in_quantized=st.booleans(),
        out_quantized=st.booleans(),
        **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0

            

Reported by Pylint.

                      in_quantized=st.booleans(),
        out_quantized=st.booleans(),
        **hu.gcs_cpu_only
    )
    def test_dnnlowp_elementwise_linear_int(
        self, N, D, empty_batch, in_quantized, out_quantized, gc, dc
    ):
        if empty_batch:
            N = 0

            

Reported by Pylint.

              import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial

import hypothesis.extra.numpy as hnp
import hypothesis.strategies as st
import numpy as np


@st.composite

            

Reported by Pylint.

              import caffe2.python.serialized_test.serialized_test_util as serial

import hypothesis.extra.numpy as hnp
import hypothesis.strategies as st
import numpy as np


@st.composite
def id_list_batch(draw):

            

Reported by Pylint.

              




from caffe2.python import core
import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial


            

Reported by Pylint.

              

@st.composite
def id_list_batch(draw):
    num_inputs = draw(st.integers(1, 3))
    batch_size = draw(st.integers(5, 10))
    values_dtype = draw(st.sampled_from([np.int32, np.int64]))
    inputs = []
    for _ in range(num_inputs):

            

Reported by Pylint.

                  return inputs


def merge_id_lists_ref(*args):
    n = len(args)
    assert n > 0
    assert n % 2 == 0
    batch_size = len(args[0])
    num_inputs = int(n / 2)

            

Reported by Pylint.

              

def merge_id_lists_ref(*args):
    n = len(args)
    assert n > 0
    assert n % 2 == 0
    batch_size = len(args[0])
    num_inputs = int(n / 2)
    lengths = np.array([np.insert(args[2 * i], 0, 0)

            

Reported by Pylint.

              
def merge_id_lists_ref(*args):
    n = len(args)
    assert n > 0
    assert n % 2 == 0
    batch_size = len(args[0])
    num_inputs = int(n / 2)
    lengths = np.array([np.insert(args[2 * i], 0, 0)
                        for i in range(num_inputs)])

            

Reported by Bandit.

              def merge_id_lists_ref(*args):
    n = len(args)
    assert n > 0
    assert n % 2 == 0
    batch_size = len(args[0])
    num_inputs = int(n / 2)
    lengths = np.array([np.insert(args[2 * i], 0, 0)
                        for i in range(num_inputs)])
    values = [args[2 * i + 1] for i in range(num_inputs)]

            

Reported by Bandit.

                  values = [args[2 * i + 1] for i in range(num_inputs)]
    offsets = [np.cumsum(lengths[j]) for j in range(num_inputs)]

    def merge_arrays(vs, offs, j):
        concat = np.concatenate([vs[i][offs[i][j]:offs[i][j + 1]]
                                for i in range(num_inputs)])
        return np.sort(np.unique(concat))

    merged = [merge_arrays(values, offsets, j) for j in range(batch_size)]

            

Reported by Pylint.

                  return merged_lengths, merged_values


class TestMergeIdListsOp(serial.SerializedTestCase):
    def test_merge_id_lists_ref(self):
        # Verify that the reference implementation is correct!
        lengths_0 = np.array([3, 0, 4], dtype=np.int32)
        values_0 = np.array([1, 5, 6, 2, 4, 5, 6], dtype=np.int64)
        lengths_1 = np.array([3, 2, 1], dtype=np.int32)

            

Reported by Pylint.

              

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


            

Reported by Pylint.

              import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


class TestEnsureClipped(hu.HypothesisTestCase):
    @given(
        X=hu.arrays(dims=[5, 10], elements=hu.floats(min_value=-1.0, max_value=1.0)),

            

Reported by Pylint.

                      indices=hu.arrays(dims=[5], elements=st.booleans()),
        **hu.gcs_cpu_only
    )
    def test_ensure_clipped(self, X, in_place, sparse, indices, gc, dc):
        if (not in_place) and sparse:
            return
        param = X.astype(np.float32)
        m, n = param.shape
        indices = np.array(np.nonzero(indices)[0], dtype=np.int64)

            

Reported by Pylint.

                      indices=hu.arrays(dims=[5], elements=st.booleans()),
        **hu.gcs_cpu_only
    )
    def test_ensure_clipped(self, X, in_place, sparse, indices, gc, dc):
        if (not in_place) and sparse:
            return
        param = X.astype(np.float32)
        m, n = param.shape
        indices = np.array(np.nonzero(indices)[0], dtype=np.int64)

            

Reported by Pylint.

                      workspace.FeedBlob("indices", indices)
        workspace.FeedBlob("grad", grad)
        workspace.FeedBlob("param", param)
        input = ["param", "indices", "grad"] if sparse else ["param"]
        output = "param" if in_place else "output"
        op = core.CreateOperator("EnsureClipped", input, output, min=0.0)
        workspace.RunOperatorOnce(op)

        def ref():

            

Reported by Pylint.

              

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


            

Reported by Pylint.

              from hypothesis import given


class TestEnsureClipped(hu.HypothesisTestCase):
    @given(
        X=hu.arrays(dims=[5, 10], elements=hu.floats(min_value=-1.0, max_value=1.0)),
        in_place=st.booleans(),
        sparse=st.booleans(),
        indices=hu.arrays(dims=[5], elements=st.booleans()),

            

Reported by Pylint.

                      sparse=st.booleans(),
        indices=hu.arrays(dims=[5], elements=st.booleans()),
        **hu.gcs_cpu_only
    )
    def test_ensure_clipped(self, X, in_place, sparse, indices, gc, dc):
        if (not in_place) and sparse:
            return
        param = X.astype(np.float32)
        m, n = param.shape

            

Reported by Pylint.

                      sparse=st.booleans(),
        indices=hu.arrays(dims=[5], elements=st.booleans()),
        **hu.gcs_cpu_only
    )
    def test_ensure_clipped(self, X, in_place, sparse, indices, gc, dc):
        if (not in_place) and sparse:
            return
        param = X.astype(np.float32)
        m, n = param.shape

            

Reported by Pylint.

                      sparse=st.booleans(),
        indices=hu.arrays(dims=[5], elements=st.booleans()),
        **hu.gcs_cpu_only
    )
    def test_ensure_clipped(self, X, in_place, sparse, indices, gc, dc):
        if (not in_place) and sparse:
            return
        param = X.astype(np.float32)
        m, n = param.shape

            

Reported by Pylint.

              

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
from caffe2.python import core, dyndep, workspace
from hypothesis import given, settings



            

Reported by Pylint.

              import hypothesis.strategies as st
import numpy as np
from caffe2.python import core, dyndep, workspace
from hypothesis import given, settings


dyndep.InitOpsLibrary("//caffe2/caffe2/quantization/server:dnnlowp_ops")
workspace.GlobalInit(["caffe2", "--caffe2_omp_num_threads=11"])


            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
from caffe2.python import core, dyndep, workspace
from hypothesis import given, settings



            

Reported by Pylint.

              workspace.GlobalInit(["caffe2", "--caffe2_omp_num_threads=11"])


class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)

            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              class DNNLowPBatchPermutationOpTest(hu.HypothesisTestCase):
    @given(N=st.integers(min_value=1, max_value=100), **hu.gcs_cpu_only)
    @settings(max_examples=10, deadline=None)
    def test_batch_permutation(self, N, gc, dc):
        X = np.round(np.random.rand(N, 10, 20, 3) * 255).astype(np.float32)
        indices = np.arange(N).astype(np.int32)
        np.random.shuffle(indices)

        quantize = core.CreateOperator("Quantize", ["X"], ["X_q"], engine="DNNLOWP")

            

Reported by Pylint.

              #!/usr/bin/env python3

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


            

Reported by Pylint.

              import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


class TestUnsafeCoalesceOp(hu.HypothesisTestCase):
    @given(
        n=st.integers(1, 5),

            

Reported by Pylint.

                      shape=st.lists(st.integers(0, 5), min_size=1, max_size=3),
        **hu.gcs
    )
    def test_unsafe_coalesce_op(self, n, shape, dc, gc):
        workspace.ResetWorkspace()
        test_inputs = [(100 * np.random.random(shape)).astype(np.float32) for _ in range(n)]
        test_input_blobs = ["x_{}".format(i) for i in range(n)]

        coalesce_op = core.CreateOperator(

            

Reported by Pylint.

                      )

        def reference_func(*args):
            self.assertEquals(len(args), n)
            return list(args) + [np.concatenate([x.flatten() for x in args])]

        self.assertReferenceChecks(gc, coalesce_op, test_inputs, reference_func)

    @given(

            

Reported by Pylint.

                      seed=st.integers(0, 65535),
        **hu.gcs
    )
    def test_unsafe_coalesce_op_blob_sharing(self, n, shape, seed, dc, gc):
        workspace.ResetWorkspace()
        # Can make debugging of the test more predictable
        np.random.seed(seed)
        test_inputs = [(np.random.random(shape)).astype(np.float32) for _ in range(n)]
        test_input_blobs = ["x_{}".format(i) for i in range(n)]

            

Reported by Pylint.

              #!/usr/bin/env python3

import caffe2.python.hypothesis_test_util as hu
import hypothesis.strategies as st
import numpy as np
import numpy.testing as npt
from caffe2.python import core, workspace
from hypothesis import given


            

Reported by Pylint.

              from hypothesis import given


class TestUnsafeCoalesceOp(hu.HypothesisTestCase):
    @given(
        n=st.integers(1, 5),
        shape=st.lists(st.integers(0, 5), min_size=1, max_size=3),
        **hu.gcs
    )

            

Reported by Pylint.

                      n=st.integers(1, 5),
        shape=st.lists(st.integers(0, 5), min_size=1, max_size=3),
        **hu.gcs
    )
    def test_unsafe_coalesce_op(self, n, shape, dc, gc):
        workspace.ResetWorkspace()
        test_inputs = [(100 * np.random.random(shape)).astype(np.float32) for _ in range(n)]
        test_input_blobs = ["x_{}".format(i) for i in range(n)]


            

Reported by Pylint.

                      n=st.integers(1, 5),
        shape=st.lists(st.integers(0, 5), min_size=1, max_size=3),
        **hu.gcs
    )
    def test_unsafe_coalesce_op(self, n, shape, dc, gc):
        workspace.ResetWorkspace()
        test_inputs = [(100 * np.random.random(shape)).astype(np.float32) for _ in range(n)]
        test_input_blobs = ["x_{}".format(i) for i in range(n)]


            

Reported by Pylint.

                      n=st.integers(1, 5),
        shape=st.lists(st.integers(0, 5), min_size=1, max_size=3),
        **hu.gcs
    )
    def test_unsafe_coalesce_op(self, n, shape, dc, gc):
        workspace.ResetWorkspace()
        test_inputs = [(100 * np.random.random(shape)).astype(np.float32) for _ in range(n)]
        test_input_blobs = ["x_{}".format(i) for i in range(n)]


            

Reported by Pylint.

              from caffe2.python import core
import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial
import hypothesis.strategies as st
import numpy as np


class TestElementwiseLinearOp(serial.SerializedTestCase):


            

Reported by Pylint.

              




from caffe2.python import core
import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial
import hypothesis.strategies as st

            

Reported by Pylint.

              import numpy as np


class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)

            

Reported by Pylint.

              class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)


            

Reported by Pylint.

              class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)


            

Reported by Pylint.

              class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)


            

Reported by Pylint.

              class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)


            

Reported by Pylint.

              class TestElementwiseLinearOp(serial.SerializedTestCase):

    @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)


            

Reported by Pylint.

                  @serial.given(n=st.integers(2, 100), d=st.integers(2, 10), **hu.gcs)
    # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)

        def ref_op(X, a, b):
            d = a.shape[0]

            

Reported by Pylint.

                  # @given(n=st.integers(2, 50), d=st.integers(2, 50), **hu.gcs_cpu_only)
    def test(self, n, d, gc, dc):
        X = np.random.rand(n, d).astype(np.float32)
        a = np.random.rand(d).astype(np.float32)
        b = np.random.rand(d).astype(np.float32)

        def ref_op(X, a, b):
            d = a.shape[0]
            return [np.multiply(X, a.reshape(1, d)) + b.reshape(1, d)]

            

Reported by Pylint.

              from abc import ABC, abstractmethod
from typing import Any, Callable, Optional, Tuple, cast

from torch.distributed.elastic.rendezvous import RendezvousStateError
from torch.distributed.elastic.rendezvous.dynamic_rendezvous import RendezvousBackend, Token


class RendezvousBackendTestMixin(ABC):
    _backend: RendezvousBackend

            

Reported by Pylint.

              from typing import Any, Callable, Optional, Tuple, cast

from torch.distributed.elastic.rendezvous import RendezvousStateError
from torch.distributed.elastic.rendezvous.dynamic_rendezvous import RendezvousBackend, Token


class RendezvousBackendTestMixin(ABC):
    _backend: RendezvousBackend


            

Reported by Pylint.

                  @abstractmethod
    def _corrupt_state(self) -> None:
        """Corrupts the state stored in the backend."""
        pass

    def _set_state(self, state: bytes, token: Optional[Any] = None) -> Tuple[bytes, Token, bool]:
        result = self._backend.set_state(state, token)

        self.assertIsNotNone(result)

            

Reported by Pylint.

                      self.assertTrue(has_set)

    def test_set_state_sets_backend_state_if_token_is_current(self) -> None:
        state1, token1, has_set1 = self._set_state(b"x")

        state2, token2, has_set2 = self._set_state(b"y", token1)

        self.assertEqual(b"y", state2)
        self.assertNotEqual(token1, token2)

            

Reported by Pylint.

                      self.assertTrue(has_set2)

    def test_set_state_returns_current_backend_state_if_token_is_old(self) -> None:
        state1, token1, _ = self._set_state(b"x")

        state2, token2, _ = self._set_state(b"y", token1)

        state3, token3, has_set = self._set_state(b"z", token1)


            

Reported by Pylint.

              # Copyright (c) Facebook, Inc. and its affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

from abc import ABC, abstractmethod
from typing import Any, Callable, Optional, Tuple, cast


            

Reported by Pylint.

              from torch.distributed.elastic.rendezvous.dynamic_rendezvous import RendezvousBackend, Token


class RendezvousBackendTestMixin(ABC):
    _backend: RendezvousBackend

    # Type hints
    assertEqual: Callable
    assertNotEqual: Callable

            

Reported by Pylint.

              
        return cast(Tuple[bytes, Token, bool], result)

    def test_get_state_returns_backend_state(self) -> None:
        self._backend.set_state(b"x")

        result = self._backend.get_state()

        self.assertIsNotNone(result)

            

Reported by Pylint.

                      self.assertEqual(b"x", state)
        self.assertIsNotNone(token)

    def test_get_state_returns_none_if_backend_state_does_not_exist(self) -> None:
        result = self._backend.get_state()

        self.assertIsNone(result)

    def test_get_state_raises_error_if_backend_state_is_corrupt(self) -> None:

            

Reported by Pylint.

              
        self.assertIsNone(result)

    def test_get_state_raises_error_if_backend_state_is_corrupt(self) -> None:
        self._corrupt_state()

        with self.assertRaises(RendezvousStateError):
            self._backend.get_state()


            

Reported by Pylint.

              #!/usr/bin/env python3

import hypothesis.strategies as st
import numpy as np
import torch
from caffe2.python import core
from caffe2.python.test_util import TestCase
from hypothesis import given, settings
from torch import nn

            

Reported by Pylint.

              import torch
from caffe2.python import core
from caffe2.python.test_util import TestCase
from hypothesis import given, settings
from torch import nn


class TestC2LSTM(TestCase):
    @given(

            

Reported by Pylint.

                      else:
            inputs = np.random.randn(seq_lens, bsz, emb_lens).astype(np.float32)

        py_results = py_lstm(torch.from_numpy(inputs))
        lstm_in = [
            torch.from_numpy(inputs),
            torch.from_numpy(hx),
            torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

            

Reported by Pylint.

              
        py_results = py_lstm(torch.from_numpy(inputs))
        lstm_in = [
            torch.from_numpy(inputs),
            torch.from_numpy(hx),
            torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

        c2_results = torch.ops._caffe2.InferenceLSTM(

            

Reported by Pylint.

                      py_results = py_lstm(torch.from_numpy(inputs))
        lstm_in = [
            torch.from_numpy(inputs),
            torch.from_numpy(hx),
            torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

        c2_results = torch.ops._caffe2.InferenceLSTM(
            lstm_in, num_layers, has_biases, batch_first, is_bidirectional

            

Reported by Pylint.

                      lstm_in = [
            torch.from_numpy(inputs),
            torch.from_numpy(hx),
            torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

        c2_results = torch.ops._caffe2.InferenceLSTM(
            lstm_in, num_layers, has_biases, batch_first, is_bidirectional
        )

            

Reported by Pylint.

                      is_bidirectional,
        batch_first,
    ):
        net = core.Net("test_net")
        num_directions = 2 if is_bidirectional else 1
        py_lstm = nn.LSTM(
            emb_lens,
            hidden_size,
            batch_first=batch_first,

            

Reported by Pylint.

                          torch.from_numpy(inputs),
            torch.from_numpy(hx),
            torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

        c2_results = torch.ops._caffe2.InferenceLSTM(
            lstm_in, num_layers, has_biases, batch_first, is_bidirectional
        )


            

Reported by Pylint.

                          torch.from_numpy(hx),
        ] + [param.detach() for param in py_lstm._flat_weights]

        c2_results = torch.ops._caffe2.InferenceLSTM(
            lstm_in, num_layers, has_biases, batch_first, is_bidirectional
        )

        np.testing.assert_array_almost_equal(
            py_results[0].detach().numpy(), c2_results[0].detach().numpy()

            

Reported by Pylint.

              #!/usr/bin/env python3

import hypothesis.strategies as st
import numpy as np
import torch
from caffe2.python import core
from caffe2.python.test_util import TestCase
from hypothesis import given, settings
from torch import nn

            

Reported by Pylint.

              from caffe2.python import core, workspace
import caffe2.python.hypothesis_test_util as hu

from hypothesis import given
import numpy as np


class TestCastOp(hu.HypothesisTestCase):


            

Reported by Pylint.

                      self.assertGradientChecks(gc, op, [data], 0, [0])

    @given(data=hu.tensor(dtype=np.int32), **hu.gcs_cpu_only)
    def test_cast_int_to_string(self, data, gc, dc):
        op = core.CreateOperator(
            'Cast', 'data', 'data_cast', to=core.DataType.STRING)

        def ref(data):
            ret = data.astype(dtype=np.str)

            

Reported by Pylint.

              





from caffe2.python import core, workspace
import caffe2.python.hypothesis_test_util as hu


            

Reported by Pylint.

              import numpy as np


class TestCastOp(hu.HypothesisTestCase):

    @given(**hu.gcs)
    def test_cast_int_float(self, gc, dc):
        data = np.random.rand(5, 5).astype(np.int32)
        # from int to float

            

Reported by Pylint.

              class TestCastOp(hu.HypothesisTestCase):

    @given(**hu.gcs)
    def test_cast_int_float(self, gc, dc):
        data = np.random.rand(5, 5).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0

            

Reported by Pylint.

              class TestCastOp(hu.HypothesisTestCase):

    @given(**hu.gcs)
    def test_cast_int_float(self, gc, dc):
        data = np.random.rand(5, 5).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0

            

Reported by Pylint.

              class TestCastOp(hu.HypothesisTestCase):

    @given(**hu.gcs)
    def test_cast_int_float(self, gc, dc):
        data = np.random.rand(5, 5).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0

            

Reported by Pylint.

                  def test_cast_int_float(self, gc, dc):
        data = np.random.rand(5, 5).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0
        self.assertGradientChecks(gc, op, [data], 0, [0])

    @given(**hu.gcs)

            

Reported by Pylint.

                      self.assertGradientChecks(gc, op, [data], 0, [0])

    @given(**hu.gcs)
    def test_cast_int_float_empty(self, gc, dc):
        data = np.random.rand(0).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0

            

Reported by Pylint.

                      self.assertGradientChecks(gc, op, [data], 0, [0])

    @given(**hu.gcs)
    def test_cast_int_float_empty(self, gc, dc):
        data = np.random.rand(0).astype(np.int32)
        # from int to float
        op = core.CreateOperator('Cast', 'data', 'data_cast', to=1, from_type=2)
        self.assertDeviceChecks(dc, op, [data], [0])
        # This is actually 0

            

Reported by Pylint.

              



from caffe2.python import core, workspace
from caffe2.python.test_util import TestCase

import numpy as np


            

Reported by Pylint.

              import numpy as np


class TestSparseToDenseMask(TestCase):

    def test_sparse_to_dense_mask_float(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],

            

Reported by Pylint.

              
class TestSparseToDenseMask(TestCase):

    def test_sparse_to_dense_mask_float(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 6])

            

Reported by Pylint.

              class TestSparseToDenseMask(TestCase):

    def test_sparse_to_dense_mask_float(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 6])
        workspace.FeedBlob(

            

Reported by Pylint.

                      self.assertEqual(output.shape, expected.shape)
        np.testing.assert_array_equal(output, expected)

    def test_sparse_to_dense_mask_invalid_inputs(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2],

            

Reported by Pylint.

                      np.testing.assert_array_equal(output, expected)

    def test_sparse_to_dense_mask_invalid_inputs(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2],
            max_skipped_indices=3)

            

Reported by Pylint.

                      with self.assertRaises(RuntimeError):
            workspace.RunOperatorMultiple(op, 3)

    def test_sparse_to_dense_mask_subtensor(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 888, 6])

            

Reported by Pylint.

                          workspace.RunOperatorMultiple(op, 3)

    def test_sparse_to_dense_mask_subtensor(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 888, 6])
        workspace.FeedBlob(

            

Reported by Pylint.

                      self.assertEqual(output.shape, expected.shape)
        np.testing.assert_array_equal(output, expected)

    def test_sparse_to_dense_mask_string(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 6])

            

Reported by Pylint.

                      np.testing.assert_array_equal(output, expected)

    def test_sparse_to_dense_mask_string(self):
        op = core.CreateOperator(
            'SparseToDenseMask',
            ['indices', 'values', 'default', 'lengths'],
            ['output'],
            mask=[999999999, 2, 6])
        workspace.FeedBlob(

            

Reported by Pylint.