Repository: redis/redis-py

Issue:
[Bug] Aggregation Request Generates Incorrect Command Arguments for LIMIT 
**Version**: redis-py 4.3.4, Redis 7.0 

**Platform**: Python 3.8 Ubuntu 

**Description**: 
On the current master branch, it says in comments that if limit is applied before group by, it will apply the limit to the initial query. If it is applied after group by, it will be applied to the result of the aggregation. Based on the example, you can also specify a limit at multiple stages in the pipeline. However, in practice, the limit is only set on the last stage of the aggregation. 

For example, given the following code on a very large database
```
AggregateRequest("*").limit(0, 100).group_by('@my_field', reducers.count()).limit(0, 1000)
```
You would expect at most 100 results to be returned, since the first limit should limit the amount of rows the aggregation request queries initially. However, 1000 results are returned, and based on runtime, the original query isn't limited at all. 

**Why it happens**:
This happens because `AggregateRequest` generates an incorrect list of arguments for LIMIT. Given the above query, you would expect the following arguments to be generated: 
```
['LIMIT', '0', '100', 'GROUPBY', '1', '@future_type', 'REDUCE', 'COUNT', '0', 'LIMIT', '0, '1000']
```
I have checked that running FT Aggregate with the above command produces the expected result. However, the arguments that are actually generated are 
```
['*', 'GROUPBY', '1', '@future_type', 'REDUCE', 'COUNT', '0', 'LIMIT', '0', '1000']
```
Note that there is only one limit in the arguments. The reason incorrect arguments are generated is because when we call `.limit()` on an AggregationRequest, it **overwrites** any limit previously set (at [this location](https://github.com/redis/redis-py/blob/master/redis/commands/search/aggregation.py#L214)). Furthermore, when we generate the arguments, we always append the `LIMIT` argument to the end of the query (line 326). The expected behavior would be to insert the command arguments for LIMIT in corresponding location in the aggregation request. 

**Fix Suggestion**:
I'm no expert in redis-py, but after looking at the code, I believe we would want to replace line 214 in aggregations.py: 
```
        self._limit = Limit(offset, num)
```
with the following: 
```
_limit = Limit(offset, num)
self._aggregateplan.extend(_limit.build_args())
```
And remove lines 107 and 326. 

**I believe at minimum, the documentation should be updated so that it reflects the current incorrect behavior**

Discussion:
codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2393?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Base: **92.25**% // Head: **92.24**% // Decreases project coverage by **`-0.01%`** :warning:
> Coverage data is based on head [(`c2a376e`)](https://codecov.io/gh/redis/redis-py/pull/2393?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) compared to base [(`947b8e8`)](https://codecov.io/gh/redis/redis-py/commit/947b8e8595e3aff4aeb3eb3ebbe56b86fb5e76fb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> Patch coverage: 100.00% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #2393      +/-   ##
==========================================
- Coverage   92.25%   92.24%   -0.02%     
==========================================
  Files         110      110              
  Lines       28906    28905       -1     
==========================================
- Hits        26668    26662       -6     
- Misses       2238     2243       +5     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2393?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/commands/search/aggregation.py](https://codecov.io/gh/redis/redis-py/pull/2393/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29tbWFuZHMvc2VhcmNoL2FnZ3JlZ2F0aW9uLnB5) | `78.87% <100.00%> (-0.85%)` | :arrow_down: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2393/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `97.28% <0.00%> (-0.14%)` | :arrow_down: |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2393/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `96.96% <0.00%> (-0.12%)` | :arrow_down: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2393?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).


Patch:
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
index b92c52f..e82e7e1 100644
--- a/.github/workflows/codeql-analysis.yml
+++ b/.github/workflows/codeql-analysis.yml
@@ -36,11 +36,11 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v2
+      uses: actions/checkout@v3
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -51,7 +51,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild
-      uses: github/codeql-action/autobuild@v1
+      uses: github/codeql-action/autobuild@v2
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -65,4 +65,4 @@ jobs:
     #   make release
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 1089b27..0003930 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -31,9 +31,8 @@ jobs:
      name: Code linters
      runs-on: ubuntu-latest
      steps:
-       - uses: actions/checkout@v2
-       - name: install python
-         uses: actions/setup-python@v4
+       - uses: actions/checkout@v3
+       - uses: actions/setup-python@v4
          with:
            python-version: 3.9
            cache: 'pip'
@@ -55,9 +54,8 @@ jobs:
        ACTIONS_ALLOW_UNSECURE_COMMANDS: true
      name: Python ${{ matrix.python-version }} ${{matrix.test-type}}-${{matrix.connection-type}} tests
      steps:
-       - uses: actions/checkout@v2
-       - name: install python
-         uses: actions/setup-python@v4
+       - uses: actions/checkout@v3
+       - uses: actions/setup-python@v4
          with:
            python-version: ${{ matrix.python-version }}
            cache: 'pip'
@@ -93,9 +91,8 @@ jobs:
       matrix:
         extension: ['tar.gz', 'whl']
     steps:
-      - uses: actions/checkout@v2
-      - name: install python
-        uses: actions/setup-python@v3
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
         with:
           python-version: 3.9
       - name: Run installed unit tests
@@ -109,9 +106,8 @@ jobs:
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
     steps:
-      - uses: actions/checkout@v2
-      - name: install python ${{ matrix.python-version }}
-        uses: actions/setup-python@v3
+      - uses: actions/checkout@v3
+      - uses: actions/setup-python@v4
         with:
           python-version: ${{ matrix.python-version }}
           cache: 'pip'
diff --git a/.github/workflows/pypi-publish.yaml b/.github/workflows/pypi-publish.yaml
index 0d100dd..3e7f801 100644
--- a/.github/workflows/pypi-publish.yaml
+++ b/.github/workflows/pypi-publish.yaml
@@ -9,9 +9,9 @@ jobs:
   build_and_package:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v3
       - name: install python
-        uses: actions/setup-python@v3
+        uses: actions/setup-python@v4
         with:
           python-version: 3.9
       - name: Install dev tools
diff --git a/CHANGES b/CHANGES
index 47600cf..75fa83c 100644
--- a/CHANGES
+++ b/CHANGES
@@ -19,6 +19,7 @@
     * Fix reusing the old nodes' connections when cluster topology refresh is being done
     * Fix RedisCluster to immediately raise AuthenticationError without a retry
     * ClusterPipeline Doesn't Handle ConnectionError for Dead Hosts (#2225)
+    * Remove compatibility code for old versions of Hiredis, drop Packaging dependency
 
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)
diff --git a/README.md b/README.md
index c7b839d..8b7333e 100644
--- a/README.md
+++ b/README.md
@@ -27,7 +27,8 @@ To install redis-py, simply:
 $ pip install redis
 ```
 
-For faster performance, install redis with hiredis support, this provides a compiled response parser, and *for most cases* requires zero code changes. By default, if hiredis is available, redis-py will attempt to use it for response parsing.
+For faster performance, install redis with hiredis support, this provides a compiled response parser, and *for most cases* requires zero code changes.
+By default, if hiredis >= 1.0 is available, redis-py will attempt to use it for response parsing.
 
 ``` bash
 $ pip install redis[hiredis]
diff --git a/dev_requirements.txt b/dev_requirements.txt
index 31ae26e..66be6d1 100644
--- a/dev_requirements.txt
+++ b/dev_requirements.txt
@@ -4,6 +4,7 @@ flake8==4.0.1
 flynt~=0.69.0
 isort==5.10.1
 mock==4.0.3
+packaging>=20.4
 pytest==6.2.5
 pytest-timeout==2.0.1
 pytest-asyncio>=0.16.0
diff --git a/docs/examples/search_vector_similarity_examples.ipynb b/docs/examples/search_vector_similarity_examples.ipynb
index 42c03f0..2b02610 100644
--- a/docs/examples/search_vector_similarity_examples.ipynb
+++ b/docs/examples/search_vector_similarity_examples.ipynb
@@ -74,7 +74,7 @@
     "r.hset(\"b\", \"v\", \"aaaabaaa\")\n",
     "r.hset(\"c\", \"v\", \"aaaaabaa\")\n",
     "\n",
-    "q = Query(\"*=>[KNN 2 @v $vec]\").return_field(\"__v_score\")\n",
+    "q = Query(\"*=>[KNN 2 @v $vec]\").return_field(\"__v_score\").dialect(2)\n",
     "r.ft().search(q, query_params={\"vec\": \"aaaaaaaa\"})"
    ]
   }
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 78dcd05..8d34b9a 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -390,7 +390,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         # regardless of the server type. If this is a primary connection,
         # READONLY would not affect executing write commands.
         await connection.send_command("READONLY")
