use common parser and server for test_asyncio/test_connect.py

Author: kristjanvalur

tests/test_asyncio/test_connect.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import re
 import socket
 import ssl
 
@@ -11,16 +10,13 @@
     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
@@ -102,46 +98,34 @@ async def _handler(reader, writer):
 
 
 async def _redis_request_handler(reader, writer, stop_event):
+    parser = resp.RespParser()
+    server = resp.RespServer()
     buffer = b""
-    command = None
-    command_ptr = None
-    fragment_length = None
-    while not stop_event.is_set() or buffer:
-        _logger.info(str(stop_event.is_set()))
-        try:
-            buffer += await asyncio.wait_for(reader.read(1024), timeout=0.5)
-        except TimeoutError:
-            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 not stop_event.is_set() or buffer:
+            _logger.info(str(stop_event.is_set()))
+            try:
+                command = parser.parse(buffer)
+                buffer = b""
+            except resp.NeedMoreData:
+                try:
+                    buffer = await asyncio.wait_for(reader.read(1024), timeout=0.5)
+                except TimeoutError:
+                    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 from %s", resp)
-            writer.write(resp)
+            response = server.command(command)
+            _logger.info("Response %s", response)
+            writer.write(response)
             await writer.drain()
-            command = None
-    _logger.info("Exit handler")
+    except Exception:
+        _logger.exception("Error in handler")
+    finally:
+        _logger.info("Exit handler")