Added cache support for SSLConnection

Author: vladvildanov

redis/connection.py

@@ -863,7 +863,9 @@ def _enable_tracking_callback(self, conn: ConnectionInterface) -> None:
         conn._parser.set_invalidation_push_handler(self._on_invalidation_callback)
 
     def _process_pending_invalidations(self):
-        while self.can_read():
+        while self.retry.call_with_retry_on_false(
+            lambda: self.can_read()
+        ):
             self._conn.read_response(push_request=True)
 
     def _on_invalidation_callback(

redis/retry.py

@@ -1,6 +1,7 @@
 import socket
+import time
 from time import sleep
-from typing import TYPE_CHECKING, Any, Callable, Iterable, Tuple, Type, TypeVar
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Tuple, Type, TypeVar, Optional
 
 from redis.exceptions import ConnectionError, TimeoutError
 
@@ -68,3 +69,42 @@ def call_with_retry(
                 backoff = self._backoff.compute(failures)
                 if backoff > 0:
                     sleep(backoff)
+
+    def call_with_retry_on_false(
+            self,
+            do: Callable[[], T],
+            on_false: Optional[Callable[[], T]] = None,
+            max_retries: Optional[int] = 3,
+            timeout: Optional[float] = 0,
+            exponent: Optional[int] = 2,
+    ) -> bool:
+        """
+        Execute an operation that returns boolean value with retry
+        logic in case if false value been returned.
+        `do`: the operation to call. Expects no argument.
+        `on_false`: Callback to be executed on retry fail.
+        """
+        res = do()
+
+        if res:
+            return res
+
+        if on_false is not None:
+            on_false()
+
+        if max_retries > 0:
+            if timeout > 0:
+                time.sleep(timeout)
+
+            return self.call_with_retry_on_false(
+                do,
+                on_false,
+                max_retries - 1,
+                timeout * exponent,
+                exponent
+            )
+
+        return False
+
+
+

tests/conftest.py

@@ -11,9 +11,10 @@
 from packaging.version import Version
 from redis import Sentinel
 from redis.backoff import NoBackoff
-from redis.connection import Connection, parse_url
+from redis.connection import Connection, parse_url, SSLConnection
 from redis.exceptions import RedisClusterException
 from redis.retry import Retry
+from tests.ssl_utils import get_ssl_filename
 
 REDIS_INFO = {}
 default_redis_url = "redis://localhost:6379/0"
@@ -323,6 +324,18 @@ def _get_client(
     cluster_mode = REDIS_INFO["cluster_enabled"]
     if not cluster_mode:
         url_options = parse_url(redis_url)
+        connection_class = Connection
+        ssl = kwargs.pop("ssl", False)
+        if ssl:
+            connection_class = SSLConnection
+            kwargs["ssl_certfile"] = get_ssl_filename("client-cert.pem")
+            kwargs["ssl_keyfile"] = get_ssl_filename("client-key.pem")
+            # When you try to assign "required" as single string, it assigns tuple instead of string.
+            # Probably some reserved keyword, I can't explain how does it work -_-
+            kwargs["ssl_cert_reqs"] = "require"+"d"
+            kwargs["ssl_ca_certs"] = get_ssl_filename("ca-cert.pem")
+            kwargs["port"] = 6666
+        kwargs["connection_class"] = connection_class
         url_options.update(kwargs)
         pool = redis.ConnectionPool(**url_options)
         client = cls(connection_pool=pool)

tests/test_cache.py

@@ -15,11 +15,13 @@ def r(request):
     cache = request.param.get("cache")
     kwargs = request.param.get("kwargs", {})
     protocol = request.param.get("protocol", 3)
+    ssl = request.param.get("ssl", False)
     single_connection_client = request.param.get("single_connection_client", False)
     with _get_client(
             redis.Redis,
             request,
             protocol=protocol,
+            ssl=ssl,
             single_connection_client=single_connection_client,
             use_cache=use_cache,
             cache=cache,
@@ -630,3 +632,29 @@ def test_cache_clears_on_disconnect(self, master, cache):
         master.connection_pool.get_connection('_').disconnect()
         # Make sure cache is empty
         assert cache.currsize == 0
+
+@pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
+@pytest.mark.onlynoncluster
+class TestSSLCache:
+    @pytest.mark.parametrize("r", [
+        {
+            "cache": TTLCache(128, 300),
+            "use_cache": True,
+            "ssl": True,
+        }
+    ], indirect=True)
+    @pytest.mark.onlynoncluster
+    def test_get_from_cache(self, r, r2, cache):
+        r, cache = r
+        # add key to redis
+        r.set("foo", "bar")
+        # get key from redis and save in local cache
+        assert r.get("foo") == b"bar"
+        # get key from local cache
+        assert cache.get(("GET", "foo")) == b"bar"
+        # change key in redis (cause invalidation)
+        assert r2.set("foo", "barbar")
+        # Retrieves a new value from server and cache it
+        assert r.get("foo") == b"barbar"
+        # Make sure that new value was cached
+        assert cache.get(("GET", "foo")) == b"barbar"

tests/test_cluster.py

@@ -643,10 +643,10 @@ def parse_response_mock_third(connection, *args, **options):
                 mocks["send_command"].assert_has_calls(
                     [
                         call("READONLY"),
-                        call("GET", "foo"),
+                        call("GET", "foo", keys=['foo']),
                         call("READONLY"),
-                        call("GET", "foo"),
-                        call("GET", "foo"),
+                        call("GET", "foo", keys=['foo']),
+                        call("GET", "foo", keys=['foo']),
                     ]
                 )