Merge branch 'master' into ps_remove_sync_context_manager_from_async_cluster

petyaslavova · web-flow · commit 78be4c36af68 · 2025-06-27T13:16:38.000+03:00
diff --git a/.github/actions/run-tests/action.yml b/.github/actions/run-tests/action.yml
@@ -73,8 +73,8 @@ runs:
 
           # Mapping of redis version to stack version
           declare -A redis_stack_version_mapping=(
-            ["7.4.2"]="rs-7.4.0-v2"
-            ["7.2.7"]="rs-7.2.0-v14"
+            ["7.4.4"]="rs-7.4.0-v5"
+            ["7.2.9"]="rs-7.2.0-v17"
           )
 
           if [[ -v redis_stack_version_mapping[$REDIS_VERSION] ]]; then
@@ -91,8 +91,10 @@ runs:
           fi
 
           invoke devenv --endpoints=all-stack
+
         else
           echo "Using redis CE for module tests"
+          export CLIENT_LIBS_TEST_STACK_IMAGE_TAG=$REDIS_VERSION
           echo "REDIS_MOD_URL=redis://127.0.0.1:6379" >> $GITHUB_ENV
           invoke devenv --endpoints all
         fi
diff --git a/.github/workflows/hiredis-py-integration.yaml b/.github/workflows/hiredis-py-integration.yaml
@@ -23,8 +23,8 @@ env:
   CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
   # this speeds up coverage with Python 3.12: https://github.com/nedbat/coveragepy/issues/1665
   COVERAGE_CORE: sysmon
-  CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: 'rs-7.4.0-v2'
-  CURRENT_REDIS_VERSION: '7.4.2'
+  CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: '8.0.2'
+  CURRENT_REDIS_VERSION: '8.0.2'
 
 jobs:
   redis_version:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
@@ -27,8 +27,8 @@ env:
   CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
   # this speeds up coverage with Python 3.12: https://github.com/nedbat/coveragepy/issues/1665
   COVERAGE_CORE: sysmon
-  CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: 'rs-7.4.0-v2'
-  CURRENT_REDIS_VERSION: '7.4.2'
+  CURRENT_CLIENT_LIBS_TEST_STACK_IMAGE_TAG: '8.0.2'
+  CURRENT_REDIS_VERSION: '8.0.2'
 
 jobs:
   dependency-audit:
@@ -74,7 +74,7 @@ jobs:
       max-parallel: 15
       fail-fast: false
       matrix:
-        redis-version: ['8.0.1-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.2.7']
+        redis-version: ['8.2-M01-pre', '${{ needs.redis_version.outputs.CURRENT }}', '7.4.4', '7.2.9']
         python-version: ['3.9', '3.13']
         parser-backend: ['plain']
         event-loop: ['asyncio']
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -1,10 +1,10 @@
 ---
 # image tag 8.0-RC2-pre is the one matching the 8.0 GA release
 x-client-libs-stack-image: &client-libs-stack-image
-  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_STACK_IMAGE_TAG:-8.0-RC2-pre}"
+  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_STACK_IMAGE_TAG:-8.0.2}"
 
 x-client-libs-image: &client-libs-image
-  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_IMAGE_TAG:-8.0-RC2-pre}"
+  image: "redislabs/client-libs-test:${CLIENT_LIBS_TEST_IMAGE_TAG:-8.0.2}"
 
 services:
 
diff --git a/redis/client.py b/redis/client.py
@@ -368,9 +368,7 @@ def __init__(
         ]:
             raise RedisError("Client caching is only supported with RESP version 3")
 
-        # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
-        # TODO: Remove this before next major version (7.0.0)
-        self.single_connection_lock = threading.Lock()
+        self.single_connection_lock = threading.RLock()
         self.connection = None
         self._single_connection_client = single_connection_client
         if self._single_connection_client:
@@ -776,9 +774,7 @@ def __init__(
         else:
             self._event_dispatcher = event_dispatcher
 
-        # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
-        # TODO: Remove this before next major version (7.0.0)
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
         if self.encoder is None:
             self.encoder = self.connection_pool.get_encoder()
         self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE)
diff --git a/redis/commands/search/aggregation.py b/redis/commands/search/aggregation.py
@@ -26,7 +26,7 @@ class Reducer:
 
     NAME = None
 
-    def __init__(self, *args: List[str]) -> None:
+    def __init__(self, *args: str) -> None:
         self._args = args
         self._field = None
         self._alias = None
@@ -116,7 +116,7 @@ def __init__(self, query: str = "*") -> None:
         self._add_scores = False
         self._scorer = "TFIDF"
 
-    def load(self, *fields: List[str]) -> "AggregateRequest":
+    def load(self, *fields: str) -> "AggregateRequest":
         """
         Indicate the fields to be returned in the response. These fields are
         returned in addition to any others implicitly specified.
@@ -223,7 +223,7 @@ def limit(self, offset: int, num: int) -> "AggregateRequest":
         self._aggregateplan.extend(_limit.build_args())
         return self
 
-    def sort_by(self, *fields: List[str], **kwargs) -> "AggregateRequest":
+    def sort_by(self, *fields: str, **kwargs) -> "AggregateRequest":
         """
         Indicate how the results should be sorted. This can also be used for
         *top-N* style queries
