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


def update_counter_ref(prev_iter, update_counter, indices, curr_iter, counter_halflife):
    prev_iter_out = prev_iter.copy()
    update_counter_out = update_counter.copy()

    counter_neg_log_rho = np.log(2) / counter_halflife
    for i in indices:

            

Reported by Pylint.

                  return prev_iter_out, update_counter_out


class TestRowWiseCounter(hu.HypothesisTestCase):
    def test_rowwise_counter(self):
        h = 8 * 20
        n = 5
        curr_iter = np.array([100], dtype=np.int64)


            

Reported by Pylint.

              

class TestRowWiseCounter(hu.HypothesisTestCase):
    def test_rowwise_counter(self):
        h = 8 * 20
        n = 5
        curr_iter = np.array([100], dtype=np.int64)

        update_counter = np.random.randint(99, size=h).astype(np.float64)

            

Reported by Pylint.

              

class TestRowWiseCounter(hu.HypothesisTestCase):
    def test_rowwise_counter(self):
        h = 8 * 20
        n = 5
        curr_iter = np.array([100], dtype=np.int64)

        update_counter = np.random.randint(99, size=h).astype(np.float64)

            

Reported by Pylint.

              
class TestRowWiseCounter(hu.HypothesisTestCase):
    def test_rowwise_counter(self):
        h = 8 * 20
        n = 5
        curr_iter = np.array([100], dtype=np.int64)

        update_counter = np.random.randint(99, size=h).astype(np.float64)
        prev_iter = np.random.rand(h, 1).astype(np.int64)

            

Reported by Pylint.

              class TestRowWiseCounter(hu.HypothesisTestCase):
    def test_rowwise_counter(self):
        h = 8 * 20
        n = 5
        curr_iter = np.array([100], dtype=np.int64)

        update_counter = np.random.randint(99, size=h).astype(np.float64)
        prev_iter = np.random.rand(h, 1).astype(np.int64)
        indices = np.unique(np.random.randint(0, h, size=n))

            

Reported by Pylint.

                          curr_iter,
            counter_halflife=counter_halflife,
        )
        assert np.allclose(prev_iter_out, prev_iter_out_ref, rtol=1e-3)
        assert np.allclose(update_counter_out, update_counter_out_ref, rtol=1e-3)


if __name__ == "__main__":
    global_options = ["caffe2"]

            

Reported by Bandit.

                          counter_halflife=counter_halflife,
        )
        assert np.allclose(prev_iter_out, prev_iter_out_ref, rtol=1e-3)
        assert np.allclose(update_counter_out, update_counter_out_ref, rtol=1e-3)


if __name__ == "__main__":
    global_options = ["caffe2"]
    core.GlobalInit(global_options)

            

Reported by Bandit.

              import torch


def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)

            

Reported by Pylint.

              def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)
        print(desc)
        print('-' * 80)
        print(error_message)

            

Reported by Pylint.

                          lambda: torch.IntStorage(10).fill_('asdf'),
            'str')

# TODO: frombuffer

            

Reported by Pylint.

              import torch


def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)

            

Reported by Pylint.

              import torch


def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)

            

Reported by Pylint.

              import torch


def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)

            

Reported by Pylint.

              def check_error(desc, fn, *required_substrings):
    try:
        fn()
    except Exception as e:
        error_message = e.args[0]
        print('=' * 80)
        print(desc)
        print('-' * 80)
        print(error_message)

            

Reported by Pylint.

                      print(error_message)
        print('')
        for sub in required_substrings:
            assert sub in error_message
        return
    raise AssertionError("given function ({}) didn't raise an error".format(desc))

check_error(
    'Wrong argument types',

            

Reported by Bandit.

                          'str')


def assign():
    torch.FloatStorage(10)[1:-1] = '1'
check_error('Invalid value type',
            assign,
            'str')


            

Reported by Pylint.

              
import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

            

Reported by Pylint.

              import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)


            

Reported by Pylint.

                  print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,

            

Reported by Pylint.

                  sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,
    TENSORPIPE_TESTS,

            

Reported by Pylint.

              from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,
    TENSORPIPE_TESTS,
    MultiProcess,
    generate_tests,
)

            

Reported by Pylint.

              #!/usr/bin/env python3

import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)

            

Reported by Pylint.

                  print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,

            

Reported by Pylint.

                  sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,
    TENSORPIPE_TESTS,

            

