def fuse_conv_bn_weights(conv_w, conv_b, bn_rm, bn_rv, bn_eps, bn_w, bn_b):
    if conv_b is None:
        conv_b = torch.zeros_like(bn_rm)
    if bn_w is None:
        bn_w = torch.ones_like(bn_rm)
    if bn_b is None:
        bn_b = torch.zeros_like(bn_rm)
    bn_var_rsqrt = torch.rsqrt(bn_rv + bn_eps)

            

Reported by Pylint.

                  if conv_b is None:
        conv_b = torch.zeros_like(bn_rm)
    if bn_w is None:
        bn_w = torch.ones_like(bn_rm)
    if bn_b is None:
        bn_b = torch.zeros_like(bn_rm)
    bn_var_rsqrt = torch.rsqrt(bn_rv + bn_eps)

    conv_w = conv_w * (bn_w * bn_var_rsqrt).reshape([-1] + [1] * (len(conv_w.shape) - 1))

            

Reported by Pylint.

                  if bn_w is None:
        bn_w = torch.ones_like(bn_rm)
    if bn_b is None:
        bn_b = torch.zeros_like(bn_rm)
    bn_var_rsqrt = torch.rsqrt(bn_rv + bn_eps)

    conv_w = conv_w * (bn_w * bn_var_rsqrt).reshape([-1] + [1] * (len(conv_w.shape) - 1))
    conv_b = (conv_b - bn_rm) * bn_var_rsqrt * bn_w + bn_b


            

Reported by Pylint.

                      bn_w = torch.ones_like(bn_rm)
    if bn_b is None:
        bn_b = torch.zeros_like(bn_rm)
    bn_var_rsqrt = torch.rsqrt(bn_rv + bn_eps)

    conv_w = conv_w * (bn_w * bn_var_rsqrt).reshape([-1] + [1] * (len(conv_w.shape) - 1))
    conv_b = (conv_b - bn_rm) * bn_var_rsqrt * bn_w + bn_b

    return torch.nn.Parameter(conv_w), torch.nn.Parameter(conv_b)

            

Reported by Pylint.

              
def fuse_linear_bn_weights(linear_w, linear_b, bn_rm, bn_rv, bn_eps, bn_w, bn_b):
    if linear_b is None:
        linear_b = torch.zeros_like(bn_rm)
    bn_scale = bn_w * torch.rsqrt(bn_rv + bn_eps)

    fused_w = linear_w * bn_scale.unsqueeze(-1)
    fused_b = (linear_b - bn_rm) * bn_scale + bn_b


            

Reported by Pylint.

              def fuse_linear_bn_weights(linear_w, linear_b, bn_rm, bn_rv, bn_eps, bn_w, bn_b):
    if linear_b is None:
        linear_b = torch.zeros_like(bn_rm)
    bn_scale = bn_w * torch.rsqrt(bn_rv + bn_eps)

    fused_w = linear_w * bn_scale.unsqueeze(-1)
    fused_b = (linear_b - bn_rm) * bn_scale + bn_b

    return torch.nn.Parameter(fused_w), torch.nn.Parameter(fused_b)

            

Reported by Pylint.

              

import copy
import torch

def fuse_conv_bn_eval(conv, bn):
    assert(not (conv.training or bn.training)), "Fusion only for eval!"
    fused_conv = copy.deepcopy(conv)


            

Reported by Pylint.

              import copy
import torch