-        if str_if_bytes(await connection.read_response_without_lock()) != "OK":
+        if str_if_bytes(await connection.read_response()) != "OK":
             raise ConnectionError("READONLY command failed")
 
     def get_nodes(self) -> List["ClusterNode"]:
@@ -866,11 +866,9 @@ class ClusterNode:
     ) -> Any:
         try:
             if NEVER_DECODE in kwargs:
-                response = await connection.read_response_without_lock(
-                    disable_decoding=True
-                )
+                response = await connection.read_response(disable_decoding=True)
             else:
-                response = await connection.read_response_without_lock()
+                response = await connection.read_response()
         except ResponseError:
             if EMPTY_RESPONSE in kwargs:
                 return kwargs[EMPTY_RESPONSE]
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index db8c240..16f33e2 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -671,7 +671,6 @@ class Connection:
         self.set_parser(parser_class)
         self._connect_callbacks: List[weakref.WeakMethod[ConnectCallbackT]] = []
         self._buffer_cutoff = 6000
-        self._lock = asyncio.Lock()
 
     def __repr__(self):
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
@@ -940,39 +939,6 @@ class Connection:
             )
 
     async def read_response(self, disable_decoding: bool = False):
-        """Read the response from a previously sent command"""
-        try:
-            async with self._lock:
-                if self.socket_timeout:
-                    async with async_timeout.timeout(self.socket_timeout):
-                        response = await self._parser.read_response(
-                            disable_decoding=disable_decoding
-                        )
-                else:
-                    response = await self._parser.read_response(
-                        disable_decoding=disable_decoding
-                    )
-        except asyncio.TimeoutError:
-            await self.disconnect()
-            raise TimeoutError(f"Timeout reading from {self.host}:{self.port}")
-        except OSError as e:
-            await self.disconnect()
-            raise ConnectionError(
-                f"Error while reading from {self.host}:{self.port} : {e.args}"
-            )
-        except BaseException:
-            await self.disconnect()
-            raise
-
-        if self.health_check_interval:
-            next_time = asyncio.get_running_loop().time() + self.health_check_interval
-            self.next_health_check = next_time
-
-        if isinstance(response, ResponseError):
-            raise response from None
-        return response
-
-    async def read_response_without_lock(self, disable_decoding: bool = False):
         """Read the response from a previously sent command"""
         try:
             if self.socket_timeout:
@@ -1241,7 +1207,6 @@ class UnixDomainSocketConnection(Connection):  # lgtm [py/missing-call-to-init]
         self.set_parser(parser_class)
         self._connect_callbacks = []
         self._buffer_cutoff = 6000
-        self._lock = asyncio.Lock()
 
     def repr_pieces(self) -> Iterable[Tuple[str, Union[str, int]]]:
         pieces = [("path", self.path), ("db", self.db)]
diff --git a/redis/commands/search/aggregation.py b/redis/commands/search/aggregation.py
index 061e69c..a171fa1 100644
--- a/redis/commands/search/aggregation.py
+++ b/redis/commands/search/aggregation.py
@@ -104,7 +104,6 @@ class AggregateRequest:
         self._aggregateplan = []
         self._loadfields = []
         self._loadall = False
-        self._limit = Limit()
         self._max = 0
         self._with_schema = False
         self._verbatim = False
@@ -211,7 +210,8 @@ class AggregateRequest:
         `sort_by()` instead.
 
         """
-        self._limit = Limit(offset, num)
+        _limit = Limit(offset, num)
+        self._aggregateplan.extend(_limit.build_args())
         return self
 
     def sort_by(self, *fields, **kwargs):
@@ -323,8 +323,6 @@ class AggregateRequest:
 
         ret.extend(self._aggregateplan)
 
-        ret += self._limit.build_args()
-
         return ret
 
 
diff --git a/redis/connection.py b/redis/connection.py
index 3281b14..491df8e 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -10,8 +10,6 @@ from queue import Empty, Full, LifoQueue
 from time import time
 from urllib.parse import parse_qs, unquote, urlparse
 
-from packaging.version import Version
-
 from redis.backoff import NoBackoff
 from redis.exceptions import (
     AuthenticationError,
@@ -54,16 +52,6 @@ NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys())
 if HIREDIS_AVAILABLE:
     import hiredis
 
-    hiredis_version = Version(hiredis.__version__)
-    HIREDIS_SUPPORTS_CALLABLE_ERRORS = hiredis_version >= Version("0.1.3")
-    HIREDIS_SUPPORTS_BYTE_BUFFER = hiredis_version >= Version("0.1.4")
-    HIREDIS_SUPPORTS_ENCODING_ERRORS = hiredis_version >= Version("1.0.0")
-
-    HIREDIS_USE_BYTE_BUFFER = True
-    # only use byte buffer if hiredis supports it
-    if not HIREDIS_SUPPORTS_BYTE_BUFFER:
-        HIREDIS_USE_BYTE_BUFFER = False
-
 SYM_STAR = b"*"
 SYM_DOLLAR = b"$"
 SYM_CRLF = b"\r\n"
@@ -380,9 +368,7 @@ class HiredisParser(BaseParser):
         if not HIREDIS_AVAILABLE:
             raise RedisError("Hiredis is not installed")
         self.socket_read_size = socket_read_size
-
-        if HIREDIS_USE_BYTE_BUFFER:
-            self._buffer = bytearray(socket_read_size)
+        self._buffer = bytearray(socket_read_size)
 
     def __del__(self):
         try:
@@ -393,16 +379,14 @@ class HiredisParser(BaseParser):
     def on_connect(self, connection, **kwargs):
         self._sock = connection._sock
         self._socket_timeout = connection.socket_timeout
-        kwargs = {"protocolError": InvalidResponse, "replyError": self.parse_error}
-
-        # hiredis < 0.1.3 doesn't support functions that create exceptions
-        if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
-            kwargs["replyError"] = ResponseError
+        kwargs = {
+            "protocolError": InvalidResponse,
+            "replyError": self.parse_error,
+            "errors": connection.encoder.encoding_errors,
+        }
 
         if connection.encoder.decode_responses:
             kwargs["encoding"] = connection.encoder.encoding
-        if HIREDIS_SUPPORTS_ENCODING_ERRORS:
-            kwargs["errors"] = connection.encoder.encoding_errors
         self._reader = hiredis.Reader(**kwargs)
         self._next_response = False
 
@@ -427,17 +411,10 @@ class HiredisParser(BaseParser):
         try:
             if custom_timeout:
                 sock.settimeout(timeout)
-            if HIREDIS_USE_BYTE_BUFFER:
-                bufflen = self._sock.recv_into(self._buffer)
-                if bufflen == 0:
-                    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-                self._reader.feed(self._buffer, 0, bufflen)
-            else:
-                buffer = self._sock.recv(self.socket_read_size)
-                # an empty string indicates the server shutdown the socket
-                if not isinstance(buffer, bytes) or len(buffer) == 0:
-                    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-                self._reader.feed(buffer)
+            bufflen = self._sock.recv_into(self._buffer)
+            if bufflen == 0:
+                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+            self._reader.feed(self._buffer, 0, bufflen)
             # data was read from the socket and added to the buffer.
             # return True to indicate that data was read.
             return True
@@ -479,17 +456,6 @@ class HiredisParser(BaseParser):
                 response = self._reader.gets(False)
             else:
                 response = self._reader.gets()
-        # if an older version of hiredis is installed, we need to attempt
-        # to convert ResponseErrors to their appropriate types.
-        if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
-            if isinstance(response, ResponseError):
-                response = self.parse_error(response.args[0])
-            elif (
-                isinstance(response, list)
-                and response
-                and isinstance(response[0], ResponseError)
-            ):
-                response[0] = self.parse_error(response[0].args[0])
         # if the response is a ConnectionError or the response is a list and
         # the first item is a ConnectionError, raise it as something bad
         # happened
diff --git a/redis/utils.py b/redis/utils.py
index 0c34e1e..25d2491 100644
--- a/redis/utils.py
+++ b/redis/utils.py
@@ -4,7 +4,8 @@ from typing import Any, Dict, Mapping, Union
 try:
     import hiredis  # noqa
 
-    HIREDIS_AVAILABLE = True
+    # Only support Hiredis >= 1.0:
+    HIREDIS_AVAILABLE = not hiredis.__version__.startswith("0.")
 except ImportError:
     HIREDIS_AVAILABLE = False
 
diff --git a/requirements.txt b/requirements.txt
index c40eca7..b764668 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,3 @@
 async-timeout>=4.0.2
 deprecated>=1.2.3
-packaging>=20.4
 typing-extensions; python_version<"3.8"
diff --git a/setup.py b/setup.py
index fb6d32a..e08251f 100644
--- a/setup.py
+++ b/setup.py
@@ -33,7 +33,6 @@ setup(
     python_requires=">=3.7",
     install_requires=[
         "deprecated>=1.2.3",
-        "packaging>=20.4",
         'importlib-metadata >= 1.0; python_version < "3.8"',
         'typing-extensions; python_version<"3.8"',
         "async-timeout>=4.0.2",


Test patch:
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index e299395..88cfb1f 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -120,7 +120,7 @@ async def get_mocked_redis_client(*args, **kwargs) -> RedisCluster:
 def mock_node_resp(node: ClusterNode, response: Any) -> ClusterNode:
     connection = mock.AsyncMock()
     connection.is_connected = True
-    connection.read_response_without_lock.return_value = response
+    connection.read_response.return_value = response
     while node._free:
         node._free.pop()
     node._free.append(connection)
@@ -130,7 +130,7 @@ def mock_node_resp(node: ClusterNode, response: Any) -> ClusterNode:
 def mock_node_resp_exc(node: ClusterNode, exc: Exception) -> ClusterNode:
     connection = mock.AsyncMock()
     connection.is_connected = True
-    connection.read_response_without_lock.side_effect = exc
+    connection.read_response.side_effect = exc
     while node._free:
         node._free.pop()
     node._free.append(connection)
@@ -275,16 +275,12 @@ class TestRedisClusterObj:
         for node in rc.get_nodes():
             assert node.max_connections == 10
 
-        with mock.patch.object(
-            Connection, "read_response_without_lock"
-        ) as read_response_without_lock:
+        with mock.patch.object(Connection, "read_response") as read_response:
 
-            async def read_response_without_lock_mocked(
-                *args: Any, **kwargs: Any
-            ) -> None:
+            async def read_response_mocked(*args: Any, **kwargs: Any) -> None:
                 await asyncio.sleep(10)
 
-            read_response_without_lock.side_effect = read_response_without_lock_mocked
+            read_response.side_effect = read_response_mocked
 
             with pytest.raises(MaxConnectionsError):
                 await asyncio.gather(
@@ -316,10 +312,10 @@ class TestRedisClusterObj:
         assert await r.ping(target_nodes=RedisCluster.PRIMARIES) is True
         for primary in primaries:
             conn = primary._free.pop()
-            assert conn.read_response_without_lock.called is True
+            assert conn.read_response.called is True
         for replica in replicas:
             conn = replica._free.pop()
-            assert conn.read_response_without_lock.called is not True
+            assert conn.read_response.called is not True
 
     async def test_execute_command_node_flag_replicas(self, r: RedisCluster) -> None:
         """