Reported by Pylint.

              from torch.testing._internal.distributed.rpc.tensorpipe_rpc_agent_test_fixture import (
    TensorPipeRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    GENERIC_TESTS,
    TENSORPIPE_TESTS,
    MultiProcess,
    generate_tests,
)

            

Reported by Pylint.

              
import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

            

Reported by Pylint.

              import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)


            

Reported by Pylint.

                  print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,

            

Reported by Pylint.

                  sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,
    MultiProcess,

            

Reported by Pylint.

              from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,
    MultiProcess,
    generate_tests,
)


            

Reported by Pylint.

              #!/usr/bin/env python3

import sys

import torch
import torch.distributed as dist

if not dist.is_available():
    print("Distributed not available, skipping tests", file=sys.stderr)

            

Reported by Pylint.

                  print("Distributed not available, skipping tests", file=sys.stderr)
    sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,

            

Reported by Pylint.

                  sys.exit(0)

from torch.testing._internal.common_utils import IS_IN_CI, run_tests
from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,
    MultiProcess,

            

Reported by Pylint.

              from torch.testing._internal.distributed.rpc.faulty_rpc_agent_test_fixture import (
    FaultyRpcAgentTestFixture,
)
from torch.testing._internal.distributed.rpc_utils import (
    FAULTY_AGENT_TESTS,
    MultiProcess,
    generate_tests,
)


            

Reported by Pylint.

              import torch

from .MetricBase import MetricBase


class CUDAMetric(MetricBase):
    def __init__(self, rank: int, name: str):
        self.rank = rank
        self.name = name

            

Reported by Pylint.

              import torch

from .MetricBase import MetricBase


class CUDAMetric(MetricBase):
    def __init__(self, rank: int, name: str):
        self.rank = rank
        self.name = name

            

Reported by Pylint.

              import torch

from .MetricBase import MetricBase


class CUDAMetric(MetricBase):
    def __init__(self, rank: int, name: str):
        self.rank = rank
        self.name = name

            

Reported by Pylint.

              import torch

from .MetricBase import MetricBase


class CUDAMetric(MetricBase):
    def __init__(self, rank: int, name: str):
        self.rank = rank
        self.name = name

            

Reported by Pylint.

              from .MetricBase import MetricBase


class CUDAMetric(MetricBase):
    def __init__(self, rank: int, name: str):
        self.rank = rank
        self.name = name
        self.start = None
        self.end = None

            

Reported by Pylint.

                      self.start = None
        self.end = None

    def record_start(self):
        self.start = torch.cuda.Event(enable_timing=True)
        with torch.cuda.device(self.rank):
            self.start.record()

    def record_end(self):

            

Reported by Pylint.

                      with torch.cuda.device(self.rank):
            self.start.record()

    def record_end(self):
        self.end = torch.cuda.Event(enable_timing=True)
        with torch.cuda.device(self.rank):
            self.end.record()

    def elapsed_time(self):

            

Reported by Pylint.

                      with torch.cuda.device(self.rank):
            self.end.record()

    def elapsed_time(self):
        if not self.start.query():
            raise RuntimeError("start event did not complete")
        if not self.end.query():
            raise RuntimeError("end event did not complete")
        return self.start.elapsed_time(self.end)

            

Reported by Pylint.

                          raise RuntimeError("end event did not complete")
        return self.start.elapsed_time(self.end)

    def synchronize(self):
        self.start.synchronize()
        self.end.synchronize()

            

Reported by Pylint.

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

from utils import NUM_LOOP_ITERS

workspace.GlobalInit(['caffe2'])

def add_blob(ws, blob_name, tensor_size):
    blob_tensor = np.random.randn(*tensor_size).astype(np.float32)

            

Reported by Pylint.

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

from utils import NUM_LOOP_ITERS

workspace.GlobalInit(['caffe2'])

def add_blob(ws, blob_name, tensor_size):
    blob_tensor = np.random.randn(*tensor_size).astype(np.float32)

            

Reported by Pylint.

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

from utils import NUM_LOOP_ITERS

workspace.GlobalInit(['caffe2'])

def add_blob(ws, blob_name, tensor_size):
    blob_tensor = np.random.randn(*tensor_size).astype(np.float32)

            

Reported by Pylint.

              
workspace.GlobalInit(['caffe2'])

def add_blob(ws, blob_name, tensor_size):
    blob_tensor = np.random.randn(*tensor_size).astype(np.float32)
    ws.FeedBlob(blob_name, blob_tensor)

