from caffe2.python import core, workspace
from caffe2.python.test_util import TestCase
import numpy as np
import unittest


            

Reported by Pylint.

              from caffe2.python import core, workspace
from caffe2.python.test_util import TestCase
import numpy as np
import unittest


class DoOpTest(TestCase):
    def test_operator(self):
        def make_net():

            

Reported by Pylint.

              import unittest


class DoOpTest(TestCase):
    def test_operator(self):
        def make_net():
            subnet = core.Net('subnet')
            subnet.Add(["X", "Y"], "Z")


            

Reported by Pylint.

              

class DoOpTest(TestCase):
    def test_operator(self):
        def make_net():
            subnet = core.Net('subnet')
            subnet.Add(["X", "Y"], "Z")

            net = core.Net("net")

            

Reported by Pylint.

                      workspace.FeedBlob("outer_Y", np.asarray([3, 4]))

        workspace.RunNetOnce(net)
        outer_Z_val = workspace.FetchBlob("outer_Z")
        self.assertTrue(np.all(outer_Z_val == np.asarray([4, 6])))

    def test_reuse_workspace(self):
        def make_net():
            param_init_subnet = core.Net('param_init_subnet')

            

Reported by Pylint.

                      outer_Z_val = workspace.FetchBlob("outer_Z")
        self.assertTrue(np.all(outer_Z_val == np.asarray([4, 6])))

    def test_reuse_workspace(self):
        def make_net():
            param_init_subnet = core.Net('param_init_subnet')
            param_init_subnet.ConstantFill([], "X", shape=[1], value=1)
            param_init_subnet.ConstantFill([], "Y", shape=[1], value=2)


            

Reported by Pylint.

              
        workspace.ResetWorkspace()
        workspace.RunNetOnce(net)
        outer_Z_val = workspace.FetchBlob("outer_Z")
        self.assertTrue(np.all(outer_Z_val == np.asarray([3])))


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

            

Reported by Pylint.

              class Transformer(unittest.TestCase):
    def test_fuse(self):
        net_swish = core.Net("test_swish")
        net_swish_init = core.Net("test_swish_init")

        deq = core.CreateOperator("Int8DequantizeNNPI", ["Xq"], ["X"])
        swish = core.CreateOperator("SwishFakeFp16NNPI", ["X"], ["Y"])
        quant = core.CreateOperator("Int8QuantizeNNPI", ["Y"], ["Y_q"])
        net_swish.Proto().op.extend(

            

Reported by Pylint.

              



import unittest
from caffe2.python.fakefp16_transform_lib import fakeFp16FuseOps
from caffe2.python import core

class Transformer(unittest.TestCase):

            

Reported by Pylint.

              from caffe2.python.fakefp16_transform_lib import fakeFp16FuseOps
from caffe2.python import core

class Transformer(unittest.TestCase):
    def test_fuse(self):
        net_swish = core.Net("test_swish")
        net_swish_init = core.Net("test_swish_init")

        deq = core.CreateOperator("Int8DequantizeNNPI", ["Xq"], ["X"])

            

Reported by Pylint.

              from caffe2.python import core

class Transformer(unittest.TestCase):
    def test_fuse(self):
        net_swish = core.Net("test_swish")
        net_swish_init = core.Net("test_swish_init")

        deq = core.CreateOperator("Int8DequantizeNNPI", ["Xq"], ["X"])
        swish = core.CreateOperator("SwishFakeFp16NNPI", ["X"], ["Y"])

            

Reported by Pylint.

              from caffe2.python import core

class Transformer(unittest.TestCase):
    def test_fuse(self):
        net_swish = core.Net("test_swish")
        net_swish_init = core.Net("test_swish_init")

        deq = core.CreateOperator("Int8DequantizeNNPI", ["Xq"], ["X"])
        swish = core.CreateOperator("SwishFakeFp16NNPI", ["X"], ["Y"])

            

Reported by Pylint.

                      )
        print(net_swish.Proto())
        out_net = fakeFp16FuseOps(net_swish.Proto())
        assert(len(out_net.op) == 1)

            

