Fixing check for correct parser when protocol is 3

Author: petyaslavova

redis/connection.py

@@ -386,6 +386,13 @@ def _configure_maintenance_notifications(
                 "To configure maintenance notifications, a parser must be provided!"
             )
 
+        if not isinstance(parser, _HiredisParser) and not isinstance(
+            parser, _RESP3Parser
+        ):
+            raise RedisError(
+                "Maintenance notifications are only supported with hiredis and RESP3 parsers!"
+            )
+
         if maint_notifications_pool_handler:
             # Extract a reference to a new pool handler that copies all properties
             # of the original one and has a different connection reference
@@ -741,7 +748,12 @@ def __init__(
                 raise ConnectionError("protocol must be either 2 or 3")
                 # p = DEFAULT_RESP_VERSION
             self.protocol = p
-        if self.protocol == 3 and parser_class == DefaultParser:
+        if self.protocol == 3 and parser_class == _RESP2Parser:
+            # If the protocol is 3 but the parser is RESP2, change it to RESP3
+            # This is needed because the parser might be set before the protocol
+            # or might be provided as a kwarg to the constructor
+            # We need to react on discrepancy only for RESP2 and RESP3
+            # as hiredis supports both
             parser_class = _RESP3Parser
         self.set_parser(parser_class)
 

tests/test_maint_notifications_handling.py

@@ -304,6 +304,38 @@ def recv(self, bufsize):
 
             raise BlockingIOError(errno.EAGAIN, "Resource temporarily unavailable")
 
+    def recv_into(self, buffer, nbytes=0):
+        """
+        Receive data from Redis and write it into the provided buffer.
+        Returns the number of bytes written.
+
+        This method is used by the hiredis parser for efficient data reading.
+        """
+        if self.closed:
+            raise ConnectionError("Socket is closed")
+
+        # Use pending responses that were prepared when commands were sent
+        if self.pending_responses:
+            response = self.pending_responses.pop(0)
+            if b"MOVING" in response:
+                self.moving_sent = True
+
+            # Determine how many bytes to write
+            if nbytes == 0:
+                nbytes = len(buffer)
+
+            # Write data into the buffer (up to nbytes or response length)
+            bytes_to_write = min(len(response), nbytes, len(buffer))
+            buffer[:bytes_to_write] = response[:bytes_to_write]
+
+            return bytes_to_write
+        else:
+            # No data available - this should block or raise an exception
+            # For can_read checks, we should indicate no data is available
+            import errno
+
+            raise BlockingIOError(errno.EAGAIN, "Resource temporarily unavailable")
+
     def fileno(self):
         """Return a fake file descriptor for select/poll operations."""
         return 1  # Fake file descriptor