class C2SimpleNet(object):
    """

            

Reported by Pylint.

              
workspace.GlobalInit(['caffe2'])

def add_blob(ws, blob_name, tensor_size):
    blob_tensor = np.random.randn(*tensor_size).astype(np.float32)
    ws.FeedBlob(blob_name, blob_tensor)

class C2SimpleNet(object):
    """

            

Reported by Pylint.

                  blob_tensor = np.random.randn(*tensor_size).astype(np.float32)
    ws.FeedBlob(blob_name, blob_tensor)

class C2SimpleNet(object):
    """
    This module constructs a net with 'op_name' operator. The net consist
    a series of such operator.
    It initializes the workspace with input blob equal to the number of parameters
    needed for the op.

            

Reported by Pylint.

                  blob_tensor = np.random.randn(*tensor_size).astype(np.float32)
    ws.FeedBlob(blob_name, blob_tensor)

class C2SimpleNet(object):
    """
    This module constructs a net with 'op_name' operator. The net consist
    a series of such operator.
    It initializes the workspace with input blob equal to the number of parameters
    needed for the op.

            

Reported by Pylint.

                      for _ in range(NUM_LOOP_ITERS):
            output_name = self.net._net.op[-1].output
            self.input_names[-1] = output_name[0]
            assert len(self.input_names) == num_inputs
            op_constructor(self.input_names)
        workspace.CreateNet(self.net)
        if debug:
            print(self.net._net)


            

Reported by Bandit.

                      if debug:
            print(self.net._net)

    def forward(self, niters):
        workspace.RunNet(self.net, niters, False)

            

Reported by Pylint.

              




from caffe2.python import schema
from caffe2.python.layers.layers import (
    IdList,
    ModelLayer,

            

Reported by Pylint.

              # where the 2nd item values [3,4,5] were replaced with [-1] and the length got
# set to 1.

class SparseDropoutWithReplacement(ModelLayer):
    def __init__(
            self,
            model,
            input_record,
            dropout_prob_train,

            

Reported by Pylint.

              # set to 1.

class SparseDropoutWithReplacement(ModelLayer):
    def __init__(
            self,
            model,
            input_record,
            dropout_prob_train,
            dropout_prob_eval,

            

Reported by Pylint.

                          name='sparse_dropout',
            **kwargs):

        super(SparseDropoutWithReplacement, self).__init__(model, name, input_record, **kwargs)
        assert schema.equal_schemas(input_record, IdList), "Incorrect input type"

        self.dropout_prob_train = float(dropout_prob_train)
        self.dropout_prob_eval = float(dropout_prob_eval)
        self.dropout_prob_predict = float(dropout_prob_predict)

            

Reported by Pylint.

                          **kwargs):

        super(SparseDropoutWithReplacement, self).__init__(model, name, input_record, **kwargs)
        assert schema.equal_schemas(input_record, IdList), "Incorrect input type"

        self.dropout_prob_train = float(dropout_prob_train)
        self.dropout_prob_eval = float(dropout_prob_eval)
        self.dropout_prob_predict = float(dropout_prob_predict)
        self.replacement_value = int(replacement_value)

            

Reported by Bandit.

                      self.dropout_prob_eval = float(dropout_prob_eval)
        self.dropout_prob_predict = float(dropout_prob_predict)
        self.replacement_value = int(replacement_value)
        assert (self.dropout_prob_train >= 0 and
                self.dropout_prob_train <= 1.0), \
            "Expected 0 <= dropout_prob_train <= 1, but got %s" \
            % self.dropout_prob_train
        assert (self.dropout_prob_eval >= 0 and
                self.dropout_prob_eval <= 1.0), \

            

Reported by Bandit.

                              self.dropout_prob_train <= 1.0), \
            "Expected 0 <= dropout_prob_train <= 1, but got %s" \
            % self.dropout_prob_train
        assert (self.dropout_prob_eval >= 0 and
                self.dropout_prob_eval <= 1.0), \
            "Expected 0 <= dropout_prob_eval <= 1, but got %s" \
            % dropout_prob_eval
        assert (self.dropout_prob_predict >= 0 and
                self.dropout_prob_predict <= 1.0), \

            