@@ -333,10 +329,10 @@ class TestRedisClusterObj:
         assert await r.ping(target_nodes=RedisCluster.REPLICAS) is True
         for replica in replicas:
             conn = replica._free.pop()
-            assert conn.read_response_without_lock.called is True
+            assert conn.read_response.called is True
         for primary in primaries:
             conn = primary._free.pop()
-            assert conn.read_response_without_lock.called is not True
+            assert conn.read_response.called is not True
 
         await r.close()
 
@@ -348,7 +344,7 @@ class TestRedisClusterObj:
         assert await r.ping(target_nodes=RedisCluster.ALL_NODES) is True
         for node in r.get_nodes():
             conn = node._free.pop()
-            assert conn.read_response_without_lock.called is True
+            assert conn.read_response.called is True
 
     async def test_execute_command_node_flag_random(self, r: RedisCluster) -> None:
         """
@@ -359,7 +355,7 @@ class TestRedisClusterObj:
         called_count = 0
         for node in r.get_nodes():
             conn = node._free.pop()
-            if conn.read_response_without_lock.called is True:
+            if conn.read_response.called is True:
                 called_count += 1
         assert called_count == 1
 
@@ -372,7 +368,7 @@ class TestRedisClusterObj:
         mock_node_resp(def_node, "PONG")
         assert await r.ping() is True
         conn = def_node._free.pop()
-        assert conn.read_response_without_lock.called
+        assert conn.read_response.called
 
     async def test_ask_redirection(self, r: RedisCluster) -> None:
         """