def fuse_conv_bn_eval(conv, bn):
    assert(not (conv.training or bn.training)), "Fusion only for eval!"
    fused_conv = copy.deepcopy(conv)

    fused_conv.weight, fused_conv.bias = \
        fuse_conv_bn_weights(fused_conv.weight, fused_conv.bias,

            

Reported by Pylint.

              import copy
import torch

def fuse_conv_bn_eval(conv, bn):
    assert(not (conv.training or bn.training)), "Fusion only for eval!"
    fused_conv = copy.deepcopy(conv)

    fused_conv.weight, fused_conv.bias = \
        fuse_conv_bn_weights(fused_conv.weight, fused_conv.bias,

            

Reported by Pylint.

              import torch

def fuse_conv_bn_eval(conv, bn):
    assert(not (conv.training or bn.training)), "Fusion only for eval!"
    fused_conv = copy.deepcopy(conv)

    fused_conv.weight, fused_conv.bias = \
        fuse_conv_bn_weights(fused_conv.weight, fused_conv.bias,
                             bn.running_mean, bn.running_var, bn.eps, bn.weight, bn.bias)

            

Reported by Bandit.

              
import torch
from torch import nn
import torch.nn.utils.parametrize as parametrize

from torch.testing._internal.jit_utils import JitTestCase


if __name__ == '__main__':

            

Reported by Pylint.

              
import torch
from torch import nn
import torch.nn.utils.parametrize as parametrize

from torch.testing._internal.jit_utils import JitTestCase


if __name__ == '__main__':

            

Reported by Pylint.

              
import torch
from torch import nn
import torch.nn.utils.parametrize as parametrize

from torch.testing._internal.jit_utils import JitTestCase


if __name__ == '__main__':

            

Reported by Pylint.

              from torch import nn
import torch.nn.utils.parametrize as parametrize

from torch.testing._internal.jit_utils import JitTestCase


if __name__ == '__main__':
    raise RuntimeError("This test file is not meant to be run directly, use:\n\n"
                       "\tpython test/test_jit.py TESTNAME\n\n"

            

Reported by Pylint.

                              traced_model = torch.jit.trace_module(model, {'forward': x})

    def test_scriptable(self):
        # TODO: Need to fix the scripting in parametrizations
        #       Currently, all the tests below will throw UnsupportedNodeError
        model = nn.Linear(5, 5)
        parametrize.register_parametrization(model, "weight", self.Symmetric())

        x = torch.randn(3, 5)

            

Reported by Pylint.

              
import torch
from torch import nn
import torch.nn.utils.parametrize as parametrize

from torch.testing._internal.jit_utils import JitTestCase


if __name__ == '__main__':

            

Reported by Pylint.

                                     "\tpython test/test_jit.py TESTNAME\n\n"
                       "instead.")

class TestParametrization(JitTestCase):
    # Define some parametrization
    class Symmetric(nn.Module):
        def forward(self, X):
            return X.triu() + X.triu(1).transpose(-1, -2)


            

Reported by Pylint.

              
class TestParametrization(JitTestCase):
    # Define some parametrization
    class Symmetric(nn.Module):
        def forward(self, X):
            return X.triu() + X.triu(1).transpose(-1, -2)

    def test_traceable(self):
        r"""Test the jit scripting and tracing of a parametrized model."""

            

Reported by Pylint.

              
class TestParametrization(JitTestCase):
    # Define some parametrization
    class Symmetric(nn.Module):
        def forward(self, X):
            return X.triu() + X.triu(1).transpose(-1, -2)

    def test_traceable(self):
        r"""Test the jit scripting and tracing of a parametrized model."""

            

Reported by Pylint.

              class TestParametrization(JitTestCase):
    # Define some parametrization
    class Symmetric(nn.Module):
        def forward(self, X):
            return X.triu() + X.triu(1).transpose(-1, -2)

    def test_traceable(self):
        r"""Test the jit scripting and tracing of a parametrized model."""
        model = nn.Linear(5, 5)

            

Reported by Pylint.

              

class TestQuantile(hu.HypothesisTestCase):
    def _test_quantile(self, inputs, quantile, abs, tol):
        net = core.Net("test_net")
        net.Proto().type = "dag"
        input_tensors = []
        for i, input in enumerate(inputs):
            workspace.FeedBlob("t_{}".format(i), input)

            

Reported by Pylint.

                      net = core.Net("test_net")
        net.Proto().type = "dag"
        input_tensors = []
        for i, input in enumerate(inputs):
            workspace.FeedBlob("t_{}".format(i), input)
            input_tensors.append("t_{}".format(i))
        net.Quantile(
            input_tensors, ["quantile_value"], quantile=quantile, abs=abs, tol=tol
        )

            

Reported by Pylint.

                  def test_quantile_1(self):
        inputs = []
        num_tensors = 5
        for i in range(num_tensors):
            dim = np.random.randint(5, 100)
            inputs.append(np.random.rand(dim))
        self._test_quantile(inputs=inputs, quantile=0.2, abs=1, tol=1e-4)

    def test_quantile_2(self):

            

Reported by Pylint.

                  def test_quantile_2(self):
        inputs = []
        num_tensors = 5
        for i in range(num_tensors):
            dim = np.random.randint(5, 100)
            inputs.append(np.random.rand(dim))
        self._test_quantile(inputs=inputs, quantile=1e-6, abs=0, tol=1e-3)

    def test_quantile_3(self):

            

Reported by Pylint.

                  def test_quantile_3(self):
        inputs = []
        num_tensors = 5
        for i in range(num_tensors):
            dim1 = np.random.randint(5, 100)
            dim2 = np.random.randint(5, 100)
            inputs.append(np.random.rand(dim1, dim2))
        self._test_quantile(inputs=inputs, quantile=1 - 1e-6, abs=1, tol=1e-5)


            

Reported by Pylint.

                  def test_quantile_4(self):
        inputs = []
        num_tensors = 5
        for i in range(num_tensors):
            dim1 = np.random.randint(5, 100)
            dim2 = np.random.randint(5, 100)
            inputs.append(np.random.rand(dim1, dim2))
            inputs.append(np.random.rand(dim1))
        self._test_quantile(inputs=inputs, quantile=0.168, abs=1, tol=1e-4)

            

Reported by Pylint.

              

import unittest

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



            

Reported by Pylint.

              from caffe2.python import core, workspace


class TestQuantile(hu.HypothesisTestCase):
    def _test_quantile(self, inputs, quantile, abs, tol):
        net = core.Net("test_net")
        net.Proto().type = "dag"
        input_tensors = []
        for i, input in enumerate(inputs):

            

Reported by Pylint.

              

class TestQuantile(hu.HypothesisTestCase):
    def _test_quantile(self, inputs, quantile, abs, tol):
        net = core.Net("test_net")
        net.Proto().type = "dag"
        input_tensors = []
        for i, input in enumerate(inputs):
            workspace.FeedBlob("t_{}".format(i), input)

            

Reported by Pylint.

              

class TestQuantile(hu.HypothesisTestCase):
    def _test_quantile(self, inputs, quantile, abs, tol):
        net = core.Net("test_net")
        net.Proto().type = "dag"
        input_tensors = []
        for i, input in enumerate(inputs):
            workspace.FeedBlob("t_{}".format(i), input)

            

Reported by Pylint.

              

import numpy as np
from scipy.sparse import coo_matrix

from hypothesis import given, settings
import hypothesis.strategies as st

from caffe2.python import core

            

Reported by Pylint.

              import numpy as np
from scipy.sparse import coo_matrix

from hypothesis import given, settings
import hypothesis.strategies as st

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


            

Reported by Pylint.

              from scipy.sparse import coo_matrix

from hypothesis import given, settings
import hypothesis.strategies as st

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



            

Reported by Pylint.

                         sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)
    @settings(deadline=10000)
    def test_sparse_gradient(self, M, N, K, sparsity, gc, dc):
        X = np.random.randn(M, K).astype(np.float32)
        X[X > sparsity] = 0
        X_coo = coo_matrix(X)
        val, key, seg = X_coo.data, X_coo.col, X_coo.row


            

Reported by Pylint.

              




import numpy as np
from scipy.sparse import coo_matrix

from hypothesis import given, settings

            

Reported by Pylint.

              import caffe2.python.hypothesis_test_util as hu


class TestSparseGradient(hu.HypothesisTestCase):
    @given(M=st.integers(min_value=5, max_value=20),
           N=st.integers(min_value=5, max_value=20),
           K=st.integers(min_value=5, max_value=15),
           sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)

            

Reported by Pylint.

                         sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)
    @settings(deadline=10000)
    def test_sparse_gradient(self, M, N, K, sparsity, gc, dc):
        X = np.random.randn(M, K).astype(np.float32)
        X[X > sparsity] = 0
        X_coo = coo_matrix(X)
        val, key, seg = X_coo.data, X_coo.col, X_coo.row


            