Reported by Bandit.

                              self.dropout_prob_eval <= 1.0), \
            "Expected 0 <= dropout_prob_eval <= 1, but got %s" \
            % dropout_prob_eval
        assert (self.dropout_prob_predict >= 0 and
                self.dropout_prob_predict <= 1.0), \
            "Expected 0 <= dropout_prob_predict <= 1, but got %s" \
            % dropout_prob_predict
        assert(self.dropout_prob_train > 0 or
               self.dropout_prob_eval > 0 or

            

Reported by Bandit.

                              self.dropout_prob_predict <= 1.0), \
            "Expected 0 <= dropout_prob_predict <= 1, but got %s" \
            % dropout_prob_predict
        assert(self.dropout_prob_train > 0 or
               self.dropout_prob_eval > 0 or
               self.dropout_prob_predict > 0), \
            "Ratios all set to 0.0 for train, eval and predict"

        self.output_schema = schema.NewRecord(model.net, IdList)

            

Reported by Bandit.

                      """
        List of parameter blobs for prediction net
        """
        pass

    @property
    def train_param_blobs(self):
        """
        If train_param_blobs is not set before used, default to param_blobs

            

Reported by Pylint.

                      """
        Add ops to the given net, using the given param_blobs
        """
        pass

    def add_ops(self, net):
        self._add_ops(net, self.param_blobs)

    def add_train_ops(self, net):

            

Reported by Pylint.

              ## @package sampling_trainable_mixin
# Module caffe2.python.layers.sampling_trainable_mixin





import abc


            

Reported by Pylint.

              import abc


class SamplingTrainableMixin(metaclass=abc.ABCMeta):

    def __init__(self, *args, **kwargs):
        super(SamplingTrainableMixin, self).__init__(*args, **kwargs)
        self._train_param_blobs = None
        self._train_param_blobs_frozen = False

            

Reported by Pylint.

              class SamplingTrainableMixin(metaclass=abc.ABCMeta):

    def __init__(self, *args, **kwargs):
        super(SamplingTrainableMixin, self).__init__(*args, **kwargs)
        self._train_param_blobs = None
        self._train_param_blobs_frozen = False

    @property
    @abc.abstractmethod

            

Reported by Pylint.

              
    @train_param_blobs.setter
    def train_param_blobs(self, blobs):
        assert not self._train_param_blobs_frozen
        assert blobs is not None
        self._train_param_blobs_frozen = True
        self._train_param_blobs = blobs

    @abc.abstractmethod

            

Reported by Bandit.

                  @train_param_blobs.setter
    def train_param_blobs(self, blobs):
        assert not self._train_param_blobs_frozen
        assert blobs is not None
        self._train_param_blobs_frozen = True
        self._train_param_blobs = blobs

    @abc.abstractmethod
    def _add_ops(self, net, param_blobs):

            

Reported by Bandit.

                      """
        pass

    def add_ops(self, net):
        self._add_ops(net, self.param_blobs)

    def add_train_ops(self, net):
        self._add_ops(net, self.train_param_blobs)

            

Reported by Pylint.

                  def add_ops(self, net):
        self._add_ops(net, self.param_blobs)

    def add_train_ops(self, net):
        self._add_ops(net, self.train_param_blobs)

            

Reported by Pylint.

              




from caffe2.python import schema
from caffe2.python.layers.layers import ModelLayer

import numpy as np

            

