Use common parser and server for non-async test_connect.py

Author: kristjanvalur

tests/resp.py

@@ -388,3 +388,25 @@ def parse_chunks(buffers: List[bytes]) -> Tuple[List[Any], bytes]:
                 except NeedMoreData:
                     break
         return result, parser.get_unparsed()
+
+
+class RespServer:
+    """A simple, dummy, REDIS server for unit tests.
+    Accepts RESP commands and returns RESP responses.
+    """
+
+    _CLIENT_NAME = "test-suite-client"
+    _SUCCESS_RESP = b"+OK" + CRNL
+    _ERROR_RESP = b"-ERR" + CRNL
+    _SUPPORTED_CMDS = {f"CLIENT SETNAME {_CLIENT_NAME}": _SUCCESS_RESP}
+
+    def command(self, cmd: Any) -> bytes:
+        """Process a single command and return the response"""
+        if not isinstance(cmd, list):
+            return f"-ERR unknown command {cmd!r}\r\n".encode()
+
+        # currently supports only a single command
+        command = " ".join(cmd)
+        if command in self._SUPPORTED_CMDS:
+            return self._SUPPORTED_CMDS[command]
+        return self._ERROR_RESP

tests/test_connect.py

@@ -1,5 +1,4 @@
 import logging
-import re
 import socket
 import socketserver
 import ssl
@@ -8,16 +7,13 @@
 import pytest
 from redis.connection import Connection, SSLConnection, UnixDomainSocketConnection
 
+from . import resp
 from .ssl_utils import get_ssl_filename
 
 _logger = logging.getLogger(__name__)
 
 
 _CLIENT_NAME = "test-suite-client"
-_CMD_SEP = b"\r\n"
-_SUCCESS_RESP = b"+OK" + _CMD_SEP
-_ERROR_RESP = b"-ERR" + _CMD_SEP
-_SUPPORTED_CMDS = {f"CLIENT SETNAME {_CLIENT_NAME}": _SUCCESS_RESP}
 
 
 @pytest.fixture
@@ -148,44 +144,31 @@ def finish(self):
         _logger.info("%s disconnected", self.client_address)
 
     def handle(self):
+        parser = resp.RespParser()
+        server = resp.RespServer()
         buffer = b""
-        command = None
-        command_ptr = None
-        fragment_length = None
-        while self.server.is_serving() or buffer:
-            try:
-                buffer += self.request.recv(1024)
-            except socket.timeout:
-                continue
-            if not buffer:
-                continue
-            parts = re.split(_CMD_SEP, buffer)
-            buffer = parts[-1]
-            for fragment in parts[:-1]:
-                fragment = fragment.decode()
-                _logger.info("Command fragment: %s", fragment)
-
-                if fragment.startswith("*") and command is None:
-                    command = [None for _ in range(int(fragment[1:]))]
-                    command_ptr = 0
-                    fragment_length = None
-                    continue
-
-                if fragment.startswith("$") and command[command_ptr] is None:
-                    fragment_length = int(fragment[1:])
-                    continue
-
-                assert len(fragment) == fragment_length
-                command[command_ptr] = fragment
-                command_ptr += 1
-
-                if command_ptr < len(command):
+        try:
+            # if client performs pipelining, we may need
+            # to adjust this code to not block when sending
+            # responses.
+            while self.server.is_serving():
+                try:
+                    command = parser.parse(buffer)
+                    buffer = b""
+                except resp.NeedMoreData:
+                    try:
+                        buffer = self.request.recv(1024)
+                    except socket.timeout:
+                        buffer = b""
+                        continue
+                    if not buffer:
+                        break  # EOF
                     continue
-
-                command = " ".join(command)
                 _logger.info("Command %s", command)
-                resp = _SUPPORTED_CMDS.get(command, _ERROR_RESP)
-                _logger.info("Response %s", resp)
-                self.request.sendall(resp)
-                command = None
-        _logger.info("Exit handler")
+                response = server.command(command)
+                _logger.info("Response %s", response)
+                self.request.sendall(response)
+        except Exception:
+            _logger.exception("Exception in handler")
+        finally:
+            _logger.info("Exit handler")