diff --git a/redis/commands/search/commands.py b/redis/commands/search/commands.py
@@ -542,7 +542,7 @@ def explain_cli(self, query: Union[str, Query]):  # noqa
 
     def aggregate(
         self,
-        query: Union[str, Query],
+        query: Union[AggregateRequest, Cursor],
         query_params: Dict[str, Union[str, int, float]] = None,
     ):
         """
@@ -573,7 +573,7 @@ def aggregate(
         )
 
     def _get_aggregate_result(
-        self, raw: List, query: Union[str, Query, AggregateRequest], has_cursor: bool
+        self, raw: List, query: Union[AggregateRequest, Cursor], has_cursor: bool
     ):
         if has_cursor:
             if isinstance(query, Cursor):
@@ -967,7 +967,7 @@ async def search(
 
     async def aggregate(
         self,
-        query: Union[str, Query],
+        query: Union[AggregateResult, Cursor],
         query_params: Dict[str, Union[str, int, float]] = None,
     ):
         """
diff --git a/redis/connection.py b/redis/connection.py
@@ -810,7 +810,7 @@ def __init__(
         self,
         conn: ConnectionInterface,
         cache: CacheInterface,
-        pool_lock: threading.Lock,
+        pool_lock: threading.RLock,
     ):
         self.pid = os.getpid()
         self._conn = conn
@@ -1422,13 +1422,7 @@ def __init__(
         # release the lock.
 
         self._fork_lock = threading.RLock()
-
-        if self.cache is None:
-            self._lock = threading.RLock()
-        else:
-            # TODO: To avoid breaking changes during the bug fix, we have to keep non-reentrant lock.
-            # TODO: Remove this before next major version (7.0.0)
-            self._lock = threading.Lock()
+        self._lock = threading.RLock()
 
         self.reset()
 
diff --git a/redis/event.py b/redis/event.py
@@ -152,7 +152,7 @@ def __init__(
         self,
         connection,
         client_type: ClientType,
-        connection_lock: Union[threading.Lock, asyncio.Lock],
+        connection_lock: Union[threading.RLock, asyncio.Lock],
     ):
         self._connection = connection
         self._client_type = client_type
@@ -167,7 +167,7 @@ def client_type(self) -> ClientType:
         return self._client_type
 
     @property
-    def connection_lock(self) -> Union[threading.Lock, asyncio.Lock]:
+    def connection_lock(self) -> Union[threading.RLock, asyncio.Lock]:
         return self._connection_lock
 
 
@@ -177,7 +177,7 @@ def __init__(
         pubsub_connection,
         connection_pool,
         client_type: ClientType,
-        connection_lock: Union[threading.Lock, asyncio.Lock],
+        connection_lock: Union[threading.RLock, asyncio.Lock],
     ):
         self._pubsub_connection = pubsub_connection
         self._connection_pool = connection_pool
@@ -197,7 +197,7 @@ def client_type(self) -> ClientType:
         return self._client_type
 
     @property
-    def connection_lock(self) -> Union[threading.Lock, asyncio.Lock]:
+    def connection_lock(self) -> Union[threading.RLock, asyncio.Lock]:
         return self._connection_lock
 
 
diff --git a/tests/test_asyncio/compat.py b/tests/test_asyncio/compat.py
@@ -1,10 +1,4 @@
 import asyncio
-from unittest import mock
-
-try:
-    mock.AsyncMock
-except AttributeError:
-    from unittest import mock
 
 try:
     from contextlib import aclosing
diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py
@@ -1,6 +1,7 @@
 import random
 from contextlib import asynccontextmanager as _asynccontextmanager
 from typing import Union
+from unittest import mock
 
 import pytest
 import pytest_asyncio
@@ -14,8 +15,6 @@
 from redis.credentials import CredentialProvider
 from tests.conftest import REDIS_INFO, get_credential_provider
 
-from .compat import mock
-
 
 async def _get_info(redis_url):
     client = redis.Redis.from_url(redis_url)
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
@@ -4,6 +4,7 @@
 import ssl
 import warnings
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Type, Union
+from unittest import mock
 from urllib.parse import urlparse
 
 import pytest
@@ -48,7 +49,7 @@
 )
 
 from ..ssl_utils import get_tls_certificates
-from .compat import aclosing, mock
+from .compat import aclosing
 
 pytestmark = pytest.mark.onlycluster
 
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
@@ -1,6 +1,7 @@
 import asyncio
 import socket
 import types
+from unittest import mock
 from errno import ECONNREFUSED
 from unittest.mock import patch
 
@@ -25,7 +26,6 @@
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import skip_if_server_version_lt
 
-from .compat import mock
 from .mocks import MockStream
 
 
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
@@ -1,5 +1,6 @@
 import asyncio
 import re
+from unittest import mock
 
 import pytest
 import pytest_asyncio