Reported by Pylint.

                         sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)
    @settings(deadline=10000)
    def test_sparse_gradient(self, M, N, K, sparsity, gc, dc):
        X = np.random.randn(M, K).astype(np.float32)
        X[X > sparsity] = 0
        X_coo = coo_matrix(X)
        val, key, seg = X_coo.data, X_coo.col, X_coo.row


            

Reported by Pylint.

                         sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)
    @settings(deadline=10000)
    def test_sparse_gradient(self, M, N, K, sparsity, gc, dc):
        X = np.random.randn(M, K).astype(np.float32)
        X[X > sparsity] = 0
        X_coo = coo_matrix(X)
        val, key, seg = X_coo.data, X_coo.col, X_coo.row


            

Reported by Pylint.

                         sparsity=st.floats(min_value=0.1, max_value=1.0),
           **hu.gcs_cpu_only)
    @settings(deadline=10000)
    def test_sparse_gradient(self, M, N, K, sparsity, gc, dc):
        X = np.random.randn(M, K).astype(np.float32)
        X[X > sparsity] = 0
        X_coo = coo_matrix(X)
        val, key, seg = X_coo.data, X_coo.col, X_coo.row


            

Reported by Pylint.

              


import hypothesis.strategies as st

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


            

Reported by Pylint.

              import hypothesis.strategies as st

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

