from typing import List, Union, Optional

from tools.codegen.context import with_native_function_and_index
from tools.codegen.utils import mapMaybe
from tools.codegen.model import NativeFunction, NativeFunctionsGroup, BackendIndex
from tools.codegen.api.types import kernel_signature
import tools.codegen.api.meta as meta
import tools.codegen.api.structured as structured


            

Reported by Pylint.

              import tools.codegen.api.structured as structured

@with_native_function_and_index
def gen_unstructured(f: NativeFunction, backend_index: BackendIndex) -> Optional[str]:
    sig = kernel_signature(f, backend_index)
    metadata = backend_index.get_kernel(f)
    if metadata is None:
        return None
    if "legacy::" in metadata.kernel:

            

Reported by Pylint.

              import tools.codegen.api.structured as structured

@with_native_function_and_index
def gen_unstructured(f: NativeFunction, backend_index: BackendIndex) -> Optional[str]:
    sig = kernel_signature(f, backend_index)
    metadata = backend_index.get_kernel(f)
    if metadata is None:
        return None
    if "legacy::" in metadata.kernel:

            

Reported by Pylint.

                  metadata = backend_index.get_kernel(f)
    if metadata is None:
        return None
    if "legacy::" in metadata.kernel:
        return None
    else:
        prefix = 'static' if backend_index.external else 'TORCH_API'
        return f"{prefix} {sig.decl(name=metadata.kernel)};"


            

Reported by Pylint.

                      return f"{prefix} {sig.decl(name=metadata.kernel)};"

@with_native_function_and_index
def gen_structured(g: NativeFunctionsGroup, backend_index: BackendIndex) -> List[str]:
    meta_name = meta.name(g)
    out_args = structured.impl_arguments(g)
    metadata = backend_index.get_kernel(g)
    if metadata is None:
        return []

            

Reported by Pylint.

                      return f"{prefix} {sig.decl(name=metadata.kernel)};"

@with_native_function_and_index
def gen_structured(g: NativeFunctionsGroup, backend_index: BackendIndex) -> List[str]:
    meta_name = meta.name(g)
    out_args = structured.impl_arguments(g)
    metadata = backend_index.get_kernel(g)
    if metadata is None:
        return []

            

Reported by Pylint.

              # Generates NativeFunctions.h, a list of forward declarations of all
# actual kernel definitions we keep in aten/src/ATen/native/
@with_native_function_and_index
def compute_native_function_declaration(
        g: Union[NativeFunctionsGroup, NativeFunction],
        backend_index: BackendIndex
) -> List[str]:
    metadata = backend_index.get_kernel(g)
    if isinstance(g, NativeFunctionsGroup):

            

Reported by Pylint.

              # Generates NativeFunctions.h, a list of forward declarations of all
# actual kernel definitions we keep in aten/src/ATen/native/
@with_native_function_and_index
def compute_native_function_declaration(
        g: Union[NativeFunctionsGroup, NativeFunction],
        backend_index: BackendIndex
) -> List[str]:
    metadata = backend_index.get_kernel(g)
    if isinstance(g, NativeFunctionsGroup):

            

Reported by Pylint.

                  metadata = backend_index.get_kernel(g)
    if isinstance(g, NativeFunctionsGroup):
        if metadata is not None and metadata.structured:
            if backend_index.external:
                # Structured hasn't been tested with external backends yet.
                raise AssertionError("Structured external backend functions are not implemented yet.")
            else:
                return gen_structured(g, backend_index)
        else:

            

Reported by Pylint.

                      if metadata is not None and metadata.structured:
            if backend_index.external:
                # Structured hasn't been tested with external backends yet.
                raise AssertionError("Structured external backend functions are not implemented yet.")
            else:
                return gen_structured(g, backend_index)
        else:
            return list(mapMaybe(lambda f: gen_unstructured(f, backend_index), g.functions()))
    else:

            

Reported by Pylint.

              const char* simple_jit = "torch/csrc/deploy/example/generated/simple_jit";

const char* path(const char* envname, const char* path) {
  const char* e = getenv(envname);
  return e ? e : path;
}

