Make sure the CI actually runs RESP3 tests (redis#3270)

The CI tests were not running with RESP3 protocol, it was just an
illusion that they do. Fix this, and also preserve coverage and test
artifacts from those runs too.

Some issues have surfaced after the change.

The most notable issue is a bug in hiredis-py, which prevents it
from being used in cluster mode at least. Make sure cluster tests do
not run with hiredis-py. Also make sure some specific unit tests do
not run with hiredis-py.

One other issue with hiredis-py is fixed in this commit. Use a
sentinel object instance to signal lack of data in hiredis-py, instead
of piggybacking of `False`, which can also be returned by parsing
valid RESP payloads.

Some of the unit tests, mostly for modules, were failing, they are now
updated so that they pass.

Remove async parser from test fixture params. Leave the decision for
the async parser to be used in tests to be taken based on the availability
of hiredis-py, and on the protocol that is set for the tests. Otherwise 
when hiredis-py is available we would also run the non-hiredis tests.

.github/workflows/integration.yaml

@@ -115,7 +115,9 @@ jobs:
         python-version: ['3.8', '3.11']
         test-type: ['standalone', 'cluster']
         connection-type: ['hiredis', 'plain']
-        protocol: ['3']
+        exclude:
+          - test-type: 'cluster'
+            connection-type: 'hiredis'
     env:
       ACTIONS_ALLOW_UNSECURE_COMMANDS: true
     name: RESP3 [${{ matrix.python-version }} ${{matrix.test-type}}-${{matrix.connection-type}}]
@@ -134,9 +136,33 @@ jobs:
             pip install hiredis
           fi
           invoke devenv
-          sleep 5 # time to settle
-          invoke ${{matrix.test-type}}-tests
-          invoke ${{matrix.test-type}}-tests --uvloop
+          sleep 10 # time to settle
+          invoke ${{matrix.test-type}}-tests --protocol=3
+          invoke ${{matrix.test-type}}-tests --uvloop --protocol=3
+
+      - uses: actions/upload-artifact@v4
+        if: success() || failure()
+        with:
+          name: pytest-results-${{matrix.test-type}}-${{matrix.connection-type}}-${{matrix.python-version}}-resp3
+          path: '${{matrix.test-type}}*results.xml'
+
+      - name: Upload codecov coverage
+        uses: codecov/codecov-action@v4
+        with:
+          fail_ci_if_error: false
+
+      - name: View Test Results
+        uses: dorny/test-reporter@v1
+        if: success() || failure()
+        continue-on-error: true
+        with:
+          name: Test Results ${{matrix.python-version}} ${{matrix.test-type}}-${{matrix.connection-type}}-resp3
+          path: '*.xml'
+          reporter: java-junit
+          list-suites: all
+          list-tests: all
+          max-annotations: 10
+          fail-on-error: 'false'
 
   build_and_test_package:
     name: Validate building and installing the package

redis/_parsers/hiredis.py

@@ -19,6 +19,11 @@
     SERVER_CLOSED_CONNECTION_ERROR,
 )
 
+# Used to signal that hiredis-py does not have enough data to parse.
+# Using `False` or `None` is not reliable, given that the parser can
+# return `False` or `None` for legitimate reasons from RESP payloads.
+NOT_ENOUGH_DATA = object()
+
 
 class _HiredisReaderArgs(TypedDict, total=False):
     protocolError: Callable[[str], Exception]
@@ -51,25 +56,26 @@ def on_connect(self, connection, **kwargs):
             "protocolError": InvalidResponse,
             "replyError": self.parse_error,
             "errors": connection.encoder.encoding_errors,
+            "notEnoughData": NOT_ENOUGH_DATA,
         }
 
         if connection.encoder.decode_responses:
             kwargs["encoding"] = connection.encoder.encoding
         self._reader = hiredis.Reader(**kwargs)
-        self._next_response = False
+        self._next_response = NOT_ENOUGH_DATA
 
     def on_disconnect(self):
         self._sock = None
         self._reader = None
-        self._next_response = False
+        self._next_response = NOT_ENOUGH_DATA
 
     def can_read(self, timeout):
         if not self._reader:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
 
-        if self._next_response is False:
+        if self._next_response is NOT_ENOUGH_DATA:
             self._next_response = self._reader.gets()