import numpy as np



            

Reported by Pylint.

                      out_of_vcb,
        max_categorical_limit,
        max_in_vcb_val,
        gc,
        dc,
    ):
        np.random.seed(seed)
        col_num = max(int(D / 2), 1)
        col_ids = np.random.choice(D, col_num, False).astype(np.int32)

            

Reported by Pylint.

                      max_categorical_limit,
        max_in_vcb_val,
        gc,
        dc,
    ):
        np.random.seed(seed)
        col_num = max(int(D / 2), 1)
        col_ids = np.random.choice(D, col_num, False).astype(np.int32)
        categorical_limits = np.random.randint(

            

Reported by Pylint.

              




import hypothesis.strategies as st

from caffe2.python import core, workspace
from hypothesis import given

            

Reported by Pylint.

              import numpy as np


class TestNGramOps(hu.HypothesisTestCase):
    @given(
        seed=st.integers(0, 2**32 - 1),
        N=st.integers(min_value=10, max_value=100),
        D=st.integers(min_value=2, max_value=10),
        out_of_vcb=st.floats(min_value=0, max_value=0.5),

            

Reported by Pylint.

                      max_categorical_limit=st.integers(min_value=5, max_value=20),
        max_in_vcb_val=st.integers(min_value=1000, max_value=10000),
        **hu.gcs_cpu_only
    )
    def test_ngram_from_categorical_op(
        self,
        seed,
        N,
        D,

            

Reported by Pylint.

                      max_categorical_limit=st.integers(min_value=5, max_value=20),
        max_in_vcb_val=st.integers(min_value=1000, max_value=10000),
        **hu.gcs_cpu_only
    )
    def test_ngram_from_categorical_op(
        self,
        seed,
        N,
        D,

            

Reported by Pylint.

                      max_categorical_limit=st.integers(min_value=5, max_value=20),
        max_in_vcb_val=st.integers(min_value=1000, max_value=10000),
        **hu.gcs_cpu_only
    )
    def test_ngram_from_categorical_op(
        self,
        seed,
        N,
        D,

            

Reported by Pylint.

                      max_categorical_limit=st.integers(min_value=5, max_value=20),
        max_in_vcb_val=st.integers(min_value=1000, max_value=10000),
        **hu.gcs_cpu_only
    )
    def test_ngram_from_categorical_op(
        self,
        seed,
        N,
        D,

            

Reported by Pylint.

                      Y_full_tt = workspace.FetchBlob("Y").flatten()

        assert(len(Y_fc) == len(Y_full_tt))
        self.assertAlmostEquals(np.linalg.norm(Y_fc - Y_full_tt), 0, delta=1e-3)

        # Testing TT-decomposition with minimal ranks
        sparse_tt_ranks = [1, 1, 1, 1, 1]
        sparse_cores = tt_core.matrix_to_tt(W, inp_sizes, out_sizes,
                                            sparse_tt_ranks)

            

Reported by Pylint.

                      Y_sparse_tt = workspace.FetchBlob("Y").flatten()

        assert(len(Y_fc) == len(Y_sparse_tt))
        self.assertAlmostEquals(np.linalg.norm(Y_fc - Y_sparse_tt),
                                39.974, delta=1e-3)


if __name__ == '__main__':
    unittest.main()

            

Reported by Pylint.

              




import numpy as np
import unittest

from caffe2.python import core, workspace, tt_core

            

Reported by Pylint.

              

import numpy as np
import unittest

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



            

Reported by Pylint.

              import caffe2.python.hypothesis_test_util as hu


class TestTTSVD(hu.HypothesisTestCase):
    def test_full_tt_svd(self):
        size = 256
        np.random.seed(1234)
        X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)

            

Reported by Pylint.

              

class TestTTSVD(hu.HypothesisTestCase):
    def test_full_tt_svd(self):
        size = 256
        np.random.seed(1234)
        X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)
        W = np.random.rand(size, size).astype(np.float32)

            

