Support error packet without sqlstate (#1160)

Fix #1156

Author: methane

pymysql/connections.py

@@ -765,8 +765,6 @@ def _read_packet(self, packet_type=MysqlPacket):
                 dump_packet(recv_data)
             buff += recv_data
             # https://dev.mysql.com/doc/internals/en/sending-more-than-16mbyte.html
-            if bytes_to_read == 0xFFFFFF:
-                continue
             if bytes_to_read < MAX_PACKET_LEN:
                 break
 

pymysql/err.py

@@ -136,7 +136,14 @@ def _map_error(exc, *errors):
 
 def raise_mysql_exception(data):
     errno = struct.unpack("<h", data[1:3])[0]
-    errval = data[9:].decode("utf-8", "replace")
+    # https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_basic_err_packet.html
+    # Error packet has optional sqlstate that is 5 bytes and starts with '#'.
+    if data[3] == 0x23:  # '#'
+        # sqlstate = data[4:9].decode()
+        # TODO: Append (sqlstate) in the error message. This will be come in next minor release.
+        errval = data[9:].decode("utf-8", "replace")
+    else:
+        errval = data[3:].decode("utf-8", "replace")
     errorclass = error_map.get(errno)
     if errorclass is None:
         errorclass = InternalError if errno < 1000 else OperationalError

pymysql/tests/test_err.py

@@ -1,14 +1,16 @@
-import unittest
-
+import pytest
 from pymysql import err
 
 
-__all__ = ["TestRaiseException"]
-
+def test_raise_mysql_exception():
+    data = b"\xff\x15\x04#28000Access denied"
+    with pytest.raises(err.OperationalError) as cm:
+        err.raise_mysql_exception(data)
+    assert cm.type == err.OperationalError
+    assert cm.value.args == (1045, "Access denied")
 
-class TestRaiseException(unittest.TestCase):
-    def test_raise_mysql_exception(self):
-        data = b"\xff\x15\x04#28000Access denied"
-        with self.assertRaises(err.OperationalError) as cm:
-            err.raise_mysql_exception(data)
-        self.assertEqual(cm.exception.args, (1045, "Access denied"))
+    data = b"\xff\x10\x04Too many connections"
+    with pytest.raises(err.OperationalError) as cm:
+        err.raise_mysql_exception(data)
+    assert cm.type == err.OperationalError
+    assert cm.value.args == (1040, "Too many connections")

pyproject.toml

@@ -56,6 +56,8 @@ version = {attr = "pymysql.VERSION_STRING"}
 exclude = [
     "pymysql/tests/thirdparty",
 ]
+
+[tool.ruff.lint]
 ignore = ["E721"]
 
 [tool.pdm.dev-dependencies]