-            if self._next_response is False:
+            if self._next_response is NOT_ENOUGH_DATA:
                 return self.read_from_socket(timeout=timeout, raise_on_timeout=False)
         return True
 
@@ -108,17 +114,17 @@ def read_response(self, disable_decoding=False):
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
 
         # _next_response might be cached from a can_read() call
-        if self._next_response is not False:
+        if self._next_response is not NOT_ENOUGH_DATA:
             response = self._next_response
-            self._next_response = False
+            self._next_response = NOT_ENOUGH_DATA
             return response
 
         if disable_decoding:
             response = self._reader.gets(False)
         else:
             response = self._reader.gets()
 
-        while response is False:
+        while response is NOT_ENOUGH_DATA:
             self.read_from_socket()
             if disable_decoding:
                 response = self._reader.gets(False)
@@ -156,6 +162,7 @@ def on_connect(self, connection):
         kwargs: _HiredisReaderArgs = {
             "protocolError": InvalidResponse,
             "replyError": self.parse_error,
+            "notEnoughData": NOT_ENOUGH_DATA,
         }
         if connection.encoder.decode_responses:
             kwargs["encoding"] = connection.encoder.encoding
@@ -170,7 +177,7 @@ def on_disconnect(self):
     async def can_read_destructive(self):
         if not self._connected:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        if self._reader.gets():
+        if self._reader.gets() is not NOT_ENOUGH_DATA:
             return True
         try:
             async with async_timeout(0):
@@ -200,7 +207,7 @@ async def read_response(
             response = self._reader.gets(False)
         else:
             response = self._reader.gets()
-        while response is False:
+        while response is NOT_ENOUGH_DATA:
             await self.read_from_socket()
             if disable_decoding:
                 response = self._reader.gets(False)

tests/test_asyncio/conftest.py

@@ -6,13 +6,11 @@
 import pytest_asyncio
 import redis.asyncio as redis
 from packaging.version import Version
-from redis._parsers import _AsyncHiredisParser, _AsyncRESP2Parser
 from redis.asyncio import Sentinel
 from redis.asyncio.client import Monitor
 from redis.asyncio.connection import Connection, parse_url
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
-from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import REDIS_INFO
 
 from .compat import mock
@@ -28,41 +26,21 @@ async def _get_info(redis_url):
 @pytest_asyncio.fixture(
     params=[
         pytest.param(
-            (True, _AsyncRESP2Parser),
+            (True,),
             marks=pytest.mark.skipif(
                 'config.REDIS_INFO["cluster_enabled"]', reason="cluster mode enabled"
             ),
         ),
-        (False, _AsyncRESP2Parser),
-        pytest.param(
-            (True, _AsyncHiredisParser),
-            marks=[
-                pytest.mark.skipif(
-                    'config.REDIS_INFO["cluster_enabled"]',
-                    reason="cluster mode enabled",
-                ),
-                pytest.mark.skipif(
-                    not HIREDIS_AVAILABLE, reason="hiredis is not installed"
-                ),
-            ],
-        ),
-        pytest.param(
-            (False, _AsyncHiredisParser),
-            marks=pytest.mark.skipif(
-                not HIREDIS_AVAILABLE, reason="hiredis is not installed"
-            ),
-        ),
+        (False,),
     ],
     ids=[
-        "single-python-parser",
-        "pool-python-parser",
-        "single-hiredis",
-        "pool-hiredis",
+        "single",
+        "pool",
     ],
 )
 async def create_redis(request):
     """Wrapper around redis.create_redis."""
-    single_connection, parser_cls = request.param
+    (single_connection,) = request.param
 
     teardown_clients = []
 
@@ -78,10 +56,9 @@ async def client_factory(
         cluster_mode = REDIS_INFO["cluster_enabled"]
         if not cluster_mode:
             single = kwargs.pop("single_connection_client", False) or single_connection
-            parser_class = kwargs.pop("parser_class", None) or parser_cls
             url_options = parse_url(url)
             url_options.update(kwargs)
-            pool = redis.ConnectionPool(parser_class=parser_class, **url_options)
+            pool = redis.ConnectionPool(**url_options)
             client = cls(connection_pool=pool)
         else:
             client = redis.RedisCluster.from_url(url, **kwargs)