Reported by Pylint.

              

class TestTTSVD(hu.HypothesisTestCase):
    def test_full_tt_svd(self):
        size = 256
        np.random.seed(1234)
        X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)
        W = np.random.rand(size, size).astype(np.float32)

            

Reported by Pylint.

                  def test_full_tt_svd(self):
        size = 256
        np.random.seed(1234)
        X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)
        W = np.random.rand(size, size).astype(np.float32)
        b = np.zeros(size).astype(np.float32)
        inp_sizes = [4, 4, 4, 4]
        out_sizes = [4, 4, 4, 4]

            

Reported by Pylint.

                      np.random.seed(1234)
        X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)
        W = np.random.rand(size, size).astype(np.float32)
        b = np.zeros(size).astype(np.float32)
        inp_sizes = [4, 4, 4, 4]
        out_sizes = [4, 4, 4, 4]

        op_fc = core.CreateOperator(

            

Reported by Pylint.

                      X = np.expand_dims(
            np.random.rand(size).astype(np.float32), axis=0)
        W = np.random.rand(size, size).astype(np.float32)
        b = np.zeros(size).astype(np.float32)
        inp_sizes = [4, 4, 4, 4]
        out_sizes = [4, 4, 4, 4]

        op_fc = core.CreateOperator(
            "FC",

            

Reported by Pylint.

              
import numpy as np

from hypothesis import given, settings
import hypothesis.strategies as st

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.

              import numpy as np

from hypothesis import given, settings
import hypothesis.strategies as st

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.

              




import numpy as np

from hypothesis import given, settings
import hypothesis.strategies as st

            

Reported by Pylint.

              import caffe2.python.serialized_test.serialized_test_util as serial


class TestClip(serial.SerializedTestCase):
    @given(X=hu.tensor(min_dim=0),
           min_=st.floats(min_value=-2, max_value=0),
           max_=st.floats(min_value=0, max_value=2),
           inplace=st.booleans(),
           **hu.gcs)

            

Reported by Pylint.

                         inplace=st.booleans(),
           **hu.gcs)
    @settings(deadline=10000)
    def test_clip(self, X, min_, max_, inplace, gc, dc):
        # go away from the origin point to avoid kink problems
        if np.isscalar(X):
            X = np.array([], dtype=np.float32)
        else:
            X[np.abs(X - min_) < 0.05] += 0.1

            

Reported by Pylint.

                         inplace=st.booleans(),
           **hu.gcs)
    @settings(deadline=10000)
    def test_clip(self, X, min_, max_, inplace, gc, dc):
        # go away from the origin point to avoid kink problems
        if np.isscalar(X):
            X = np.array([], dtype=np.float32)
        else:
            X[np.abs(X - min_) < 0.05] += 0.1

            