Reported by Pylint.

                      )
        print(net_swish.Proto())
        out_net = fakeFp16FuseOps(net_swish.Proto())
        assert(len(out_net.op) == 1)

            

Reported by Bandit.

              import torch
# This file contains definitions of script classes.
# They are used by test_jit.py to test ScriptClass imports


@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              import torch
# This file contains definitions of script classes.
# They are used by test_jit.py to test ScriptClass imports


@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              import torch
# This file contains definitions of script classes.
# They are used by test_jit.py to test ScriptClass imports


@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              

@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              

@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              

@torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              @torch.jit.script  # noqa: B903
class FooSameName(object):  # noqa: B903
    def __init__(self, y):
        self.y = y

            

Reported by Pylint.

              import pytest

import torch
from torch.distributions.utils import tril_matrix_to_vec, vec_to_tril_matrix


@pytest.mark.parametrize('shape', [
    (2, 2),
    (3, 3),

            

Reported by Pylint.

              import pytest

import torch
from torch.distributions.utils import tril_matrix_to_vec, vec_to_tril_matrix


@pytest.mark.parametrize('shape', [
    (2, 2),
    (3, 3),

            

Reported by Pylint.

              import pytest

import torch
from torch.distributions.utils import tril_matrix_to_vec, vec_to_tril_matrix


@pytest.mark.parametrize('shape', [
    (2, 2),
    (3, 3),

            

Reported by Pylint.

              import pytest

import torch
from torch.distributions.utils import tril_matrix_to_vec, vec_to_tril_matrix


@pytest.mark.parametrize('shape', [
    (2, 2),
    (3, 3),

            

Reported by Pylint.

                  (3, 3),
    (2, 4, 4),
    (2, 2, 4, 4),
])
def test_tril_matrix_to_vec(shape):
    mat = torch.randn(shape)
    n = mat.shape[-1]
    for diag in range(-n, n):
        actual = mat.tril(diag)

            

Reported by Pylint.

              ])
def test_tril_matrix_to_vec(shape):
    mat = torch.randn(shape)
    n = mat.shape[-1]
    for diag in range(-n, n):
        actual = mat.tril(diag)
        vec = tril_matrix_to_vec(actual, diag)
        tril_mat = vec_to_tril_matrix(vec, diag)
        assert torch.allclose(tril_mat, actual)

            

Reported by Pylint.

                      actual = mat.tril(diag)
        vec = tril_matrix_to_vec(actual, diag)
        tril_mat = vec_to_tril_matrix(vec, diag)
        assert torch.allclose(tril_mat, actual)


if __name__ == '__main__':
    pytest.main([__file__])

            

Reported by Bandit.

              


from hypothesis import given
import numpy as np
import hypothesis.strategies as st

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

            

Reported by Pylint.

              
from hypothesis import given
import numpy as np
import hypothesis.strategies as st

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



            

Reported by Pylint.

                      input=hu.tensor(dtype=np.float32),
        dev_options=_dev_options()
    )
    def test_ensure_cpu_output(self, input, dev_options):
        op_dev, input_blob_dev = dev_options
        net = core.Net('test_net')
        data = net.GivenTensorFill(
            [],
            ["data"],

            

Reported by Pylint.

              




from hypothesis import given
import numpy as np
import hypothesis.strategies as st


            

Reported by Pylint.

                  return op_dev, input_blob_dev


class TestEnsureCPUOutputOp(hu.HypothesisTestCase):

    @given(
        input=hu.tensor(dtype=np.float32),
        dev_options=_dev_options()
    )

            

Reported by Pylint.

                  @given(
        input=hu.tensor(dtype=np.float32),
        dev_options=_dev_options()
    )
    def test_ensure_cpu_output(self, input, dev_options):
        op_dev, input_blob_dev = dev_options
        net = core.Net('test_net')
        data = net.GivenTensorFill(
            [],

            

Reported by Pylint.

                  @given(
        input=hu.tensor(dtype=np.float32),
        dev_options=_dev_options()
    )
    def test_ensure_cpu_output(self, input, dev_options):
        op_dev, input_blob_dev = dev_options
        net = core.Net('test_net')
        data = net.GivenTensorFill(
            [],

            

Reported by Pylint.

              # LICENSE file in the root directory of this source tree.
import unittest

import torch.distributed.elastic.utils.logging as logging
from torch.testing._internal.common_utils import run_tests

log = logging.get_logger()



            

Reported by Pylint.

              import unittest

import torch.distributed.elastic.utils.logging as logging
from torch.testing._internal.common_utils import run_tests

log = logging.get_logger()


class LoggingTest(unittest.TestCase):

            

Reported by Pylint.

                      self.assertEqual("foobar", name_override_log.name)

    def test_derive_module_name(self):
        module_name = logging._derive_module_name(depth=1)
        self.assertEqual(__name__, module_name)


if __name__ == "__main__":
    run_tests()

            

Reported by Pylint.

              #!/usr/bin/env python3

# 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 unittest


            

Reported by Pylint.

              log = logging.get_logger()


class LoggingTest(unittest.TestCase):
    def setUp(self):
        self.clazz_log = logging.get_logger()

    def test_logger_name(self):
        local_log = logging.get_logger()

            

Reported by Pylint.

                  def setUp(self):
        self.clazz_log = logging.get_logger()

    def test_logger_name(self):
        local_log = logging.get_logger()
        name_override_log = logging.get_logger("foobar")

        self.assertEqual(__name__, log.name)
        self.assertEqual(__name__, self.clazz_log.name)

            

Reported by Pylint.

                      self.assertEqual(__name__, local_log.name)
        self.assertEqual("foobar", name_override_log.name)

    def test_derive_module_name(self):
        module_name = logging._derive_module_name(depth=1)
        self.assertEqual(__name__, module_name)


if __name__ == "__main__":

            

Reported by Pylint.

              }

C10_EXPORT bool ReadProtoFromTextFile(const char* filename, Message* proto) {
  int fd = open(filename, O_RDONLY);
  CAFFE_ENFORCE_NE(fd, -1, "File not found: ", filename);
  FileInputStream* input = new FileInputStream(fd);
  bool success = google::protobuf::TextFormat::Parse(input, proto);
  delete input;
  close(fd);

            

Reported by FlawFinder.

                  const Message& proto,
    const char* filename,
    bool throwIfError) {
  int fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0644);
  FileOutputStream* output = new FileOutputStream(fd);
  if(!google::protobuf::TextFormat::Print(proto, output)) {
     if (throwIfError) {
       CAFFE_THROW("Cannot write proto to text file: ", filename);
     } else {

            

Reported by FlawFinder.

                  const char* filename,
    MessageLite* proto) {
#if defined(_MSC_VER) // for MSC compiler binary flag needs to be specified
  int fd = open(filename, O_RDONLY | O_BINARY);
#else
  int fd = open(filename, O_RDONLY);
#endif
  CAFFE_ENFORCE_NE(fd, -1, "File not found: ", filename);
  std::unique_ptr<ZeroCopyInputStream> raw_input(new FileInputStream(fd));

            

Reported by FlawFinder.

              #if defined(_MSC_VER) // for MSC compiler binary flag needs to be specified
  int fd = open(filename, O_RDONLY | O_BINARY);
#else
  int fd = open(filename, O_RDONLY);
#endif
  CAFFE_ENFORCE_NE(fd, -1, "File not found: ", filename);
  std::unique_ptr<ZeroCopyInputStream> raw_input(new FileInputStream(fd));
  std::unique_ptr<CodedInputStream> coded_input(
      new CodedInputStream(raw_input.get()));

            

Reported by FlawFinder.

              C10_EXPORT void WriteProtoToBinaryFile(
    const MessageLite& proto,
    const char* filename) {
  int fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0644);
  CAFFE_ENFORCE_NE(
      fd, -1, "File cannot be created: ", filename, " error number: ", errno);
  std::unique_ptr<ZeroCopyOutputStream> raw_output(new FileOutputStream(fd));
  std::unique_ptr<CodedOutputStream> coded_output(
      new CodedOutputStream(raw_output.get()));

            

Reported by FlawFinder.

                size_t n = ifs.tellg();
  str->resize(n);
  ifs.seekg(0);
  ifs.read(&(*str)[0], n);
  return true;
}

C10_EXPORT bool WriteStringToFile(const string& str, const char* filename) {
  std::ofstream ofs(filename, std::ios::out | std::ios::trunc);

            

Reported by FlawFinder.

                  if (!ifs_) {
      return -1;
    }
    ifs_.read(static_cast<char*>(buffer), size);
    return ifs_.gcount();
  }

 private:
  std::ifstream ifs_;

            

Reported by FlawFinder.

              #!/usr/bin/env python3
import os
import sys
import unittest
from enum import Flag, auto
from typing import Dict, List, Type

from torch.testing._internal.common_distributed import MultiProcessTestCase
from torch.testing._internal.common_utils import (

            

Reported by Pylint.

                  TensorPipeAgentRpcTest,
    TensorPipeAgentCudaRpcTest,
)
from torch.testing._internal.distributed.rpc.examples.parameter_server_test import ParameterServerTest
from torch.testing._internal.distributed.rpc.examples.reinforcement_learning_rpc_test import (
    ReinforcementLearningRpcTest,
)



            

Reported by Pylint.

              

@unittest.skipIf(TEST_WITH_TSAN, "TSAN and fork() is broken")
class ForkHelper(MultiProcessTestCase):
    def setUp(self):
        super().setUp()
        _check_and_set_tcp_init()
        self._fork_processes()


            

Reported by Pylint.

              @unittest.skipIf(
    TEST_WITH_DEV_DBG_ASAN, "Skip ASAN as torch + multiprocessing spawn have known issues"
)
class SpawnHelper(MultiProcessTestCase):
    def setUp(self):
        super().setUp()
        _check_and_set_tcp_init()
        self._spawn_processes()


            

Reported by Pylint.

                      super().tearDown()


class MultiProcess(Flag):
    FORK = auto()
    SPAWN = auto()


MP_HELPERS_AND_SUFFIXES = {

            

Reported by Pylint.

                          if mp_type & mp_type_filter:
                mp_helper, suffix = MP_HELPERS_AND_SUFFIXES[mp_type]
                if IS_SANDCASTLE:
                    if mp_helper == SpawnHelper and TEST_WITH_DEV_DBG_ASAN:
                        print(
                            f'Skipping test {test_class} on sandcastle for the following reason: '
                            'Skip dev-asan as torch + multiprocessing spawn have known issues', file=sys.stderr)
                        continue
                    elif mp_helper == ForkHelper and TEST_WITH_TSAN:

            

Reported by Pylint.

                                  if mp_helper == SpawnHelper and TEST_WITH_DEV_DBG_ASAN:
                        print(
                            f'Skipping test {test_class} on sandcastle for the following reason: '
                            'Skip dev-asan as torch + multiprocessing spawn have known issues', file=sys.stderr)
                        continue
                    elif mp_helper == ForkHelper and TEST_WITH_TSAN:
                        print(
                            f'Skipping test {test_class} on sandcastle for the following reason: '
                            'TSAN and fork() is broken'

            

Reported by Pylint.

                  def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super(BNReLU2d, self).__init__(num_features, eps=eps, momentum=momentum)

    def forward(self, input):
        # Temporarily using len(shape) instead of ndim due to JIT issue
        # https://github.com/pytorch/pytorch/issues/23890
        if len(input.shape) != 4:
            raise ValueError("Input shape must be `(N, C, H, W)`!")
        return torch.ops.quantized.batch_norm2d_relu(

            

Reported by Pylint.

              
    @classmethod
    def from_float(cls, mod):
        # TODO: Add qat support for BNReLU2d
        return super(BNReLU2d, cls).from_float(mod)


class BNReLU3d(nnq.BatchNorm3d):
    r"""

            

Reported by Pylint.

                  def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super(BNReLU3d, self).__init__(num_features, eps=eps, momentum=momentum)

    def forward(self, input):
        # Temporarily using len(shape) instead of ndim due to JIT issue
        # https://github.com/pytorch/pytorch/issues/23890
        if len(input.shape) != 5:
            raise ValueError("Input shape must be `(N, C, D, H, W)`!")
        return torch.ops.quantized.batch_norm3d_relu(

            

Reported by Pylint.

              
    @classmethod
    def from_float(cls, mod):
        # TODO: Add qat support for BNReLU3d
        return super(BNReLU3d, cls).from_float(mod)

            

Reported by Pylint.

              
import torch
import torch.nn.intrinsic
import torch.nn.intrinsic.qat
import torch.nn.quantized as nnq


class BNReLU2d(nnq.BatchNorm2d):
    r"""

            

Reported by Pylint.

                  _FLOAT_MODULE = torch.nn.intrinsic.BNReLU2d

    def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super(BNReLU2d, self).__init__(num_features, eps=eps, momentum=momentum)

    def forward(self, input):
        # Temporarily using len(shape) instead of ndim due to JIT issue
        # https://github.com/pytorch/pytorch/issues/23890
        if len(input.shape) != 4:

            

Reported by Pylint.

                  _FLOAT_MODULE = torch.nn.intrinsic.BNReLU3d

    def __init__(self, num_features, eps=1e-5, momentum=0.1):
        super(BNReLU3d, self).__init__(num_features, eps=eps, momentum=momentum)

    def forward(self, input):
        # Temporarily using len(shape) instead of ndim due to JIT issue
        # https://github.com/pytorch/pytorch/issues/23890
        if len(input.shape) != 5:

            

Reported by Pylint.

              from torch._C import _from_dlpack as from_dlpack
from torch._C import _to_dlpack as to_dlpack

torch._C._add_docstr(from_dlpack, r"""from_dlpack(dlpack) -> Tensor

Decodes a DLPack to a tensor.

Args:
    dlpack: a PyCapsule object with the dltensor

            

Reported by Pylint.

              from torch._C import _from_dlpack as from_dlpack
from torch._C import _to_dlpack as to_dlpack

torch._C._add_docstr(from_dlpack, r"""from_dlpack(dlpack) -> Tensor

Decodes a DLPack to a tensor.

Args:
    dlpack: a PyCapsule object with the dltensor

            

Reported by Pylint.

              Note that each dlpack can only be consumed once.
""")

torch._C._add_docstr(to_dlpack, r"""to_dlpack(tensor) -> PyCapsule

Returns a DLPack representing the tensor.

Args:
    tensor: a tensor to be exported

            

Reported by Pylint.

              Note that each dlpack can only be consumed once.
""")

torch._C._add_docstr(to_dlpack, r"""to_dlpack(tensor) -> PyCapsule

Returns a DLPack representing the tensor.

Args:
    tensor: a tensor to be exported

            

Reported by Pylint.

              import torch

from torch._C import _from_dlpack as from_dlpack
from torch._C import _to_dlpack as to_dlpack

torch._C._add_docstr(from_dlpack, r"""from_dlpack(dlpack) -> Tensor

Decodes a DLPack to a tensor.


            

Reported by Pylint.

              from torch._C import _from_dlpack as from_dlpack
from torch._C import _to_dlpack as to_dlpack

torch._C._add_docstr(from_dlpack, r"""from_dlpack(dlpack) -> Tensor

Decodes a DLPack to a tensor.

Args:
    dlpack: a PyCapsule object with the dltensor

            

Reported by Pylint.

              Note that each dlpack can only be consumed once.
""")

torch._C._add_docstr(to_dlpack, r"""to_dlpack(tensor) -> PyCapsule

Returns a DLPack representing the tensor.

Args:
    tensor: a tensor to be exported

            

Reported by Pylint.