Moved CredentialsProvider to a separate file, added type hints

Author: barshaul

redis/__init__.py

@@ -6,10 +6,10 @@
     BlockingConnectionPool,
     Connection,
     ConnectionPool,
-    CredentialsProvider,
     SSLConnection,
     UnixDomainSocketConnection,
 )
+from redis.credentials import CredentialsProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,

redis/connection.py

@@ -13,6 +13,7 @@
 from packaging.version import Version
 
 from redis.backoff import NoBackoff
+from redis.credentials import CredentialsProvider
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
@@ -500,42 +501,6 @@ def read_response(self, disable_decoding=False):
     DefaultParser = PythonParser
 
 
-class CredentialsProvider:
-    def __init__(self, username="", password="", supplier=None, *args, **kwargs):
-        """
-        Initialize a new Credentials Provider.
-        :param supplier: a supplier function that returns the username and password.
-                         def supplier(arg1, arg2, ...) -> (username, password)
-                         For examples see examples/connection_examples.ipynb
-        :param args: arguments to pass to the supplier function
-        :param kwargs: keyword arguments to pass to the supplier function
-        """
-        self.username = username
-        self.password = password
-        self.supplier = supplier
-        self.args = args
-        self.kwargs = kwargs
-
-    def get_credentials(self):
-        if self.supplier:
-            self.username, self.password = self.supplier(*self.args, **self.kwargs)
-        if self.username:
-            auth_args = (self.username, self.password or "")
-        else:
-            auth_args = (self.password,)
-        return auth_args
-
-    def get_password(self, call_supplier=True):
-        if call_supplier and self.supplier:
-            self.username, self.password = self.supplier(*self.args, **self.kwargs)
-        return self.password
-
-    def get_username(self, call_supplier=True):
-        if call_supplier and self.supplier:
-            self.username, self.password = self.supplier(*self.args, **self.kwargs)
-        return self.username
-
-
 class Connection:
     "Manages TCP communication to and from a Redis server"
 

redis/credentials.py

@@ -0,0 +1,41 @@
+class CredentialsProvider:
+    def __init__(
+        self,
+        username: str = "",
+        password: str = "",
+        supplier: callable = None,
+        *args,
+        **kwargs,
+    ):
+        """
+        Initialize a new Credentials Provider.
+        :param supplier: a supplier function that returns the username and password.
+                         def supplier(arg1, arg2, ...) -> (username, password)
+                         For examples see examples/connection_examples.ipynb
+        :param args: arguments to pass to the supplier function
+        :param kwargs: keyword arguments to pass to the supplier function
+        """
+        self.username = username
+        self.password = password
+        self.supplier = supplier
+        self.args = args
+        self.kwargs = kwargs
+
+    def get_credentials(self):
+        if self.supplier:
+            self.username, self.password = self.supplier(*self.args, **self.kwargs)
+        if self.username:
+            auth_args = (self.username, self.password or "")
+        else:
+            auth_args = (self.password,)
+        return auth_args
+
+    def get_password(self, call_supplier: bool = True):
+        if call_supplier and self.supplier:
+            self.username, self.password = self.supplier(*self.args, **self.kwargs)
+        return self.password
+
+    def get_username(self, call_supplier: bool = True):
+        if call_supplier and self.supplier:
+            self.username, self.password = self.supplier(*self.args, **self.kwargs)
+        return self.username

tests/test_connection.py

@@ -1,25 +1,17 @@
-import random
 import socket
-import string
 import types
 from unittest import mock
 from unittest.mock import patch
 
 import pytest
 
-import redis
 from redis.backoff import NoBackoff
-from redis.connection import Connection, CredentialsProvider
-from redis.exceptions import (
-    ConnectionError,
-    InvalidResponse,
-    ResponseError,
-    TimeoutError,
-)
+from redis.connection import Connection
+from redis.exceptions import ConnectionError, InvalidResponse, TimeoutError
 from redis.retry import Retry
 from redis.utils import HIREDIS_AVAILABLE
 
-from .conftest import _get_client, skip_if_redis_enterprise, skip_if_server_version_lt
+from .conftest import skip_if_server_version_lt
 
 
 @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
@@ -130,86 +122,3 @@ def test_connect_timeout_error_without_retry(self):
         assert conn._connect.call_count == 1
         assert str(e.value) == "Timeout connecting to server"
         self.clear(conn)
