Add redis_url cmdline argument for tests (#1170)

Users can now specify --redis-url when running the test suite to choose a specific server. 

1 files changed, 46 insertions, 31 deletions

diff --git a/tests/conftest.py b/tests/conftest.py
index bb4c9a9..0ab6428 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -5,55 +5,70 @@ from mock import Mock
 from distutils.version import StrictVersion
 
 
-_REDIS_INFO = {}
+REDIS_INFO = {}
+default_redis_url = "redis://localhost:6379/9"
 
 
-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_INFO:
-        client = redis.Redis(**params)
-        _REDIS_INFO[key] = client.info()
-        client.connection_pool.disconnect()
-    return _REDIS_INFO[key]
+def pytest_addoption(parser):
+    parser.addoption('--redis-url', default=default_redis_url,
+                     action="store",
+                     help="Redis connection string,"
+                          " defaults to `%(default)s`")
 
 
-def _get_client(cls, request=None, **kwargs):
-    params = {'host': 'localhost', 'port': 6379, 'db': 9}
-    params.update(kwargs)
-    client = cls(**params)
-    client.flushdb()
-    if request:
-        def teardown():
-            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 _get_info(redis_url):
+    client = redis.Redis.from_url(redis_url)
+    info = client.info()
+    client.connection_pool.disconnect()
+    return info
+
+
+def pytest_sessionstart(session):
+    redis_url = session.config.getoption("--redis-url")
+    info = _get_info(redis_url)
+    version = info["redis_version"]
+    arch_bits = info["arch_bits"]
+    REDIS_INFO["version"] = version
+    REDIS_INFO["arch_bits"] = arch_bits
 
 
 def skip_if_server_version_lt(min_version):
-    redis_version = get_info()['redis_version']
+    redis_version = REDIS_INFO["version"]
     check = StrictVersion(redis_version) < StrictVersion(min_version)
-    return pytest.mark.skipif(check, reason="")
+    return pytest.mark.skipif(
+        check,
+        reason="Redis version required >= {}".format(min_version))
 
 
 def skip_if_server_version_gte(min_version):
-    redis_version = get_info()['redis_version']
+    redis_version = REDIS_INFO["version"]
     check = StrictVersion(redis_version) >= StrictVersion(min_version)
-    return pytest.mark.skipif(check, reason="")
+    return pytest.mark.skipif(
+        check,
+        reason="Redis version required < {}".format(min_version))
 
 
 def skip_unless_arch_bits(arch_bits):
-    return pytest.mark.skipif(get_info()['arch_bits'] != arch_bits,
+    return pytest.mark.skipif(REDIS_INFO["arch_bits"] != arch_bits,
                               reason="server is not {}-bit".format(arch_bits))
 
 
+def _get_client(cls, request, **kwargs):
+    redis_url = request.config.getoption("--redis-url")
+    client = cls.from_url(redis_url, **kwargs)
+    if request:
+        def teardown():
+            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
+
+
 @pytest.fixture()
 def r(request, **kwargs):
     return _get_client(redis.Redis, request, **kwargs)