Reported by Pylint.

                      b_init -- initialization options for bias parameter

    """
    def __init__(
            self,
            model,
            input_record,
            output_dims,
            sigma,  # bandwidth

            

Reported by Pylint.

                          name='random_fourier_features',
            **kwargs):

        super(RandomFourierFeatures, self).__init__(model, name, input_record,
                                                    **kwargs)
        assert isinstance(input_record, schema.Scalar), "Incorrect input type"

        input_dims = input_record.field_type().shape[0]
        assert input_dims >= 1, "Expected input dimensions >= 1, got %s" \

            

Reported by Pylint.

              
        super(RandomFourierFeatures, self).__init__(model, name, input_record,
                                                    **kwargs)
        assert isinstance(input_record, schema.Scalar), "Incorrect input type"

        input_dims = input_record.field_type().shape[0]
        assert input_dims >= 1, "Expected input dimensions >= 1, got %s" \
                                % input_dims
        self.output_dims = output_dims

            

Reported by Bandit.

                      assert isinstance(input_record, schema.Scalar), "Incorrect input type"

        input_dims = input_record.field_type().shape[0]
        assert input_dims >= 1, "Expected input dimensions >= 1, got %s" \
                                % input_dims
        self.output_dims = output_dims
        assert self.output_dims >= 1, "Expected output dimensions >= 1, got %s" \
                                      % self.output_dims


            

Reported by Bandit.

                      assert input_dims >= 1, "Expected input dimensions >= 1, got %s" \
                                % input_dims
        self.output_dims = output_dims
        assert self.output_dims >= 1, "Expected output dimensions >= 1, got %s" \
                                      % self.output_dims

        self.output_schema = schema.Scalar(
            (np.float32, (self.output_dims, )),
            self.get_next_blob_reference('output')

            

Reported by Bandit.

                          self.get_next_blob_reference('output')
        )

        assert sigma > 0.0, "Expected bandwidth > 0, got %s" % sigma

        # Initialize train_init_net parameters
        w_init = w_init if w_init else (
            'GaussianFill', {'mean': 0.0, 'std': 1.0 / sigma}
        )

            

Reported by Bandit.

                          'UniformFill', {'min': 0.0, 'max': 2 * np.pi}
        )

        self.w = self.create_param(param_name='w',
                                   shape=[self.output_dims, input_dims],
                                   initializer=w_init,
                                   optimizer=model.NoOptim)

        self.b = self.create_param(param_name='b',

            

Reported by Pylint.

                                                 initializer=w_init,
                                   optimizer=model.NoOptim)

        self.b = self.create_param(param_name='b',
                                   shape=[self.output_dims],
                                   initializer=b_init,
                                   optimizer=model.NoOptim)

    def add_ops(self, net):

            

Reported by Pylint.

              import tempfile
import shutil

from caffe2.distributed.python import StoreHandlerTimeoutError
from caffe2.distributed.store_ops_test_util import StoreOpsTests
from caffe2.python import core, workspace, dyndep
from caffe2.python.test_util import TestCase

dyndep.InitOpsLibrary("@/caffe2/caffe2/distributed:file_store_handler_ops")

            

Reported by Pylint.

              import tempfile
import shutil

from caffe2.distributed.python import StoreHandlerTimeoutError
from caffe2.distributed.store_ops_test_util import StoreOpsTests
from caffe2.python import core, workspace, dyndep
from caffe2.python.test_util import TestCase

dyndep.InitOpsLibrary("@/caffe2/caffe2/distributed:file_store_handler_ops")

            

Reported by Pylint.

              




import errno
import os
import tempfile
import shutil

            

Reported by Pylint.

              dyndep.InitOpsLibrary("@/caffe2/caffe2/distributed:store_ops")


class TestFileStoreHandlerOp(TestCase):
    testCounter = 0

    def setUp(self):
        super(TestFileStoreHandlerOp, self).setUp()
        self.tmpdir = tempfile.mkdtemp()

            

Reported by Pylint.

                  testCounter = 0

    def setUp(self):
        super(TestFileStoreHandlerOp, self).setUp()
        self.tmpdir = tempfile.mkdtemp()

        # Use counter to tell test cases apart
        TestFileStoreHandlerOp.testCounter += 1


            

Reported by Pylint.

              
    def tearDown(self):
        shutil.rmtree(self.tmpdir)
        super(TestFileStoreHandlerOp, self).tearDown()

    def create_store_handler(self):
        # Use new path for every test so they are isolated
        path = self.tmpdir + "/" + str(TestFileStoreHandlerOp.testCounter)


            

Reported by Pylint.

                      shutil.rmtree(self.tmpdir)
        super(TestFileStoreHandlerOp, self).tearDown()

    def create_store_handler(self):
        # Use new path for every test so they are isolated
        path = self.tmpdir + "/" + str(TestFileStoreHandlerOp.testCounter)

        # Ensure path exists (including counter)
        try:

            

Reported by Pylint.

              
        return store_handler

    def test_set_get(self):
        StoreOpsTests.test_set_get(self.create_store_handler)

    def test_get_timeout(self):
        with self.assertRaises(StoreHandlerTimeoutError):
            StoreOpsTests.test_get_timeout(self.create_store_handler)

            

Reported by Pylint.

                  def test_set_get(self):
        StoreOpsTests.test_set_get(self.create_store_handler)

    def test_get_timeout(self):
        with self.assertRaises(StoreHandlerTimeoutError):
            StoreOpsTests.test_get_timeout(self.create_store_handler)

            

Reported by Pylint.