Reported by Pylint.

                         inplace=st.booleans(),
           **hu.gcs)
    @settings(deadline=10000)
    def test_clip(self, X, min_, max_, inplace, gc, dc):
        # go away from the origin point to avoid kink problems
        if np.isscalar(X):
            X = np.array([], dtype=np.float32)
        else:
            X[np.abs(X - min_) < 0.05] += 0.1

            

Reported by Pylint.

                         inplace=st.booleans(),
           **hu.gcs)
    @settings(deadline=10000)
    def test_clip(self, X, min_, max_, inplace, gc, dc):
        # go away from the origin point to avoid kink problems
        if np.isscalar(X):
            X = np.array([], dtype=np.float32)
        else:
            X[np.abs(X - min_) < 0.05] += 0.1

            

Reported by Pylint.

                         inplace=st.booleans(),
           **hu.gcs)
    @settings(deadline=10000)
    def test_clip(self, X, min_, max_, inplace, gc, dc):
        # go away from the origin point to avoid kink problems
        if np.isscalar(X):
            X = np.array([], dtype=np.float32)
        else:
            X[np.abs(X - min_) < 0.05] += 0.1

            

Reported by Pylint.

                          X[np.abs(X - min_) < 0.05] += 0.1
            X[np.abs(X - max_) < 0.05] += 0.1

        def clip_ref(X):
            X = X.clip(min_, max_)
            return (X,)

        op = core.CreateOperator(
            "Clip",

            

Reported by Pylint.

              
import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial
import hypothesis.strategies as st
import itertools as it
import numpy as np


class TestMomentsOp(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

            

Reported by Pylint.

              import caffe2.python.hypothesis_test_util as hu
import caffe2.python.serialized_test.serialized_test_util as serial
import hypothesis.strategies as st
import itertools as it
import numpy as np


class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):

            

Reported by Pylint.

              import numpy as np


class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],

            

Reported by Pylint.

              

class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],

            

Reported by Pylint.

              

class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],

            

Reported by Pylint.

              

class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],

            

Reported by Pylint.

              

class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],

            

Reported by Pylint.

              

class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],

            

Reported by Pylint.

              class TestMomentsOp(serial.SerializedTestCase):
    def run_moments_test(self, X, axes, keepdims, gc, dc):
        if axes is None:
            op = core.CreateOperator(
                "Moments",
                ["X"],
                ["mean", "variance"],
                keepdims=keepdims,
            )

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script

            

Reported by Pylint.

                      # zero base and negative exponent case that should trigger RunTimeError
        self.assertRaises(RuntimeError, fn_int_int, 0, -2)

        '''
        2. Testing a = int, b = float
        '''
        @torch.jit.script
        def fn_int_float(a: int, b: float):
            return a ** b

            

Reported by Pylint.

                      # zero base and negative exponent case that should trigger RunTimeError
        self.assertRaises(RuntimeError, fn_int_float, 0, -2.5)

        '''
        3. Testing a = float, b = int
        '''
        @torch.jit.script
        def fn_float_int(a: float, b: int):
            return a ** b

            

Reported by Pylint.

                      # zero base and negative exponent case that should trigger RunTimeError
        self.assertRaises(RuntimeError, fn_float_int, 0.0, -2)

        '''
        4. Testing a = float, b = float
        '''
        @torch.jit.script
        def fn_float_float(a: float, b: float):
            return a ** b

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script

            

Reported by Pylint.

              from torch.testing._internal.common_utils import TestCase

