<commit_before>import threading
import unittest

import mock
import numpy
import six

import chainer
from chainer import cuda
import chainer.functions as F
from chainer import testing
from chainer.testing import attr
from chainer.utils import type_check


class TestFunction(unittest.TestCase):

    def _get_method(self, prefix, gpu):
        suffix = 'gpu' if gpu else 'cpu'
        return getattr(self.f, prefix + '_' + suffix)

    def setUp(self):
        y1 = numpy.arange(4).astype(numpy.float32)
        y2 = numpy.arange(4).astype(numpy.float32) + 1
        gx1 = numpy.arange(3).astype(numpy.float32)
        gx2 = None
        gy1 = numpy.arange(4).astype(numpy.float32)
        gy2 = numpy.arange(4).astype(numpy.float32)

        f = chainer.Function()
        f.check_type_forward = mock.MagicMock()
        f.forward_cpu = mock.MagicMock(return_value=(y1, y2))
        f.forward_gpu = mock.MagicMock()
        f.backward_cpu = mock.MagicMock(return_value=(gx1, gx2))
        f.backward_gpu = mock.MagicMock()
        self.f = f

        self.x1 = numpy.arange(3).astype(numpy.float32)
        self.x2 = numpy.arange(3).astype(numpy.int32)
        self.y1 = y1
        self.y2 = y2
        self.gx1 = gx1
        self.gx2 = gx2
        self.gy1 = gy1
        self.gy2 = gy2

    def tearDown(self):
        # Set None to delete cuda array
        self.f = None
        self.y1 = None
        self.y2 = None
        self.gx1 = None

    def setup_gpu(self):
        self.x1 = cuda.to_gpu(self.x1)
        self.x2 = cuda.to_gpu(self.x2)
        self.y1 = cuda.to_gpu(self.y1)
        self.y2 = cuda.to_gpu(self.y2)
        self.gx1 = cuda.to_gpu(self.gx1)
        self.gx2 = None
        self.gy1 = cuda.to_gpu(self.gy1)
        self.gy2 = cuda.to_gpu(self.gy2)
        self.f.forward_gpu = mock.MagicMock(return_value=(self.y1, self.y2))
        self.f.backward_gpu = mock.MagicMock(return_value=(self.gx1, self.gx2))

    def check_forward(self, gpu):
        y1, y2 = self.f.forward((self.x1, self.x2))
        self.assertEqual(self.f.check_type_forward.call_count, 0)
        self.assertEqual(self._get_method('forward', not gpu).call_count, 0)
        self._get_method('forward', gpu).assert_called_once_with(
            (self.x1, self.x2))
        self.assertTrue((cuda.to_cpu(y1) == cuda.to_cpu(self.y1)).all())
        self.assertTrue((cuda.to_cpu(y2) == cuda.to_cpu(self.y2)).all())

    def test_forward_cpu(self):
        self.check_forward(False)

    @attr.gpu
    def test_forward_gpu(self):
        self.setup_gpu()
        self.check_forward(True)

    def check_backward(self, gpu):
        gx1, gx2 = self.f.backward((self.x1, self.x2), (self.gy1, self.gy2))
        self.assertEqual(self._get_method('backward', not gpu).call_count, 0)
        self._get_method('backward', gpu).assert_called_once_with(
            (self.x1, self.x2), (self.gy1, self.gy2))
        self.assertTrue((cuda.to_cpu(gx1) == cuda.to_cpu(self.gx1)).all())
        self.assertIsNone(gx2)

    def test_backward_cpu(self):
        self.check_backward(False)

    @attr.gpu
    def test_backward_gpu(self):
        self.setup_gpu()
        self.check_backward(True)

    def check_check_type_forward(self):
        self.assertEqual(self.f.check_type_forward.call_count, 1)
        ts = self.f.check_type_forward.call_args[0][0]
        self.assertIsInstance(ts, type_check.TypeInfoTuple)
        self.assertEqual(len(ts), 2)

        self.assertEqual(ts[0].name, 'in_types[0]')
        t1 = ts[0].eval()
        self.assertEqual(t1.shape, (3,))
        self.assertEqual(t1.dtype, numpy.float32)

        self.assertEqual(ts[1].name, 'in_types[1]')
        t2 = ts[1].eval()
        self.assertEqual(t2.shape, (3,))
        self.assertEqual(t2.dtype, numpy.int32)

    def check_call(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        x1.rank = 1
        x2.rank = 3
        ys = self.f(x1, x2)

        self.assertEqual(len(ys), 2)
        self.check_check_type_forward()

        for y in ys:
            self.assertIsInstance(y, chainer.Variable)
            # rank is (maximum rank in xs) + 1
            self.assertEqual(y.rank, 4)
            self.assertIs(y.creator, self.f)

        self.assertIsInstance(y.creator.outputs, tuple)

    def test_call_cpu(self):
        self.check_call()

    @attr.gpu
    def test_call_gpu(self):
        self.setup_gpu()
        self.check_call()

    def check_call_ndarray(self):
        x1 = chainer.Variable(self.x1)
        x2 = self.x2
        x1.rank = 1
        ys = self.f(x1, x2)

        self.assertEqual(len(ys), 2)
        self.check_check_type_forward()

        for y in ys:
            self.assertIsInstance(y, chainer.Variable)
            # rank is (maximum rank in xs) + 1
            self.assertEqual(y.rank, 2)
            self.assertIs(y.creator, self.f)

        self.assertIsInstance(y.creator.outputs, tuple)

    def test_call_ndarray_cpu(self):
        self.check_call_ndarray()

    @attr.gpu
    def test_call_ndarray_gpu(self):
        self.setup_gpu()
        self.check_call_ndarray()

    def check_call_single_return_value(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        ret = self.f(x1, x2)
        self.assertIsInstance(ret, chainer.Variable)

    def test_call_single_return_value_cpu(self):
        self.f.forward_cpu.return_value = (cuda.to_cpu(self.y1),)
        self.check_call_single_return_value()

    @attr.gpu
    def test_call_single_return_value_gpu(self):
        self.setup_gpu()
        self.f.forward_gpu.return_value = (cuda.to_gpu(self.y1),)
        self.check_call_single_return_value()

    def check_call_mixed_volatile(self):
        with chainer.no_backprop_mode():
            x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        with self.assertRaises(ValueError):
            self.f(x1, x2)

    def test_call_mixed_volatile_cpu(self):
        self.check_call_mixed_volatile()

    @attr.gpu
    def test_call_mixed_volatile_gpu(self):
        self.setup_gpu()
        self.check_call_mixed_volatile()

    def _get_f(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        y1, y2 = self.f(x1, x2)

        f = y1.creator
        # To test weak refernece, return only x1 and y1.
        # x2 and y2 are deleted by the garbage collector<commit_msg>Remove test using mixed volatility<commit_after>import threading
import unittest

import mock
import numpy
import six

import chainer
from chainer import cuda
import chainer.functions as F
from chainer import testing
from chainer.testing import attr
from chainer.utils import type_check


class TestFunction(unittest.TestCase):

    def _get_method(self, prefix, gpu):
        suffix = 'gpu' if gpu else 'cpu'
        return getattr(self.f, prefix + '_' + suffix)

    def setUp(self):
        y1 = numpy.arange(4).astype(numpy.float32)
        y2 = numpy.arange(4).astype(numpy.float32) + 1
        gx1 = numpy.arange(3).astype(numpy.float32)
        gx2 = None
        gy1 = numpy.arange(4).astype(numpy.float32)
        gy2 = numpy.arange(4).astype(numpy.float32)

        f = chainer.Function()
        f.check_type_forward = mock.MagicMock()
        f.forward_cpu = mock.MagicMock(return_value=(y1, y2))
        f.forward_gpu = mock.MagicMock()
        f.backward_cpu = mock.MagicMock(return_value=(gx1, gx2))
        f.backward_gpu = mock.MagicMock()
        self.f = f

        self.x1 = numpy.arange(3).astype(numpy.float32)
        self.x2 = numpy.arange(3).astype(numpy.int32)
        self.y1 = y1
        self.y2 = y2
        self.gx1 = gx1
        self.gx2 = gx2
        self.gy1 = gy1
        self.gy2 = gy2

    def tearDown(self):
        # Set None to delete cuda array
        self.f = None
        self.y1 = None
        self.y2 = None
        self.gx1 = None

    def setup_gpu(self):
        self.x1 = cuda.to_gpu(self.x1)
        self.x2 = cuda.to_gpu(self.x2)
        self.y1 = cuda.to_gpu(self.y1)
        self.y2 = cuda.to_gpu(self.y2)
        self.gx1 = cuda.to_gpu(self.gx1)
        self.gx2 = None
        self.gy1 = cuda.to_gpu(self.gy1)
        self.gy2 = cuda.to_gpu(self.gy2)
        self.f.forward_gpu = mock.MagicMock(return_value=(self.y1, self.y2))
        self.f.backward_gpu = mock.MagicMock(return_value=(self.gx1, self.gx2))

    def check_forward(self, gpu):
        y1, y2 = self.f.forward((self.x1, self.x2))
        self.assertEqual(self.f.check_type_forward.call_count, 0)
        self.assertEqual(self._get_method('forward', not gpu).call_count, 0)
        self._get_method('forward', gpu).assert_called_once_with(
            (self.x1, self.x2))
        self.assertTrue((cuda.to_cpu(y1) == cuda.to_cpu(self.y1)).all())
        self.assertTrue((cuda.to_cpu(y2) == cuda.to_cpu(self.y2)).all())

    def test_forward_cpu(self):
        self.check_forward(False)

    @attr.gpu
    def test_forward_gpu(self):
        self.setup_gpu()
        self.check_forward(True)

    def check_backward(self, gpu):
        gx1, gx2 = self.f.backward((self.x1, self.x2), (self.gy1, self.gy2))
        self.assertEqual(self._get_method('backward', not gpu).call_count, 0)
        self._get_method('backward', gpu).assert_called_once_with(
            (self.x1, self.x2), (self.gy1, self.gy2))
        self.assertTrue((cuda.to_cpu(gx1) == cuda.to_cpu(self.gx1)).all())
        self.assertIsNone(gx2)

    def test_backward_cpu(self):
        self.check_backward(False)

    @attr.gpu
    def test_backward_gpu(self):
        self.setup_gpu()
        self.check_backward(True)

    def check_check_type_forward(self):
        self.assertEqual(self.f.check_type_forward.call_count, 1)
        ts = self.f.check_type_forward.call_args[0][0]
        self.assertIsInstance(ts, type_check.TypeInfoTuple)
        self.assertEqual(len(ts), 2)

        self.assertEqual(ts[0].name, 'in_types[0]')
        t1 = ts[0].eval()
        self.assertEqual(t1.shape, (3,))
        self.assertEqual(t1.dtype, numpy.float32)

        self.assertEqual(ts[1].name, 'in_types[1]')
        t2 = ts[1].eval()
        self.assertEqual(t2.shape, (3,))
        self.assertEqual(t2.dtype, numpy.int32)

    def check_call(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        x1.rank = 1
        x2.rank = 3
        ys = self.f(x1, x2)

        self.assertEqual(len(ys), 2)
        self.check_check_type_forward()

        for y in ys:
            self.assertIsInstance(y, chainer.Variable)
            # rank is (maximum rank in xs) + 1
            self.assertEqual(y.rank, 4)
            self.assertIs(y.creator, self.f)

        self.assertIsInstance(y.creator.outputs, tuple)

    def test_call_cpu(self):
        self.check_call()

    @attr.gpu
    def test_call_gpu(self):
        self.setup_gpu()
        self.check_call()

    def check_call_ndarray(self):
        x1 = chainer.Variable(self.x1)
        x2 = self.x2
        x1.rank = 1
        ys = self.f(x1, x2)

        self.assertEqual(len(ys), 2)
        self.check_check_type_forward()

        for y in ys:
            self.assertIsInstance(y, chainer.Variable)
            # rank is (maximum rank in xs) + 1
            self.assertEqual(y.rank, 2)
            self.assertIs(y.creator, self.f)

        self.assertIsInstance(y.creator.outputs, tuple)

    def test_call_ndarray_cpu(self):
        self.check_call_ndarray()

    @attr.gpu
    def test_call_ndarray_gpu(self):
        self.setup_gpu()
        self.check_call_ndarray()

    def check_call_single_return_value(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        ret = self.f(x1, x2)
        self.assertIsInstance(ret, chainer.Variable)

    def test_call_single_return_value_cpu(self):
        self.f.forward_cpu.return_value = (cuda.to_cpu(self.y1),)
        self.check_call_single_return_value()

    @attr.gpu
    def test_call_single_return_value_gpu(self):
        self.setup_gpu()
        self.f.forward_gpu.return_value = (cuda.to_gpu(self.y1),)
        self.check_call_single_return_value()

    def _get_f(self):
        x1 = chainer.Variable(self.x1)
        x2 = chainer.Variable(self.x2)
        y1, y2 = self.f(x1, x2)

        f = y1.creator
        # To test weak refernece, return only x1 and y1.
        # x2 and y2 are deleted by the garbage collector
{'fraction_non_alphanumeric': 0.12209302325581395, 'fraction_numerical': 0.02648578811369509, 'mean_word_length': 2.3885088919288644, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4135661', 'n_tokens_mistral': 4616, 'n_tokens_neox': 4470, 'n_words': 737}