-
-
-class TestCredentialsProvider:
-    @skip_if_redis_enterprise()
-    def test_credentials_provider_without_supplier(self, r, request):
-        # first, test for default user (`username` is supposed to be optional)
-        default_username = "default"
-        temp_pass = "temp_pass"
-        creds_provider = CredentialsProvider(default_username, temp_pass)
-        r.config_set("requirepass", temp_pass)
-        creds = creds_provider.get_credentials()
-        assert r.auth(creds[1], creds[0]) is True
-        assert r.auth(creds_provider.get_password()) is True
-
-        # test for other users
-        username = "redis-py-auth"
-        password = "strong_password"
-
-        def teardown():
-            try:
-                r.auth(temp_pass)
-            except ResponseError:
-                r.auth("default", "")
-            r.config_set("requirepass", "")
-            r.acl_deluser(username)
-
-        request.addfinalizer(teardown)
-
-        assert r.acl_setuser(
-            username,
-            enabled=True,
-            passwords=["+" + password],
-            keys="~*",
-            commands=["+ping", "+command", "+info", "+select", "+flushdb", "+cluster"],
-        )
-
-        creds_provider2 = CredentialsProvider(username, password)
-        r2 = _get_client(
-            redis.Redis, request, flushdb=False, credentials_provider=creds_provider2
-        )
-
-        assert r2.ping() is True
-
-    @skip_if_redis_enterprise()
-    def test_credentials_provider_with_supplier(self, r, request):
-        import functools
-
-        @functools.lru_cache(maxsize=10)
-        def auth_supplier(user, endpoint):
-            def get_random_string(length):
-                letters = string.ascii_lowercase
-                result_str = "".join(random.choice(letters) for i in range(length))
-                return result_str
-
-            auth_token = get_random_string(5) + user + "_" + endpoint
-            return user, auth_token
-
-        username = "redis-py-auth"
-        creds_provider = CredentialsProvider(
-            supplier=auth_supplier,
-            user=username,
-            endpoint="localhost",
-        )
-        password = creds_provider.get_password()
-
-        assert r.acl_setuser(
-            username,
-            enabled=True,
-            passwords=["+" + password],
-            keys="~*",
-            commands=["+ping", "+command", "+info", "+select", "+flushdb", "+cluster"],
-        )
-
-        def teardown():
-            r.acl_deluser(username)
-
-        request.addfinalizer(teardown)
-
-        r2 = _get_client(
-            redis.Redis, request, flushdb=False, credentials_provider=creds_provider
-        )
-
-        assert r2.ping() is True

tests/test_credentials.py

@@ -0,0 +1,90 @@
+import random
+import string
+
+import redis
+from redis import ResponseError
+from redis.credentials import CredentialsProvider
+from tests.conftest import _get_client, skip_if_redis_enterprise
+
+
+class TestCredentialsProvider:
+    @skip_if_redis_enterprise()
+    def test_credentials_provider_without_supplier(self, r, request):
+        # first, test for default user (`username` is supposed to be optional)
+        default_username = "default"
+        temp_pass = "temp_pass"
+        creds_provider = CredentialsProvider(default_username, temp_pass)
+        r.config_set("requirepass", temp_pass)
+        creds = creds_provider.get_credentials()
+        assert r.auth(creds[1], creds[0]) is True
+        assert r.auth(creds_provider.get_password()) is True
+
+        # test for other users
+        username = "redis-py-auth"
+        password = "strong_password"
+
+        def teardown():
+            try:
+                r.auth(temp_pass)
+            except ResponseError:
+                r.auth("default", "")
+            r.config_set("requirepass", "")
+            r.acl_deluser(username)
+
+        request.addfinalizer(teardown)
+
+        assert r.acl_setuser(
+            username,
+            enabled=True,
+            passwords=["+" + password],
+            keys="~*",
+            commands=["+ping", "+command", "+info", "+select", "+flushdb", "+cluster"],
+        )
+
+        creds_provider2 = CredentialsProvider(username, password)
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, credentials_provider=creds_provider2
+        )
+
+        assert r2.ping() is True
+
+    @skip_if_redis_enterprise()
+    def test_credentials_provider_with_supplier(self, r, request):
+        import functools
+
+        @functools.lru_cache(maxsize=10)
+        def auth_supplier(user, endpoint):
+            def get_random_string(length):
+                letters = string.ascii_lowercase
+                result_str = "".join(random.choice(letters) for i in range(length))
+                return result_str
+
+            auth_token = get_random_string(5) + user + "_" + endpoint
+            return user, auth_token
+
+        username = "redis-py-auth"
+        creds_provider = CredentialsProvider(
+            supplier=auth_supplier,
+            user=username,
+            endpoint="localhost",
+        )
+        password = creds_provider.get_password()
+
+        assert r.acl_setuser(
+            username,
+            enabled=True,
+            passwords=["+" + password],
+            keys="~*",
+            commands=["+ping", "+command", "+info", "+select", "+flushdb", "+cluster"],
+        )
+
+        def teardown():
+            r.acl_deluser(username)
+
+        request.addfinalizer(teardown)
+
+        r2 = _get_client(
+            redis.Redis, request, flushdb=False, credentials_provider=creds_provider
+        )
+
+        assert r2.ping() is True