TEST(TorchpyTest, LoadLibrary) {
  torch::deploy::InterpreterManager m(1);

            

Reported by FlawFinder.

                return e ? e : path;
}

TEST(TorchpyTest, LoadLibrary) {
  torch::deploy::InterpreterManager m(1);
  torch::deploy::Package p = m.load_package(
      path("LOAD_LIBRARY", "torch/csrc/deploy/example/generated/load_library"));
  auto model = p.load_pickle("fn", "fn.pkl");
  model({});

            

Reported by FlawFinder.

                for (auto& interp : manager.all_instances()) {
    auto I = interp.acquire_session();

    const char* test_lib_path = getenv("LIBTEST_DEPLOY_LIB");
    if (!test_lib_path) {
      I.global("sys", "path").attr("append")({"torch/csrc/deploy"});
      I.global("test_deploy_python", "setup")({getenv("PATH")});
    } else {
      char buf[PATH_MAX];

            

Reported by FlawFinder.

                  const char* test_lib_path = getenv("LIBTEST_DEPLOY_LIB");
    if (!test_lib_path) {
      I.global("sys", "path").attr("append")({"torch/csrc/deploy"});
      I.global("test_deploy_python", "setup")({getenv("PATH")});
    } else {
      char buf[PATH_MAX];
      strncpy(buf, test_lib_path, PATH_MAX);
      dirname(buf);
      I.global("sys", "path").attr("append")({buf});

            

Reported by FlawFinder.

                    I.global("sys", "path").attr("append")({"torch/csrc/deploy"});
      I.global("test_deploy_python", "setup")({getenv("PATH")});
    } else {
      char buf[PATH_MAX];
      strncpy(buf, test_lib_path, PATH_MAX);
      dirname(buf);
      I.global("sys", "path").attr("append")({buf});
    }


            

Reported by FlawFinder.

              
  // Compare all to reference
  for (const auto i : c10::irange(ninterp)) {
    ASSERT_TRUE(ref_output.equal(outputs[i]));
  }

  // test kwargs api with args
  std::vector<c10::IValue> args;
  args.emplace_back(input);

            

Reported by FlawFinder.

                args.emplace_back(input);
  std::unordered_map<std::string, c10::IValue> kwargs_empty;
  auto jit_output_args = model.call_kwargs(args, kwargs_empty).toTensor();
  ASSERT_TRUE(ref_output.equal(jit_output_args));

  // and with kwargs only
  std::unordered_map<std::string, c10::IValue> kwargs;
  kwargs["input"] = input;
  auto jit_output_kwargs = model.call_kwargs(kwargs).toTensor();

            

Reported by FlawFinder.

                std::unordered_map<std::string, c10::IValue> kwargs;
  kwargs["input"] = input;
  auto jit_output_kwargs = model.call_kwargs(kwargs).toTensor();
  ASSERT_TRUE(ref_output.equal(jit_output_kwargs));

  // test hasattr
  ASSERT_TRUE(model.hasattr("forward"));
  ASSERT_FALSE(model.hasattr("make_prediction"));
}

            

Reported by FlawFinder.

              
  // Compare all to reference
  for (const auto i : c10::irange(nthreads)) {
    ASSERT_TRUE(ref_output.equal(outputs[i]));
  }
}

TEST(TorchpyTest, ThrowsSafely) {
  // See explanation in deploy.h

            

Reported by FlawFinder.

              
  // Compare all to reference
  for (const auto i : c10::irange(nthreads)) {
    ASSERT_TRUE(ref_output.equal(outputs[i]));
  }
}

#ifdef TEST_CUSTOM_LIBRARY
thread_local int in_another_module = 5;

            

Reported by FlawFinder.

                      return ((block_size(i), i) for i in range(partitions))

    while True:
        """
        (1) Fix p ∈ [k] with M(P) = bp. So Bp is a maximal block of P.
        """
        # max_size: M(P)
        max_size, p = max(leaderboard())


            

Reported by Pylint.

                      max_size, p = max(leaderboard())

        while True:
            """
            (2) If M(P) ≤ m(P) + 1, then stop.
            """
            # min_size: m(P)
            min_size, q = min(leaderboard())


            

Reported by Pylint.

                          if max_size <= min_size + 1:
                return [sequence[i:j] for i, j in zip([0] + splits[:-1], splits)]

            """
            (3) If M(P) > m(P) + 1, then let m(P) = bq for the q ∈ [k] which is
            closest to p (ties broken arbitrarily). Thus Bq is a minimal block
            of P. Let Bh be the block next to Bq between Bp and Bq. (Note that
            Bh is a non-empty block: if it were, then m(P) = 0 and we should
            have chosen Bh instead of Bq.)

            

Reported by Pylint.

                          have chosen Bh instead of Bq.)
            """
            if p < q:
                """
                So either p < q and then h = q−1 and we define P ∗ by moving
                the last element from Bh = Bq−1 to Bq,
                """
                h = q - 1
                splits[h] -= 1

            

Reported by Pylint.

                              h = q - 1
                splits[h] -= 1
            else:
                """
                or q < p, and then h = q + 1 and P ∗ is obtained by moving the
                first element of Bh = Bq+1 to Bq.
                """
                h = q + 1
                splits[q] += 1

            

Reported by Pylint.

                              h = q + 1
                splits[q] += 1

            """
            Set P = P ∗ . If p = h, then go to (1), else go to (2).
            """
            if p == h:
                break

            

Reported by Pylint.

                  if partitions < 1:
        raise ValueError(f"partitions must be a positive integer ({partitions} < 1)")

    n = len(sequence)
    if n < partitions:
        raise ValueError(f"sequence is shorter than intended partitions ({n} < {partitions})")

    # Normalize the sequence in [0, 1].
    minimum = min(sequence)

            

Reported by Pylint.

                      (1) Fix p ∈ [k] with M(P) = bp. So Bp is a maximal block of P.
        """
        # max_size: M(P)
        max_size, p = max(leaderboard())

        while True:
            """
            (2) If M(P) ≤ m(P) + 1, then stop.
            """

            

Reported by Pylint.

                          (2) If M(P) ≤ m(P) + 1, then stop.
            """
            # min_size: m(P)
            min_size, q = min(leaderboard())

            if max_size <= min_size + 1:
                return [sequence[i:j] for i, j in zip([0] + splits[:-1], splits)]

            """

            

Reported by Pylint.

                              So either p < q and then h = q−1 and we define P ∗ by moving
                the last element from Bh = Bq−1 to Bq,
                """
                h = q - 1
                splits[h] -= 1
            else:
                """
                or q < p, and then h = q + 1 and P ∗ is obtained by moving the
                first element of Bh = Bq+1 to Bq.

            

Reported by Pylint.

              
class TestMiniAlexNet(test_util.TestCase):

    def _MiniAlexNetNoDropout(self, order):
        # First, AlexNet using the cnn wrapper.
        model = model_helper.ModelHelper(name="alexnet")
        conv1 = brew.conv(
            model,
            "data",

            

Reported by Pylint.

              import numpy as np
import unittest

from caffe2.proto import caffe2_pb2
from caffe2.python import (
    workspace,
    device_checker,
    test_util,
    model_helper,

            

Reported by Pylint.

              import numpy as np
import unittest

from caffe2.proto import caffe2_pb2
from caffe2.python import (
    workspace,
    device_checker,
    test_util,
    model_helper,

            

Reported by Pylint.

              )


class TestMiniAlexNet(test_util.TestCase):

    def _MiniAlexNetNoDropout(self, order):
        # First, AlexNet using the cnn wrapper.
        model = model_helper.ModelHelper(name="alexnet")
        conv1 = brew.conv(

            

Reported by Pylint.

              
class TestMiniAlexNet(test_util.TestCase):

    def _MiniAlexNetNoDropout(self, order):
        # First, AlexNet using the cnn wrapper.
        model = model_helper.ModelHelper(name="alexnet")
        conv1 = brew.conv(
            model,
            "data",

            

Reported by Pylint.

              
class TestMiniAlexNet(test_util.TestCase):

    def _MiniAlexNetNoDropout(self, order):
        # First, AlexNet using the cnn wrapper.
        model = model_helper.ModelHelper(name="alexnet")
        conv1 = brew.conv(
            model,
            "data",

            

Reported by Pylint.

              
class TestMiniAlexNet(test_util.TestCase):

    def _MiniAlexNetNoDropout(self, order):
        # First, AlexNet using the cnn wrapper.
        model = model_helper.ModelHelper(name="alexnet")
        conv1 = brew.conv(
            model,
            "data",

            

Reported by Pylint.

                      model.AddGradientOperators([loss])
        return model

    def _testMiniAlexNet(self, order):
        # First, we get all the random initialization of parameters.
        model = self._MiniAlexNetNoDropout(order)
        workspace.ResetWorkspace()
        workspace.RunNetOnce(model.param_init_net)
        inputs = dict(

            

Reported by Pylint.

                      model = self._MiniAlexNetNoDropout(order)
        workspace.ResetWorkspace()
        workspace.RunNetOnce(model.param_init_net)
        inputs = dict(
            [(str(name), workspace.FetchBlob(str(name))) for name in
             model.params]
        )
        if order == "NCHW":
            inputs["data"] = np.random.rand(4, 3, 227, 227).astype(np.float32)

            

Reported by Pylint.

              
    @unittest.skipIf(not workspace.has_gpu_support,
                     "No GPU support. Skipping test.")
    def testMiniAlexNetNCHW(self):
        self._testMiniAlexNet("NCHW")

    # No Group convolution support for NHWC right now
    #@unittest.skipIf(not workspace.has_gpu_support,
    #                 "No GPU support. Skipping test.")

            

Reported by Pylint.

              import torch
import time
import argparse

from common import SubTensor, WithTorchFunction, SubWithTorchFunction

NUM_REPEATS = 1000
NUM_REPEAT_OF_REPEATS = 1000


            

Reported by Pylint.

              

def main():
    global NUM_REPEATS
    global NUM_REPEAT_OF_REPEATS

    parser = argparse.ArgumentParser(
        description="Run the __torch_function__ benchmarks."
    )

            

Reported by Pylint.

              
def main():
    global NUM_REPEATS
    global NUM_REPEAT_OF_REPEATS

    parser = argparse.ArgumentParser(
        description="Run the __torch_function__ benchmarks."
    )
    parser.add_argument(

            

Reported by Pylint.

              import torch
import time
import argparse

from common import SubTensor, WithTorchFunction, SubWithTorchFunction

NUM_REPEATS = 1000
NUM_REPEAT_OF_REPEATS = 1000


            

Reported by Pylint.

              import torch
import time
import argparse

from common import SubTensor, WithTorchFunction, SubWithTorchFunction

NUM_REPEATS = 1000
NUM_REPEAT_OF_REPEATS = 1000


            

Reported by Pylint.

              import torch
import time
import argparse

from common import SubTensor, WithTorchFunction, SubWithTorchFunction

NUM_REPEATS = 1000
NUM_REPEAT_OF_REPEATS = 1000


            

Reported by Pylint.

              NUM_REPEAT_OF_REPEATS = 1000


def bench(t1, t2):
    bench_times = []
    for _ in range(NUM_REPEAT_OF_REPEATS):
        time_start = time.time()
        for _ in range(NUM_REPEATS):
            torch.add(t1, t2)

            

Reported by Pylint.

              NUM_REPEAT_OF_REPEATS = 1000


def bench(t1, t2):
    bench_times = []
    for _ in range(NUM_REPEAT_OF_REPEATS):
        time_start = time.time()
        for _ in range(NUM_REPEATS):
            torch.add(t1, t2)

            

Reported by Pylint.

              NUM_REPEAT_OF_REPEATS = 1000


def bench(t1, t2):
    bench_times = []
    for _ in range(NUM_REPEAT_OF_REPEATS):
        time_start = time.time()
        for _ in range(NUM_REPEATS):
            torch.add(t1, t2)

            

Reported by Pylint.

                  return bench_time, bench_std


def main():
    global NUM_REPEATS
    global NUM_REPEAT_OF_REPEATS

    parser = argparse.ArgumentParser(
        description="Run the __torch_function__ benchmarks."

            

Reported by Pylint.

              




import numpy as np
import os
import tempfile


            

Reported by Pylint.

              

import numpy as np
import os
import tempfile

from caffe2.python import test_util, workspace
import caffe2.python.models.seq2seq.seq2seq_util as seq2seq_util
from caffe2.python.models.seq2seq.train import Seq2SeqModelCaffe2

            

Reported by Pylint.

              
import numpy as np
import os
import tempfile

from caffe2.python import test_util, workspace
import caffe2.python.models.seq2seq.seq2seq_util as seq2seq_util
from caffe2.python.models.seq2seq.train import Seq2SeqModelCaffe2
from caffe2.python.models.seq2seq.translate import (

            

Reported by Pylint.

              )


class Seq2SeqBeamSearchTest(test_util.TestCase):

    def _build_seq2seq_model(
        self,
        model_params,
        tmp_dir,

            

Reported by Pylint.

              
class Seq2SeqBeamSearchTest(test_util.TestCase):

    def _build_seq2seq_model(
        self,
        model_params,
        tmp_dir,
        source_vocab_size=20,
        target_vocab_size=20,

            

Reported by Pylint.

              
class Seq2SeqBeamSearchTest(test_util.TestCase):

    def _build_seq2seq_model(
        self,
        model_params,
        tmp_dir,
        source_vocab_size=20,
        target_vocab_size=20,

            

Reported by Pylint.

              
        return model_obj, checkpoint_path

    def _run_compare_train_inference(self, model_params):
        tmp_dir = tempfile.mkdtemp()

        model_obj, checkpoint_path = self._build_seq2seq_model(
            model_params,
            tmp_dir=tmp_dir,

            

Reported by Pylint.

                          num_gpus=0,
            batch_size=2,
        )
        assert model_obj is not None

        translate_params = dict(
            ensemble_models=[dict(
                source_vocab={i: str(i) for i in range(20)},
                target_vocab={i: str(i) for i in range(20)},

            

Reported by Bandit.

                              decimal=4,
            )

    def test_attention(self):
        model_params = dict(
            attention='regular',
            decoder_layer_configs=[
                dict(
                    num_units=32,

            

Reported by Pylint.

                      )
        self._run_compare_train_inference(model_params)

    def test_2layer_attention(self):
        model_params = dict(
            attention='regular',
            decoder_layer_configs=[
                dict(
                    num_units=32,

            

Reported by Pylint.

              from cimodel.data.simple.util.versions import MultiPartVersion
import cimodel.lib.miniutils as miniutils

XCODE_VERSION = MultiPartVersion([12, 0, 0])


class ArchVariant:
    def __init__(self, name, custom_build_name=""):
        self.name = name

            

Reported by Pylint.

              XCODE_VERSION = MultiPartVersion([12, 0, 0])


class ArchVariant:
    def __init__(self, name, custom_build_name=""):
        self.name = name
        self.custom_build_name = custom_build_name

    def render(self):

            

Reported by Pylint.

              XCODE_VERSION = MultiPartVersion([12, 0, 0])


class ArchVariant:
    def __init__(self, name, custom_build_name=""):
        self.name = name
        self.custom_build_name = custom_build_name

    def render(self):

            

Reported by Pylint.

                      self.name = name
        self.custom_build_name = custom_build_name

    def render(self):
        extra_parts = [self.custom_build_name] if len(self.custom_build_name) > 0 else []
        return "_".join([self.name] + extra_parts)


def get_platform(arch_variant_name):

            

Reported by Pylint.

                      return "_".join([self.name] + extra_parts)


def get_platform(arch_variant_name):
    return "SIMULATOR" if arch_variant_name == "x86_64" else "OS"


class IOSJob:
    def __init__(self, xcode_version, arch_variant, is_org_member_context=True, extra_props=None):

            

Reported by Pylint.

                  return "SIMULATOR" if arch_variant_name == "x86_64" else "OS"


class IOSJob:
    def __init__(self, xcode_version, arch_variant, is_org_member_context=True, extra_props=None):
        self.xcode_version = xcode_version
        self.arch_variant = arch_variant
        self.is_org_member_context = is_org_member_context
        self.extra_props = extra_props

            

Reported by Pylint.

                      self.is_org_member_context = is_org_member_context
        self.extra_props = extra_props

    def gen_name_parts(self, with_version_dots):

        version_parts = self.xcode_version.render_dots_or_parts(with_version_dots)
        build_variant_suffix = "_".join([self.arch_variant.render(), "build"])

        return [

            

Reported by Pylint.

                          build_variant_suffix,
        ]

    def gen_job_name(self):
        return "_".join(self.gen_name_parts(False))

    def gen_tree(self):

        platform_name = get_platform(self.arch_variant.name)

            

Reported by Pylint.

                  def gen_job_name(self):
        return "_".join(self.gen_name_parts(False))

    def gen_tree(self):

        platform_name = get_platform(self.arch_variant.name)

        props_dict = {
            "build_environment": "-".join(self.gen_name_parts(True)),

            

Reported by Pylint.

              WORKFLOW_DATA = [
    IOSJob(XCODE_VERSION, ArchVariant("x86_64"), is_org_member_context=False, extra_props={
        "lite_interpreter": miniutils.quote(str(int(True)))}),
    IOSJob(XCODE_VERSION, ArchVariant("x86_64", "full_jit"), is_org_member_context=False, extra_props={
        "lite_interpreter": miniutils.quote(str(int(False)))}),
    IOSJob(XCODE_VERSION, ArchVariant("arm64"), extra_props={
        "lite_interpreter": miniutils.quote(str(int(True)))}),
    IOSJob(XCODE_VERSION, ArchVariant("arm64", "metal"), extra_props={
        "use_metal": miniutils.quote(str(int(True))),

            

Reported by Pylint.

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





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

            

Reported by Pylint.

              import numpy as np


class BatchSoftmaxLoss(ModelLayer):
    def __init__(
        self,
        model,
        input_record,
        name='batch_softmax_loss',

            

Reported by Pylint.

              

class BatchSoftmaxLoss(ModelLayer):
    def __init__(
        self,
        model,
        input_record,
        name='batch_softmax_loss',
        label_smoothing_matrix=None,

            

Reported by Pylint.

                      average_by_batch_size=False,
        **kwargs
    ):
        super(BatchSoftmaxLoss, self).__init__(
            model, name, input_record, **kwargs)

        assert schema.is_schema_subset(
            schema.Struct(
                ('label', schema.Scalar()),

            

Reported by Pylint.

                      super(BatchSoftmaxLoss, self).__init__(
            model, name, input_record, **kwargs)

        assert schema.is_schema_subset(
            schema.Struct(
                ('label', schema.Scalar()),
                ('prediction', schema.Scalar()),
            ),
            input_record

            

Reported by Bandit.

                          ),
        )

    def initialize_label_smoothing_constants(self):
        assert self.label_smoothing_matrix is not None
        self.label_smoothing_matrix = np.array(
            self.label_smoothing_matrix).astype(np.float32)
        assert len(self.label_smoothing_matrix.shape) == 2
        label_dim = self.label_smoothing_matrix.shape[0]

            

Reported by Pylint.

                      )

    def initialize_label_smoothing_constants(self):
        assert self.label_smoothing_matrix is not None
        self.label_smoothing_matrix = np.array(
            self.label_smoothing_matrix).astype(np.float32)
        assert len(self.label_smoothing_matrix.shape) == 2
        label_dim = self.label_smoothing_matrix.shape[0]
        assert label_dim == self.label_smoothing_matrix.shape[1]

            

Reported by Bandit.

                      assert self.label_smoothing_matrix is not None
        self.label_smoothing_matrix = np.array(
            self.label_smoothing_matrix).astype(np.float32)
        assert len(self.label_smoothing_matrix.shape) == 2
        label_dim = self.label_smoothing_matrix.shape[0]
        assert label_dim == self.label_smoothing_matrix.shape[1]

        self.label_smoothing_matrix = self.model.add_global_constant(
            '%s_label_smoothing_matrix' % self.name,

            

Reported by Bandit.

                          self.label_smoothing_matrix).astype(np.float32)
        assert len(self.label_smoothing_matrix.shape) == 2
        label_dim = self.label_smoothing_matrix.shape[0]
        assert label_dim == self.label_smoothing_matrix.shape[1]

        self.label_smoothing_matrix = self.model.add_global_constant(
            '%s_label_smoothing_matrix' % self.name,
            array=self.label_smoothing_matrix,
            dtype=np.dtype(np.float32),

            

Reported by Bandit.

                      # but when used in label smoothing, the label must be in probabilities
        self.label_prob = True

    def compute_smoothed_label(self, net):
        assert self.label_smoothing_matrix is not None
        label = self.input_record.label()
        original_label_type = self.input_record.label.field_type()
        if original_label_type.base != np.int64:
            int64_label = net.NextScopedBlob('int64_label')

            

Reported by Pylint.

              import os
import uuid

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:redis_store_handler_ops")

            

Reported by Pylint.

              import os
import uuid

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:redis_store_handler_ops")

            

Reported by Pylint.

                      super(TestRedisStoreHandlerOp, self).setUp()
        self.uuid = str(uuid.uuid4()) + "/"

    def tearDown(self):
        super(TestRedisStoreHandlerOp, self).tearDown()

    def create_store_handler(self):
        store_handler = "store_handler"
        workspace.RunOperatorOnce(

            

Reported by Pylint.

                              [store_handler],
                prefix=self.uuid,
                host=os.getenv("REDIS_HOST", "localhost"),
                port=int(os.getenv("REDIS_PORT", 6379))))
        return store_handler

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


            

Reported by Pylint.

              




import os
import uuid

from caffe2.distributed.python import StoreHandlerTimeoutError

            

Reported by Pylint.

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


class TestRedisStoreHandlerOp(TestCase):
    def setUp(self):
        super(TestRedisStoreHandlerOp, self).setUp()
        self.uuid = str(uuid.uuid4()) + "/"

    def tearDown(self):

            

Reported by Pylint.

              
class TestRedisStoreHandlerOp(TestCase):
    def setUp(self):
        super(TestRedisStoreHandlerOp, self).setUp()
        self.uuid = str(uuid.uuid4()) + "/"

    def tearDown(self):
        super(TestRedisStoreHandlerOp, self).tearDown()


            

Reported by Pylint.

                      self.uuid = str(uuid.uuid4()) + "/"

    def tearDown(self):
        super(TestRedisStoreHandlerOp, self).tearDown()

    def create_store_handler(self):
        store_handler = "store_handler"
        workspace.RunOperatorOnce(
            core.CreateOperator(

            

Reported by Pylint.

                  def tearDown(self):
        super(TestRedisStoreHandlerOp, self).tearDown()

    def create_store_handler(self):
        store_handler = "store_handler"
        workspace.RunOperatorOnce(
            core.CreateOperator(
                "RedisStoreHandlerCreate",
                [],

            

Reported by Pylint.

                              port=int(os.getenv("REDIS_PORT", 6379))))
        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.

                char *path = NULL, *str = NULL;
  int i;

  if((str=getenv("LIBOPENCL_SO_PATH")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                if((str=getenv("LIBOPENCL_SO_PATH")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_4")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_4")) && access_file(str)) {
    path = str;
  }

  if(!path)
  {

            

Reported by FlawFinder.

                char *path = NULL, *str = NULL;
  int i;

  if((str=getenv("LIBOPENCL_SO_PATH")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                if((str=getenv("LIBOPENCL_SO_PATH")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                else if((str=getenv("LIBOPENCL_SO_PATH_2")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_4")) && access_file(str)) {
    path = str;
  }

            

Reported by FlawFinder.

                else if((str=getenv("LIBOPENCL_SO_PATH_3")) && access_file(str)) {
    path = str;
  }
  else if((str=getenv("LIBOPENCL_SO_PATH_4")) && access_file(str)) {
    path = str;
  }

  if(!path)
  {

            

Reported by FlawFinder.

                  {
      if(access_file(default_so_paths[i]))
      {
        path = (char *) default_so_paths[i];
        break;
      }
    }
  }


            

Reported by FlawFinder.

                  {
      if(access_file(default_so_paths[i]))
      {
        path = (char *) default_so_paths[i];
        break;
      }
    }
  }


            

Reported by FlawFinder.