diff options
Diffstat (limited to 'tests/test_connection_pool.py')
-rw-r--r-- | tests/test_connection_pool.py | 48 |
1 files changed, 0 insertions, 48 deletions
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py index 0f5ad72..1f5797b 100644 --- a/tests/test_connection_pool.py +++ b/tests/test_connection_pool.py @@ -5,7 +5,6 @@ import time import re from threading import Thread -from redis.client import parse_client_list from redis.connection import ssl_available, to_bool from .conftest import skip_if_server_version_lt @@ -76,39 +75,6 @@ class TestConnectionPool(object): expected = 'ConnectionPool<UnixDomainSocketConnection<path=/abc,db=1>>' assert repr(pool) == expected - def test_pool_provides_healthy_connections(self): - pool = self.get_pool(connection_class=redis.Connection, - max_connections=2) - conn1 = pool.get_connection('_') - conn2 = pool.get_connection('_') - - # set a unique name on the connection we'll be testing - conn1._same_connection_value = 'killed-client' - conn1.send_command('client', 'setname', 'redis-py-1') - assert conn1.read_response() == b'OK' - pool.release(conn1) - - # find the well named client in the client list - conn2.send_command('client', 'list') - client_list = parse_client_list(conn2.read_response()) - for client in client_list: - if client['name'] == 'redis-py-1': - break - else: - assert False, 'Client redis-py-1 not found in client list' - - # kill the well named client - conn2.send_command('client', 'kill', client['addr']) - assert conn2.read_response() == b'OK' - - # our connection should have been disconnected, but a quality - # connection pool would know this and only provide a healthy - # connection. - conn = pool.get_connection('_') - assert conn == conn1 - conn.send_command('ping') - assert conn.read_response() == b'PONG' - class TestBlockingConnectionPool(object): def get_pool(self, connection_kwargs=None, max_connections=10, timeout=20): @@ -541,17 +507,3 @@ class TestConnection(object): 'UnixDomainSocketConnection', 'path=/path/to/socket,db=0', ) - - def test_can_read(self, r): - connection = r.connection_pool.get_connection('ping') - assert not connection.can_read() - connection.send_command('ping') - # wait for the server to respond - wait_until = time.time() + 2 - while time.time() < wait_until: - if connection.can_read(): - break - time.sleep(0.01) - assert connection.can_read() - assert connection.read_response() == b'PONG' - assert not connection.can_read() |