@@ -516,7 +512,7 @@ class TestRedisClusterObj:
         with mock.patch.multiple(
             Connection,
             send_command=mock.DEFAULT,
-            read_response_without_lock=mock.DEFAULT,
+            read_response=mock.DEFAULT,
             _connect=mock.DEFAULT,
             can_read=mock.DEFAULT,
             on_connect=mock.DEFAULT,
@@ -548,7 +544,7 @@ class TestRedisClusterObj:
                 # so we'll mock some of the Connection's functions to allow it
                 execute_command.side_effect = execute_command_mock_first
                 mocks["send_command"].return_value = True
-                mocks["read_response_without_lock"].return_value = "OK"
+                mocks["read_response"].return_value = "OK"
                 mocks["_connect"].return_value = True
                 mocks["can_read"].return_value = False
                 mocks["on_connect"].return_value = True
@@ -857,8 +853,8 @@ class TestClusterRedisCommands:
         node0 = r.get_node(default_host, 7000)
         node1 = r.get_node(default_host, 7001)
         assert await r.cluster_delslots(0, 8192) == [True, True]
-        assert node0._free.pop().read_response_without_lock.called
-        assert node1._free.pop().read_response_without_lock.called
+        assert node0._free.pop().read_response.called
+        assert node1._free.pop().read_response.called
 
         await r.close()
 
@@ -1027,7 +1023,7 @@ class TestClusterRedisCommands:
         node = r.nodes_manager.get_node_from_slot(12182)
         mock_node_resp(node, "OK")
         assert await r.cluster_setslot_stable(12182) is True
-        assert node._free.pop().read_response_without_lock.called
+        assert node._free.pop().read_response.called
 
     @skip_if_redis_enterprise()
     async def test_cluster_replicas(self, r: RedisCluster) -> None:
@@ -1069,7 +1065,7 @@ class TestClusterRedisCommands:
         for res in all_replicas_results.values():
             assert res is True
         for replica in r.get_replicas():
-            assert replica._free.pop().read_response_without_lock.called
+            assert replica._free.pop().read_response.called
 
         await r.close()
 
@@ -1082,7 +1078,7 @@ class TestClusterRedisCommands:
         for res in all_replicas_results.values():
             assert res is True
         for replica in r.get_replicas():
-            assert replica._free.pop().read_response_without_lock.called
+            assert replica._free.pop().read_response.called
 
         await r.close()
 
@@ -2441,8 +2437,8 @@ class TestClusterPipeline:
             mock_node_resp_exc(first_node, AskError(ask_msg))
             mock_node_resp(ask_node, "MOCK_OK")
             res = await pipe.get(key).execute()
-            assert first_node._free.pop().read_response_without_lock.await_count
-            assert ask_node._free.pop().read_response_without_lock.await_count
+            assert first_node._free.pop().read_response.await_count
+            assert ask_node._free.pop().read_response.await_count
             assert res == ["MOCK_OK"]
 
     async def test_moved_redirection_on_slave_with_default(
@@ -2497,7 +2493,7 @@ class TestClusterPipeline:
             executed_on_replica = False
             for node in slot_nodes:
                 if node.server_type == REPLICA:
-                    if node._free.pop().read_response_without_lock.await_count:
+                    if node._free.pop().read_response.await_count:
                         executed_on_replica = True
                         break
             assert executed_on_replica
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 419c8a7..674a1b9 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -64,6 +64,10 @@ async def test_socket_param_regression(r):
 
 
 async def test_can_run_concurrent_commands(r):
+    if getattr(r, "connection", None) is not None:
+        # Concurrent commands are only supported on pooled or cluster connections
+        # since there is no synchronization on a single connection.
+        pytest.skip("pool only")
     assert await r.ping() is True
     assert all(await asyncio.gather(*(r.ping() for _ in range(10))))