54 lines
1.7 KiB
Python
Executable File
54 lines
1.7 KiB
Python
Executable File
import redis
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
class ConnectionPoolTestCase(unittest.TestCase):
|
|
def test_multiple_connections(self):
|
|
# 2 clients to the same host/port/db/pool should use the same connection
|
|
pool = redis.ConnectionPool()
|
|
r1 = redis.Redis(host='localhost', port=6379, db=9, connection_pool=pool)
|
|
r2 = redis.Redis(host='localhost', port=6379, db=9, connection_pool=pool)
|
|
self.assertEquals(r1.connection, r2.connection)
|
|
|
|
# if one of them switches, they should have
|
|
# separate conncetion objects
|
|
r2.select(db=10, host='localhost', port=6379)
|
|
self.assertNotEqual(r1.connection, r2.connection)
|
|
|
|
conns = [r1.connection, r2.connection]
|
|
conns.sort()
|
|
|
|
# but returning to the original state shares the object again
|
|
r2.select(db=9, host='localhost', port=6379)
|
|
self.assertEquals(r1.connection, r2.connection)
|
|
|
|
# the connection manager should still have just 2 connections
|
|
mgr_conns = pool.get_all_connections()
|
|
mgr_conns.sort()
|
|
self.assertEquals(conns, mgr_conns)
|
|
|
|
def test_threaded_workers(self):
|
|
r = redis.Redis(host='localhost', port=6379, db=9)
|
|
r.set('a', 'foo')
|
|
r.set('b', 'bar')
|
|
|
|
def _info_worker():
|
|
for i in range(50):
|
|
_ = r.info()
|
|
time.sleep(0.01)
|
|
|
|
def _keys_worker():
|
|
for i in range(50):
|
|
_ = r.keys()
|
|
time.sleep(0.01)
|
|
|
|
t1 = threading.Thread(target=_info_worker)
|
|
t2 = threading.Thread(target=_keys_worker)
|
|
t1.start()
|
|
t2.start()
|
|
|
|
for i in [t1, t2]:
|
|
i.join()
|
|
|