<commit_before>import os
import unittest

import redis


class DummyConnection(object):
    def __init__(self, **kwargs):
        self.kwargs = kwargs
        self.pid = os.getpid()


class ConnectionPoolTestCase(unittest.TestCase):
    def get_pool(self, connection_info=None, max_connections=None):
        connection_info = connection_info or {'a': 1, 'b': 2, 'c': 3}
        pool = redis.ConnectionPool(
            connection_class=DummyConnection, max_connections=max_connections,
            **connection_info)
        return pool

    def test_connection_creation(self):
        connection_info = {'foo': 'bar', 'biz': 'baz'}
        pool = self.get_pool(connection_info=connection_info)
        connection = pool.get_connection('_')
        self.assertEquals(connection.kwargs, connection_info)

    def test_multiple_connections(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assert_(c1 != c2)

    def test_max_connections(self):
        pool = self.get_pool(max_connections=2)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_blocking_max_connections(self):
        pool = self.get_pool(max_connections=2)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_release(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        pool.release(c1)
        c2 = pool.get_connection('_')
        self.assertEquals(c1, c2)


class BlockingConnectionPoolTestCase(unittest.TestCase):
    def get_pool(self, connection_info=None, max_connections=10, timeout=20):
        connection_info = connection_info or {'a': 1, 'b': 2, 'c': 3}
        pool = redis.BlockingConnectionPool(connection_class=DummyConnection,
                                            max_connections=max_connections,
                                            timeout=timeout, **connection_info)
        return pool

    def test_connection_creation(self):
        connection_info = {'foo': 'bar', 'biz': 'baz'}
        pool = self.get_pool(connection_info=connection_info)
        connection = pool.get_connection('_')
        self.assertEquals(connection.kwargs, connection_info)

    def test_multiple_connections(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assert_(c1 != c2)

    def test_max_connections_blocks(self):
        """Getting a connection should block for until available."""

        import time
        from copy import deepcopy
        from threading import Thread

        # We use a queue for cross thread communication within the unit test.
        try: # Python 3
            from queue import Queue
        except ImportError:
            from Queue import Queue

        q = Queue()
        pool = self.get_pool(max_connections=2, timeout=5)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        c3 = 'Not yet got'

        target = lambda: q.put_nowait(pool.get_connection('_'))
        Thread(target=target).start()

        # Blocks while non available.
        time.sleep(0.05)
        self.assertEquals(deepcopy(c3), 'Not yet got')

        # Then got when available.
        pool.release(c1)
        time.sleep(0.05)
        self.assertEquals(c1, q.get_nowait())

    def test_max_connections_timeout(self):
        """Getting a connection raises ``ConnectionError`` after timeout."""

        pool = self.get_pool(max_connections=2, timeout=0.1)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_release(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        pool.release(c1)
        c2 = pool.get_connection('_')
        self.assertEquals(c1, c2)
<commit_msg>tests: tweak blocking connection pool test.<commit_after>import os
import unittest

import redis


class DummyConnection(object):
    def __init__(self, **kwargs):
        self.kwargs = kwargs
        self.pid = os.getpid()


class ConnectionPoolTestCase(unittest.TestCase):
    def get_pool(self, connection_info=None, max_connections=None):
        connection_info = connection_info or {'a': 1, 'b': 2, 'c': 3}
        pool = redis.ConnectionPool(
            connection_class=DummyConnection, max_connections=max_connections,
            **connection_info)
        return pool

    def test_connection_creation(self):
        connection_info = {'foo': 'bar', 'biz': 'baz'}
        pool = self.get_pool(connection_info=connection_info)
        connection = pool.get_connection('_')
        self.assertEquals(connection.kwargs, connection_info)

    def test_multiple_connections(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assert_(c1 != c2)

    def test_max_connections(self):
        pool = self.get_pool(max_connections=2)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_blocking_max_connections(self):
        pool = self.get_pool(max_connections=2)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_release(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        pool.release(c1)
        c2 = pool.get_connection('_')
        self.assertEquals(c1, c2)


class BlockingConnectionPoolTestCase(unittest.TestCase):
    def get_pool(self, connection_info=None, max_connections=10, timeout=20):
        connection_info = connection_info or {'a': 1, 'b': 2, 'c': 3}
        pool = redis.BlockingConnectionPool(connection_class=DummyConnection,
                                            max_connections=max_connections,
                                            timeout=timeout, **connection_info)
        return pool

    def test_connection_creation(self):
        connection_info = {'foo': 'bar', 'biz': 'baz'}
        pool = self.get_pool(connection_info=connection_info)
        connection = pool.get_connection('_')
        self.assertEquals(connection.kwargs, connection_info)

    def test_multiple_connections(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assert_(c1 != c2)

    def test_max_connections_blocks(self):
        """Getting a connection should block for until available."""

        import time
        from copy import deepcopy
        from threading import Thread

        # We use a queue for cross thread communication within the unit test.
        try: # Python 3
            from queue import Queue
        except ImportError:
            from Queue import Queue

        q = Queue()
        q.put_nowait('Not yet got')
        pool = self.get_pool(max_connections=2, timeout=5)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        
        target = lambda: q.put_nowait(pool.get_connection('_'))
        Thread(target=target).start()

        # Blocks while non available.
        time.sleep(0.05)
        c3 = q.get_nowait()
        self.assertEquals(c3, 'Not yet got')
        
        # Then got when available.
        pool.release(c1)
        time.sleep(0.05)
        c3 = q.get_nowait()
        self.assertEquals(c1, c3)

    def test_max_connections_timeout(self):
        """Getting a connection raises ``ConnectionError`` after timeout."""

        pool = self.get_pool(max_connections=2, timeout=0.1)
        c1 = pool.get_connection('_')
        c2 = pool.get_connection('_')
        self.assertRaises(redis.ConnectionError, pool.get_connection, '_')

    def test_release(self):
        pool = self.get_pool()
        c1 = pool.get_connection('_')
        pool.release(c1)
        c2 = pool.get_connection('_')
        self.assertEquals(c1, c2)

{'fraction_non_alphanumeric': 0.10964751798999878, 'fraction_numerical': 0.013416270276863032, 'mean_word_length': 2.588621444201313, '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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '324206', 'n_tokens_mistral': 2476, 'n_tokens_neox': 2366, 'n_words': 498}