allow cert_reqs to be a string and convert it to the appropriate SSL constant.

Author: andymccurdy

redis/connection.py

@@ -570,6 +570,16 @@ def __init__(self, keyfile=None, certfile=None, cert_reqs=None,
         self.certfile = certfile
         if cert_reqs is None:
             cert_reqs = ssl.CERT_NONE
+        elif isinstance(cert_reqs, basestring):
+            CERT_REQS = {
+                'none': ssl.CERT_NONE,
+                'optional': ssl.CERT_OPTIONAL,
+                'required': ssl.CERT_REQUIRED
+            }
+            if cert_reqs not in CERT_REQS:
+                raise RedisError(
+                    "Invalid SSL Certificate Required Flag: %s" % cert_reqs)
+            cert_reqs = CERT_REQS[cert_reqs]
         self.cert_reqs = cert_reqs
         self.ca_certs = ca_certs
 

tests/test_connection_pool.py

@@ -6,6 +6,7 @@
 import re
 
 from threading import Thread
+from redis.connection import ssl_available
 from .conftest import skip_if_server_version_lt
 
 
@@ -289,6 +290,31 @@ def test_extra_querystring_options(self):
         }
 
 
+class TestSSLConnectionURLParsing(object):
+    @pytest.mark.skipif(not ssl_available, reason="SSL not installed")
+    def test_defaults(self):
+        pool = redis.ConnectionPool.from_url('rediss://localhost')
+        assert pool.connection_class == redis.SSLConnection
+        assert pool.connection_kwargs == {
+            'host': 'localhost',
+            'port': 6379,
+            'db': 0,
+            'password': None,
+        }
+
+    @pytest.mark.skipif(not ssl_available, reason="SSL not installed")
+    def test_cert_reqs_options(self):
+        import ssl
+        pool = redis.ConnectionPool.from_url('rediss://?cert_reqs=none')
+        assert pool.get_connection('_').cert_reqs == ssl.CERT_NONE
+
+        pool = redis.ConnectionPool.from_url('rediss://?cert_reqs=optional')
+        assert pool.get_connection('_').cert_reqs == ssl.CERT_OPTIONAL
+
+        pool = redis.ConnectionPool.from_url('rediss://?cert_reqs=required')
+        assert pool.get_connection('_').cert_reqs == ssl.CERT_REQUIRED
+
+
 class TestConnection(object):
     def test_on_connect_error(self):
         """