@@ -8,7 +9,7 @@
 from redis.auth.token import TokenInterface
 from tests.conftest import skip_if_redis_enterprise, skip_if_server_version_lt
 
-from .compat import aclosing, mock
+from .compat import aclosing
 from .conftest import asynccontextmanager
 from .test_pubsub import wait_for_message
 
diff --git a/tests/test_asyncio/test_pipeline.py b/tests/test_asyncio/test_pipeline.py
@@ -1,8 +1,10 @@
+from unittest import mock
+
 import pytest
 import redis
 from tests.conftest import skip_if_server_version_lt
 
-from .compat import aclosing, mock
+from .compat import aclosing
 from .conftest import wait_for_command
 
 
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
@@ -12,14 +12,16 @@
 else:
     from async_timeout import timeout as async_timeout
 
+from unittest import mock
+
 import pytest
 import pytest_asyncio
 import redis.asyncio as redis
 from redis.exceptions import ConnectionError
 from redis.typing import EncodableT
 from tests.conftest import get_protocol_version, skip_if_server_version_lt
 
-from .compat import aclosing, create_task, mock
+from .compat import aclosing, create_task
 
 
 def with_timeout(t):
diff --git a/tests/test_asyncio/test_sentinel_managed_connection.py b/tests/test_asyncio/test_sentinel_managed_connection.py
@@ -1,12 +1,11 @@
 import socket
+from unittest import mock
 
 import pytest
 from redis.asyncio.retry import Retry
 from redis.asyncio.sentinel import SentinelManagedConnection
 from redis.backoff import NoBackoff
 
-from .compat import mock
-
 pytestmark = pytest.mark.asyncio
 
 
diff --git a/tests/test_cluster_transaction.py b/tests/test_cluster_transaction.py
@@ -285,7 +285,7 @@ def test_retry_transaction_on_connection_error(self, r, mock_connection):
         mock_pool = Mock(spec=ConnectionPool)
         mock_pool.get_connection.return_value = mock_connection
         mock_pool._available_connections = [mock_connection]
-        mock_pool._lock = threading.Lock()
+        mock_pool._lock = threading.RLock()
 
         _node_migrating, node_importing = _find_source_and_target_node_for_slot(r, slot)
         node_importing.redis_connection.connection_pool = mock_pool
@@ -310,7 +310,7 @@ def test_retry_transaction_on_connection_error_with_watched_keys(
         mock_pool = Mock(spec=ConnectionPool)
         mock_pool.get_connection.return_value = mock_connection
         mock_pool._available_connections = [mock_connection]
-        mock_pool._lock = threading.Lock()
+        mock_pool._lock = threading.RLock()
 
         _node_migrating, node_importing = _find_source_and_target_node_for_slot(r, slot)
         node_importing.redis_connection.connection_pool = mock_pool
diff --git a/tests/test_commands.py b/tests/test_commands.py
@@ -1088,13 +1088,23 @@ def test_lastsave(self, r):
 
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("5.0.0")
+    @skip_if_server_version_gte("8.0.0")
     def test_lolwut(self, r):
         lolwut = r.lolwut().decode("utf-8")
         assert "Redis ver." in lolwut
 
         lolwut = r.lolwut(5, 6, 7, 8).decode("utf-8")
         assert "Redis ver." in lolwut
 
+    @pytest.mark.onlynoncluster
+    @skip_if_server_version_lt("8.0.0")
+    def test_lolwut_v8_and_higher(self, r):
+        lolwut = r.lolwut().decode("utf-8")
+        assert lolwut
+
+        lolwut = r.lolwut(5, 6, 7, 8).decode("utf-8")
+        assert lolwut
+
     @pytest.mark.onlynoncluster
     @skip_if_server_version_lt("6.2.0")
     @skip_if_redis_enterprise()
diff --git a/tests/test_connection.py b/tests/test_connection.py
@@ -442,7 +442,7 @@ def test_clears_cache_on_disconnect(self, mock_connection, cache_conf):
         mock_connection.credential_provider = UsernamePasswordCredentialProvider()
 
         proxy_connection = CacheProxyConnection(
-            mock_connection, cache, threading.Lock()
+            mock_connection, cache, threading.RLock()
         )
         proxy_connection.disconnect()
 
@@ -492,7 +492,7 @@ def test_read_response_returns_cached_reply(self, mock_cache, mock_connection):
         mock_connection.can_read.return_value = False
 
         proxy_connection = CacheProxyConnection(
-            mock_connection, mock_cache, threading.Lock()
+            mock_connection, mock_cache, threading.RLock()
         )
         proxy_connection.send_command(*["GET", "foo"], **{"keys": ["foo"]})
         assert proxy_connection.read_response() == b"bar"
@@ -554,7 +554,7 @@ def test_triggers_invalidation_processing_on_another_connection(
         mock_connection.can_read.return_value = False
 
         proxy_connection = CacheProxyConnection(
-            mock_connection, mock_cache, threading.Lock()
+            mock_connection, mock_cache, threading.RLock()
         )
         proxy_connection.send_command(*["GET", "foo"], **{"keys": ["foo"]})
 
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
