<commit_before>import threading
import unittest

import mock
from nose.tools import eq_

from kazoo.client import Callback


class TestThreadingHandler(unittest.TestCase):
    def _makeOne(self, *args):
        from kazoo.handlers.threading import SequentialThreadingHandler
        return SequentialThreadingHandler(*args)

    def _getAsync(self, *args):
        from kazoo.handlers.threading import AsyncResult
        return AsyncResult

    def test_completion_vs_session(self):
        h = self._makeOne()

        lst = []
        av = threading.Event()
        bv = threading.Event()
        cv = threading.Event()
        dv = threading.Event()

        def addToList():
            av.set()
            bv.wait()
            lst.append(True)
            dv.set()

        def anotherAdd():
            lst.append(True)
            cv.set()

        call1 = Callback('session', addToList, ())
        call2 = Callback('completion', anotherAdd, ())

        h.dispatch_callback(call1)
        av.wait()
        # Now we know the first is waiting, make sure
        # the second executes while the first has blocked
        # its thread
        h.dispatch_callback(call2)
        cv.wait()

        eq_(lst, [True])
        bv.set()
        dv.wait()
        eq_(lst, [True, True])

    def test_proper_threading(self):
        h = self._makeOne()
        assert isinstance(h.event_object(), threading._Event)

    def test_matching_async(self):
        h = self._makeOne()
        async = self._getAsync()
        assert isinstance(h.async_result(), async)


class TestThreadingAsync(unittest.TestCase):
    def _makeOne(self, *args):
        from kazoo.handlers.threading import AsyncResult
        return AsyncResult(*args)

    def test_ready(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        eq_(async.ready(), False)
        async.set('val')
        eq_(async.ready(), True)
        eq_(async.successful(), True)
        eq_(async.exception, None)

    def test_callback_queued(self):
        mock_handler = mock.Mock()
        mock_handler.completion_queue = mock.Mock()
        async = self._makeOne(mock_handler)

        async.rawlink(lambda a: a)
        async.set('val')

        assert mock_handler.completion_queue.put.called

    def test_set_exception(self):
        mock_handler = mock.Mock()
        mock_handler.completion_queue = mock.Mock()
        async = self._makeOne(mock_handler)
        async.rawlink(lambda a: a)
        async.set_exception(ImportError('Error occured'))

        assert isinstance(async.exception, ImportError)
        assert mock_handler.completion_queue.put.called
<commit_msg>More threaded handler tests.<commit_after>import threading
import unittest

import mock
from nose.tools import eq_

from kazoo.client import Callback


class TestThreadingHandler(unittest.TestCase):
    def _makeOne(self, *args):
        from kazoo.handlers.threading import SequentialThreadingHandler
        return SequentialThreadingHandler(*args)

    def _getAsync(self, *args):
        from kazoo.handlers.threading import AsyncResult
        return AsyncResult

    def test_completion_vs_session(self):
        h = self._makeOne()

        lst = []
        av = threading.Event()
        bv = threading.Event()
        cv = threading.Event()
        dv = threading.Event()

        def addToList():
            av.set()
            bv.wait()
            lst.append(True)
            dv.set()

        def anotherAdd():
            lst.append(True)
            cv.set()

        call1 = Callback('session', addToList, ())
        call2 = Callback('completion', anotherAdd, ())

        h.dispatch_callback(call1)
        av.wait()
        # Now we know the first is waiting, make sure
        # the second executes while the first has blocked
        # its thread
        h.dispatch_callback(call2)
        cv.wait()

        eq_(lst, [True])
        bv.set()
        dv.wait()
        eq_(lst, [True, True])

    def test_proper_threading(self):
        h = self._makeOne()
        assert isinstance(h.event_object(), threading._Event)

    def test_matching_async(self):
        h = self._makeOne()
        async = self._getAsync()
        assert isinstance(h.async_result(), async)


class TestThreadingAsync(unittest.TestCase):
    def _makeOne(self, *args):
        from kazoo.handlers.threading import AsyncResult
        return AsyncResult(*args)

    def test_ready(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        eq_(async.ready(), False)
        async.set('val')
        eq_(async.ready(), True)
        eq_(async.successful(), True)
        eq_(async.exception, None)

    def test_callback_queued(self):
        mock_handler = mock.Mock()
        mock_handler.completion_queue = mock.Mock()
        async = self._makeOne(mock_handler)

        async.rawlink(lambda a: a)
        async.set('val')

        assert mock_handler.completion_queue.put.called

    def test_set_exception(self):
        mock_handler = mock.Mock()
        mock_handler.completion_queue = mock.Mock()
        async = self._makeOne(mock_handler)
        async.rawlink(lambda a: a)
        async.set_exception(ImportError('Error occured'))

        assert isinstance(async.exception, ImportError)
        assert mock_handler.completion_queue.put.called

    def test_get_wait_while_setting(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        lst = []
        bv = threading.Event()
        cv = threading.Event()

        def wait_for_val():
            bv.set()
            val = async.get()
            lst.append(val)
            cv.set()
        th = threading.Thread(target=wait_for_val)
        th.start()
        bv.wait()

        async.set('fred')
        cv.wait()
        eq_(lst, ['fred'])

    def test_get_with_exception(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        lst = []
        bv = threading.Event()
        cv = threading.Event()

        def wait_for_val():
            bv.set()
            try:
                val = async.get()
            except ImportError:
                lst.append('oops')
            else:
                lst.append(val)
            cv.set()
        th = threading.Thread(target=wait_for_val)
        th.start()
        bv.wait()

        async.set_exception(ImportError)
        cv.wait()
        eq_(lst, ['oops'])

    def test_set_before_wait(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        lst = []
        cv = threading.Event()
        async.set('fred')

        def wait_for_val():
            val = async.get()
            lst.append(val)
            cv.set()
        th = threading.Thread(target=wait_for_val)
        th.start()
        cv.wait()
        eq_(lst, ['fred'])

    def test_set_exc_before_wait(self):
        mock_handler = mock.Mock()
        async = self._makeOne(mock_handler)

        lst = []
        cv = threading.Event()
        async.set_exception(ImportError)

        def wait_for_val():
            try:
                val = async.get()
            except ImportError:
                lst.append('ooops')
            else:
                lst.append(val)
            cv.set()
        th = threading.Thread(target=wait_for_val)
        th.start()
        cv.wait()
        eq_(lst, ['ooops'])


{'fraction_non_alphanumeric': 0.10670075349838536, 'fraction_numerical': 0.001076426264800861, 'mean_word_length': 2.0264657980456025, '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': '2476251', 'n_tokens_mistral': 2226, 'n_tokens_neox': 2129, 'n_words': 443}