connection_pool.py
Go to the documentation of this file.
00001 import os
00002 import unittest
00003 
00004 import redis
00005 
00006 
00007 class DummyConnection(object):
00008     def __init__(self, **kwargs):
00009         self.kwargs = kwargs
00010         self.pid = os.getpid()
00011 
00012 
00013 class ConnectionPoolTestCase(unittest.TestCase):
00014     def get_pool(self, connection_info=None, max_connections=None):
00015         connection_info = connection_info or {'a': 1, 'b': 2, 'c': 3}
00016         pool = redis.ConnectionPool(
00017             connection_class=DummyConnection, max_connections=max_connections,
00018             **connection_info)
00019         return pool
00020 
00021     def test_connection_creation(self):
00022         connection_info = {'foo': 'bar', 'biz': 'baz'}
00023         pool = self.get_pool(connection_info=connection_info)
00024         connection = pool.get_connection('_')
00025         self.assertEquals(connection.kwargs, connection_info)
00026 
00027     def test_multiple_connections(self):
00028         pool = self.get_pool()
00029         c1 = pool.get_connection('_')
00030         c2 = pool.get_connection('_')
00031         self.assert_(c1 != c2)
00032 
00033     def test_max_connections(self):
00034         pool = self.get_pool(max_connections=2)
00035         c1 = pool.get_connection('_')
00036         c2 = pool.get_connection('_')
00037         self.assertRaises(redis.ConnectionError, pool.get_connection, '_')
00038 
00039     def test_release(self):
00040         pool = self.get_pool()
00041         c1 = pool.get_connection('_')
00042         pool.release(c1)
00043         c2 = pool.get_connection('_')
00044         self.assertEquals(c1, c2)
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Properties Friends


redis
Author(s): Daniel Stonier
autogenerated on Tue Jan 15 2013 17:43:13