Merge branch 'master' of github.com:redis/redis-py into vv-test-8.0

Author: vladvildanov

redis/commands/search/aggregation.py

@@ -1,5 +1,7 @@
 from typing import List, Union
 
+from redis.commands.search.dialect import DEFAULT_DIALECT
+
 FIELDNAME = object()
 
 
@@ -110,7 +112,7 @@ def __init__(self, query: str = "*") -> None:
         self._with_schema = False
         self._verbatim = False
         self._cursor = []
-        self._dialect = None
+        self._dialect = DEFAULT_DIALECT
         self._add_scores = False
         self._scorer = "TFIDF"
 

redis/commands/search/dialect.py

@@ -0,0 +1,3 @@
+# Value for the default dialect to be used as a part of
+# Search or Aggregate query.
+DEFAULT_DIALECT = 2

redis/commands/search/query.py

@@ -1,5 +1,7 @@
 from typing import List, Optional, Union
 
+from redis.commands.search.dialect import DEFAULT_DIALECT
+
 
 class Query:
     """
@@ -40,7 +42,7 @@ def __init__(self, query_string: str) -> None:
         self._highlight_fields: List = []
         self._language: Optional[str] = None
         self._expander: Optional[str] = None
-        self._dialect: Optional[int] = None
+        self._dialect: int = DEFAULT_DIALECT
 
     def query_string(self) -> str:
         """Return the query string of this query only."""

redis/connection.py

@@ -904,9 +904,11 @@ def read_response(
                 and self._cache.get(self._current_command_cache_key).status
                 != CacheEntryStatus.IN_PROGRESS
             ):
-                return copy.deepcopy(
+                res = copy.deepcopy(
                     self._cache.get(self._current_command_cache_key).cache_value
                 )
+                self._current_command_cache_key = None
+                return res
 
         response = self._conn.read_response(
             disable_decoding=disable_decoding,
@@ -932,6 +934,8 @@ def read_response(
                 cache_entry.cache_value = response
                 self._cache.set(cache_entry)
 
+            self._current_command_cache_key = None
+
         return response
 
     def pack_command(self, *args):
@@ -1374,6 +1378,7 @@ def __init__(
         # will notice the first thread already did the work and simply
         # release the lock.
         self._fork_lock = threading.Lock()
+        self._lock = threading.Lock()
         self.reset()
 
     def __repr__(self) -> (str, str):
@@ -1391,7 +1396,6 @@ def get_protocol(self):
         return self.connection_kwargs.get("protocol", None)
 
     def reset(self) -> None:
-        self._lock = threading.Lock()
         self._created_connections = 0
         self._available_connections = []
         self._in_use_connections = set()

redis/typing.py

@@ -20,7 +20,7 @@
 
 
 Number = Union[int, float]
-EncodedT = Union[bytes, memoryview]
+EncodedT = Union[bytes, bytearray, memoryview]
 DecodedT = Union[str, int, float]
 EncodableT = Union[EncodedT, DecodedT]
 AbsExpiryT = Union[int, datetime]

tests/test_asyncio/test_search.py

@@ -1646,7 +1646,7 @@ async def test_search_commands_in_pipeline(decoded_r: redis.Redis):
 @pytest.mark.redismod
 async def test_query_timeout(decoded_r: redis.Redis):
     q1 = Query("foo").timeout(5000)
-    assert q1.get_args() == ["foo", "TIMEOUT", 5000, "LIMIT", 0, 10]
+    assert q1.get_args() == ["foo", "TIMEOUT", 5000, "DIALECT", 2, "LIMIT", 0, 10]
     q2 = Query("foo").timeout("not_a_number")
     with pytest.raises(redis.ResponseError):
         await decoded_r.ft().search(q2)

tests/test_auth/test_token_manager.py

@@ -73,20 +73,18 @@ def on_next(token):
         assert len(tokens) > 0
 
     @pytest.mark.parametrize(
-        "exp_refresh_ratio,tokens_refreshed",
+        "exp_refresh_ratio",
         [
-            (0.9, 2),
-            (0.28, 4),
+            (0.9),
+            (0.28),
         ],
         ids=[
-            "Refresh ratio = 0.9,  2 tokens in 0,1 second",
-            "Refresh ratio = 0.28, 4 tokens in 0,1 second",
+            "Refresh ratio = 0.9",
+            "Refresh ratio = 0.28",
         ],
     )
     @pytest.mark.asyncio
-    async def test_async_success_token_renewal(
-        self, exp_refresh_ratio, tokens_refreshed
-    ):
+    async def test_async_success_token_renewal(self, exp_refresh_ratio):
         tokens = []
         mock_provider = Mock(spec=IdentityProviderInterface)
         mock_provider.request_token.side_effect = [
@@ -129,7 +127,7 @@ async def on_next(token):
         await mgr.start_async(mock_listener, block_for_initial=True)
         await asyncio.sleep(0.1)
 
-        assert len(tokens) == tokens_refreshed
+        assert len(tokens) > 0
 
     @pytest.mark.parametrize(
         "block_for_initial,tokens_acquired",
@@ -203,7 +201,7 @@ def on_next(token):
         # additional token renewal.
         sleep(0.1)
 
-        assert len(tokens) == 1
+        assert len(tokens) > 0
 
     @pytest.mark.asyncio
     async def test_async_token_renewal_with_skip_initial(self):
@@ -245,7 +243,7 @@ async def on_next(token):
         # due to additional token renewal.
         await asyncio.sleep(0.2)
 
-        assert len(tokens) == 2
+        assert len(tokens) > 0
 
     def test_success_token_renewal_with_retry(self):
         tokens = []

tests/test_connection.py

@@ -501,9 +501,9 @@ def test_read_response_returns_cached_reply(self, mock_cache, mock_connection):
         )
         proxy_connection.send_command(*["GET", "foo"], **{"keys": ["foo"]})
         assert proxy_connection.read_response() == b"bar"
+        assert proxy_connection._current_command_cache_key is None
         assert proxy_connection.read_response() == b"bar"
 
-        mock_connection.read_response.assert_called_once()
         mock_cache.set.assert_has_calls(
             [
                 call(
@@ -530,9 +530,6 @@ def test_read_response_returns_cached_reply(self, mock_cache, mock_connection):
                 call(CacheKey(command="GET", redis_keys=("foo",))),
                 call(CacheKey(command="GET", redis_keys=("foo",))),
                 call(CacheKey(command="GET", redis_keys=("foo",))),
-                call(CacheKey(command="GET", redis_keys=("foo",))),
-                call(CacheKey(command="GET", redis_keys=("foo",))),
-                call(CacheKey(command="GET", redis_keys=("foo",))),
             ]
         )
 

tests/test_connection_pool.py

@@ -7,10 +7,16 @@
 
 import pytest
 import redis
-from redis.connection import to_bool
-from redis.utils import SSL_AVAILABLE
-
-from .conftest import _get_client, skip_if_redis_enterprise, skip_if_server_version_lt
+from redis.cache import CacheConfig
+from redis.connection import CacheProxyConnection, Connection, to_bool
+from redis.utils import HIREDIS_AVAILABLE, SSL_AVAILABLE
+
+from .conftest import (
+    _get_client,
+    skip_if_redis_enterprise,
+    skip_if_resp_version,
+    skip_if_server_version_lt,
+)
 from .test_pubsub import wait_for_message
 
 
@@ -196,6 +202,20 @@ def test_repr_contains_db_info_unix(self):
         expected = "path=abc,db=0,client_name=test-client"
         assert expected in repr(pool)
 
+    @pytest.mark.skipif(HIREDIS_AVAILABLE, reason="PythonParser only")
+    @pytest.mark.onlynoncluster
+    @skip_if_resp_version(2)
+    @skip_if_server_version_lt("7.4.0")
+    def test_initialise_pool_with_cache(self, master_host):
+        pool = redis.BlockingConnectionPool(
+            connection_class=Connection,
+            host=master_host[0],
+            port=master_host[1],
+            protocol=3,
+            cache_config=CacheConfig(),
+        )
+        assert isinstance(pool.get_connection("_"), CacheProxyConnection)
+
 
 class TestConnectionPoolURLParsing:
     def test_hostname(self):