diff options
| -rw-r--r-- | CHANGES | 12 | ||||
| -rwxr-xr-x | redis/client.py | 21 | ||||
| -rwxr-xr-x | redis/connection.py | 33 | ||||
| -rw-r--r-- | tests/conftest.py | 28 | ||||
| -rw-r--r-- | tests/test_commands.py | 36 | ||||
| -rw-r--r-- | tests/test_connection_pool.py | 4 |
6 files changed, 111 insertions, 23 deletions
@@ -1,4 +1,16 @@ * 3.1.0 (in development) + * Connection URLs must have one of the following schemes: + redis://, rediss://, unix://. Thanks @jdupl123. #961/#969 + * Fixed an issue with retry_on_timeout logic that caused some TimeoutErrors + to be retried. Thanks Aaron Yang. #1022/#1023 + * Added support for SNI for SSL. Thanks @oridistor and Roey Prat. #1087 + * Fixed ConnectionPool repr for pools with no connections. Thanks + Cody Scott. #1043/#995 + * Fixed GEOHASH to return a None value when specifying a place that + doesn't exist on the server. Thanks @guybe7. #1126 + * Fixed XREADGROUP to return an empty dictionary for messages that + have been deleted but still exist in the unacknowledged queue. Thanks + @xeizmendi. #1116 * Added an owned method to Lock objects. owned returns a boolean indicating whether the current lock instance still owns the lock. Thanks Dave Johansen. #1112 diff --git a/redis/client.py b/redis/client.py index 2645e83..24120d2 100755 --- a/redis/client.py +++ b/redis/client.py @@ -191,6 +191,8 @@ def parse_sentinel_get_master(response): def pairs_to_dict(response, decode_keys=False): "Create a dict given a list of key/value pairs" + if response is None: + return {} if decode_keys: # the iter form is faster, but I don't know how to make that work # with a nativestr() map @@ -244,6 +246,12 @@ def int_or_none(response): return int(response) +def nativestr_or_none(response): + if response is None: + return None + return nativestr(response) + + def parse_stream_list(response): if response is None: return None @@ -508,7 +516,7 @@ class Redis(object): 'CONFIG RESETSTAT': bool_ok, 'CONFIG SET': bool_ok, 'DEBUG OBJECT': parse_debug_object, - 'GEOHASH': lambda r: list(map(nativestr, r)), + 'GEOHASH': lambda r: list(map(nativestr_or_none, r)), 'GEOPOS': lambda r: list(map(lambda ll: (float(ll[0]), float(ll[1])) if ll is not None else None, r)), @@ -767,7 +775,8 @@ class Redis(object): return self.parse_response(connection, command_name, **options) except (ConnectionError, TimeoutError) as e: connection.disconnect() - if not connection.retry_on_timeout and isinstance(e, TimeoutError): + if not (connection.retry_on_timeout and + isinstance(e, TimeoutError)): raise connection.send_command(*args) return self.parse_response(connection, command_name, **options) @@ -2199,7 +2208,7 @@ class Redis(object): return self.execute_command('XREAD', *pieces) def xreadgroup(self, groupname, consumername, streams, count=None, - block=None): + block=None, noack=False): """ Read from a stream via a consumer group. groupname: name of the consumer group. @@ -2209,6 +2218,7 @@ class Redis(object): count: if set, only return this many items, beginning with the earliest available. block: number of milliseconds to wait, if nothing already present. + noack: do not add messages to the PEL """ pieces = [Token.get_token('GROUP'), groupname, consumername] if count is not None: @@ -2222,6 +2232,8 @@ class Redis(object): "integer") pieces.append(Token.get_token("BLOCK")) pieces.append(str(block)) + if noack: + pieces.append(Token.get_token("NOACK")) if not isinstance(streams, dict) or len(streams) == 0: raise DataError('XREADGROUP streams must be a non empty dict') pieces.append(Token.get_token('STREAMS')) @@ -3001,7 +3013,8 @@ class PubSub(object): return command(*args) except (ConnectionError, TimeoutError) as e: connection.disconnect() - if not connection.retry_on_timeout and isinstance(e, TimeoutError): + if not (connection.retry_on_timeout and + isinstance(e, TimeoutError)): raise # Connect manually here. If the Redis server is down, this will # fail and raise a ConnectionError as desired. diff --git a/redis/connection.py b/redis/connection.py index 7575b76..c81e4c1 100755 --- a/redis/connection.py +++ b/redis/connection.py @@ -727,11 +727,24 @@ class SSLConnection(Connection): def _connect(self): "Wrap the socket with SSL support" sock = super(SSLConnection, self)._connect() - sock = ssl.wrap_socket(sock, - cert_reqs=self.cert_reqs, - keyfile=self.keyfile, - certfile=self.certfile, - ca_certs=self.ca_certs) + if hasattr(ssl, "create_default_context"): + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = self.cert_reqs + if self.certfile and self.keyfile: + context.load_cert_chain(certfile=self.certfile, + keyfile=self.keyfile) + if self.ca_certs: + context.load_verify_locations(self.ca_certs) + sock = context.wrap_socket(sock, server_hostname=self.host) + else: + # In case this code runs in a version which is older than 2.7.9, + # we want to fall back to old code + sock = ssl.wrap_socket(sock, + cert_reqs=self.cert_reqs, + keyfile=self.keyfile, + certfile=self.certfile, + ca_certs=self.ca_certs) return sock @@ -870,7 +883,7 @@ class ConnectionPool(object): path = url.path hostname = url.hostname - # We only support redis:// and unix:// schemes. + # We only support redis://, rediss:// and unix:// schemes. if url.scheme == 'unix': url_options.update({ 'password': password, @@ -878,7 +891,7 @@ class ConnectionPool(object): 'connection_class': UnixDomainSocketConnection, }) - else: + elif url.scheme in ('redis', 'rediss'): url_options.update({ 'host': hostname, 'port': int(url.port or 6379), @@ -895,6 +908,10 @@ class ConnectionPool(object): if url.scheme == 'rediss': url_options['connection_class'] = SSLConnection + else: + valid_schemes = ', '.join(('redis://', 'rediss://', 'unix://')) + raise ValueError('Redis URL must specify one of the following' + 'schemes (%s)' % valid_schemes) # last shot at the db value url_options['db'] = int(url_options.get('db', db or 0)) @@ -939,7 +956,7 @@ class ConnectionPool(object): def __repr__(self): return "%s<%s>" % ( type(self).__name__, - self.connection_class.description_format % self.connection_kwargs, + repr(self.connection_class(**self.connection_kwargs)), ) def reset(self): diff --git a/tests/conftest.py b/tests/conftest.py index 6e046f2..bb4c9a9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,18 +5,18 @@ from mock import Mock from distutils.version import StrictVersion -_REDIS_VERSIONS = {} +_REDIS_INFO = {} -def get_version(**kwargs): +def get_info(**kwargs): params = {'host': 'localhost', 'port': 6379, 'db': 9} params.update(kwargs) key = '%s:%s' % (params['host'], params['port']) - if key not in _REDIS_VERSIONS: + if key not in _REDIS_INFO: client = redis.Redis(**params) - _REDIS_VERSIONS[key] = client.info()['redis_version'] + _REDIS_INFO[key] = client.info() client.connection_pool.disconnect() - return _REDIS_VERSIONS[key] + return _REDIS_INFO[key] def _get_client(cls, request=None, **kwargs): @@ -26,22 +26,34 @@ def _get_client(cls, request=None, **kwargs): client.flushdb() if request: def teardown(): - client.flushdb() + try: + client.flushdb() + except redis.ConnectionError: + # handle cases where a test disconnected a client + # just manually retry the flushdb + client.flushdb() client.connection_pool.disconnect() request.addfinalizer(teardown) return client def skip_if_server_version_lt(min_version): - check = StrictVersion(get_version()) < StrictVersion(min_version) + redis_version = get_info()['redis_version'] + check = StrictVersion(redis_version) < StrictVersion(min_version) return pytest.mark.skipif(check, reason="") def skip_if_server_version_gte(min_version): - check = StrictVersion(get_version()) >= StrictVersion(min_version) + redis_version = get_info()['redis_version'] + check = StrictVersion(redis_version) >= StrictVersion(min_version) return pytest.mark.skipif(check, reason="") +def skip_unless_arch_bits(arch_bits): + return pytest.mark.skipif(get_info()['arch_bits'] != arch_bits, + reason="server is not {}-bit".format(arch_bits)) + + @pytest.fixture() def r(request, **kwargs): return _get_client(redis.Redis, request, **kwargs) diff --git a/tests/test_commands.py b/tests/test_commands.py index 566437d..33f78d5 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -11,7 +11,8 @@ from redis._compat import (unichr, ascii_letters, iteritems, iterkeys, from redis.client import parse_info from redis import exceptions -from .conftest import skip_if_server_version_lt, skip_if_server_version_gte +from .conftest import (skip_if_server_version_lt, skip_if_server_version_gte, + skip_unless_arch_bits) @pytest.fixture() @@ -1686,9 +1687,10 @@ class TestRedisCommands(object): (2.1873744593677, 41.406342043777, 'place2') r.geoadd('barcelona', *values) - assert r.geohash('barcelona', 'place1', 'place2') ==\ - ['sp3e9yg3kd0', 'sp3e9cbc3t0'] + assert r.geohash('barcelona', 'place1', 'place2', 'place3') ==\ + ['sp3e9yg3kd0', 'sp3e9cbc3t0', None] + @skip_unless_arch_bits(64) @skip_if_server_version_lt('3.2.0') def test_geopos(self, r): values = (2.1909389952632, 41.433791470673, 'place1') +\ @@ -1735,6 +1737,7 @@ class TestRedisCommands(object): assert r.georadius('barcelona', 2.191, 41.433, 1, unit='km') ==\ [b'place1'] + @skip_unless_arch_bits(64) @skip_if_server_version_lt('3.2.0') def test_georadius_with(self, r): values = (2.1909389952632, 41.433791470673, 'place1') +\ @@ -1792,6 +1795,7 @@ class TestRedisCommands(object): r.georadius('barcelona', 2.191, 41.433, 1000, store='places_barcelona') assert r.zrange('places_barcelona', 0, -1) == [b'place1'] + @skip_unless_arch_bits(64) @skip_if_server_version_lt('3.2.0') def test_georadius_store_dist(self, r): values = (2.1909389952632, 41.433791470673, 'place1') +\ @@ -1803,6 +1807,7 @@ class TestRedisCommands(object): # instead of save the geo score, the distance is saved. assert r.zscore('places_barcelona', 'place1') == 88.05060698409301 + @skip_unless_arch_bits(64) @skip_if_server_version_lt('3.2.0') def test_georadiusmember(self, r): values = (2.1909389952632, 41.433791470673, 'place1') +\ @@ -2205,6 +2210,31 @@ class TestRedisCommands(object): # xread starting after the last message returns an empty message list assert r.xreadgroup(group, consumer, streams={stream: '>'}) == expected + # xreadgroup with noack does not have any items in the PEL + r.xgroup_destroy(stream, group) + r.xgroup_create(stream, group, '0') + assert len(r.xreadgroup(group, consumer, streams={stream: '>'}, + noack=True)[0][1]) == 2 + # now there should be nothing pending + assert len(r.xreadgroup(group, consumer, + streams={stream: '0'})[0][1]) == 0 + + r.xgroup_destroy(stream, group) + r.xgroup_create(stream, group, '0') + # delete all the messages in the stream + expected = [ + [ + stream.encode(), + [ + (m1, {}), + (m2, {}), + ] + ] + ] + r.xreadgroup(group, consumer, streams={stream: '>'}) + r.xtrim(stream, 0) + assert r.xreadgroup(group, consumer, streams={stream: '0'}) == expected + @skip_if_server_version_lt('5.0.0') def test_xrevrange(self, r): stream = 'stream' diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py index b0dec67..ca56a76 100644 --- a/tests/test_connection_pool.py +++ b/tests/test_connection_pool.py @@ -308,6 +308,10 @@ class TestConnectionPoolURLParsing(object): 'password': None, } + def test_invalid_scheme_raises_error(self): + with pytest.raises(ValueError): + redis.ConnectionPool.from_url('localhost') + class TestConnectionPoolUnixSocketURLParsing(object): def test_defaults(self): |