summaryrefslogtreecommitdiff
path: root/tests/test_connection_pool.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_connection_pool.py')
-rw-r--r--tests/test_connection_pool.py48
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()