class TestAtenPow(TestCase):
    def test_aten_pow_zero_negative_exponent(self):
        '''
        1. Testing a = int, b = int
        '''
        @torch.jit.script
        def fn_int_int(a: int, b: int):

            

Reported by Pylint.

                      1. Testing a = int, b = int
        '''
        @torch.jit.script
        def fn_int_int(a: int, b: int):
            return a ** b
        # Existing correct behaviors of aten::pow
        self.assertEqual(fn_int_int(2, 1), 2 ** 1)
        self.assertEqual(fn_int_int(2, 0), 2 ** 0)
        self.assertEqual(fn_int_int(2, -2), 2 ** (-2))

            

Reported by Pylint.

              from typing import ClassVar, cast
from unittest import TestCase

from etcd import EtcdKeyNotFound  # type: ignore[import]

from torch.distributed.elastic.rendezvous import RendezvousConnectionError, RendezvousParameters
from torch.distributed.elastic.rendezvous.etcd_rendezvous_backend import (
    EtcdRendezvousBackend,
    create_backend,

            

Reported by Pylint.

              
from etcd import EtcdKeyNotFound  # type: ignore[import]

from torch.distributed.elastic.rendezvous import RendezvousConnectionError, RendezvousParameters
from torch.distributed.elastic.rendezvous.etcd_rendezvous_backend import (
    EtcdRendezvousBackend,
    create_backend,
)
from torch.distributed.elastic.rendezvous.etcd_server import EtcdServer

            

Reported by Pylint.

              from etcd import EtcdKeyNotFound  # type: ignore[import]

from torch.distributed.elastic.rendezvous import RendezvousConnectionError, RendezvousParameters
from torch.distributed.elastic.rendezvous.etcd_rendezvous_backend import (
    EtcdRendezvousBackend,
    create_backend,
)
from torch.distributed.elastic.rendezvous.etcd_server import EtcdServer
from torch.distributed.elastic.rendezvous.etcd_store import EtcdStore

            

Reported by Pylint.

                  EtcdRendezvousBackend,
    create_backend,
)
from torch.distributed.elastic.rendezvous.etcd_server import EtcdServer
from torch.distributed.elastic.rendezvous.etcd_store import EtcdStore

from rendezvous_backend_test import RendezvousBackendTestMixin



            

Reported by Pylint.

                  create_backend,
)
from torch.distributed.elastic.rendezvous.etcd_server import EtcdServer
from torch.distributed.elastic.rendezvous.etcd_store import EtcdStore

from rendezvous_backend_test import RendezvousBackendTestMixin


class EtcdRendezvousBackendTest(TestCase, RendezvousBackendTestMixin):

            

Reported by Pylint.

              from torch.distributed.elastic.rendezvous.etcd_server import EtcdServer
from torch.distributed.elastic.rendezvous.etcd_store import EtcdStore

from rendezvous_backend_test import RendezvousBackendTestMixin


class EtcdRendezvousBackendTest(TestCase, RendezvousBackendTestMixin):
    _server: ClassVar[EtcdServer]


            

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.

import subprocess
from base64 import b64encode
from typing import ClassVar, cast

            

Reported by Pylint.

              # This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

import subprocess
from base64 import b64encode
from typing import ClassVar, cast
from unittest import TestCase

from etcd import EtcdKeyNotFound  # type: ignore[import]

            

Reported by Bandit.

              from rendezvous_backend_test import RendezvousBackendTestMixin


class EtcdRendezvousBackendTest(TestCase, RendezvousBackendTestMixin):
    _server: ClassVar[EtcdServer]

    @classmethod
    def setUpClass(cls) -> None:
        cls._server = EtcdServer()

            

Reported by Pylint.

                      self._client.write("/dummy_prefix/dummy_run_id", "non_base64")


class CreateBackendTest(TestCase):
    _server: ClassVar[EtcdServer]

    @classmethod
    def setUpClass(cls) -> None:
        cls._server = EtcdServer()

            

Reported by Pylint.