Raise exact mysql error when the packet is an error packet

aiomysql/connection.py

@@ -615,20 +615,21 @@ async def _read_packet(self, packet_type=MysqlPacket):
                 '<HBB', packet_header)
             bytes_to_read = btrl + (btrh << 16)
 
+            error = None
+
             # Outbound and inbound packets are numbered sequentialy, so
             # we increment in both write_packet and read_packet. The count
             # is reset at new COMMAND PHASE.
             if packet_number != self._next_seq_id:
-                self.close()
                 if packet_number == 0:
                     # MySQL 8.0 sends error packet with seqno==0 when shutdown
-                    raise OperationalError(
+                    error = OperationalError(
                         CR.CR_SERVER_LOST,
                         "Lost connection to MySQL server during query")
-
-                raise InternalError(
-                    "Packet sequence number wrong - got %d expected %d" %
-                    (packet_number, self._next_seq_id))
+                else:
+                    error = InternalError(
+                        "Packet sequence number wrong - got %d expected %d" %
+                        (packet_number, self._next_seq_id))
             self._next_seq_id = (self._next_seq_id + 1) % 256
 
             try:
@@ -638,6 +639,16 @@ async def _read_packet(self, packet_type=MysqlPacket):
                 raise
 
             buff += recv_data
+
+            if error is not None:
+                # if packet is error packet
+                if (isinstance(error, InternalError) and recv_data
+                        and (recv_data[0] == 0xFF)):
+                    break
+                else:
+                    self.close()
+                    raise error
+
             # https://dev.mysql.com/doc/internals/en/sending-more-than-16mbyte.html
             if bytes_to_read == 0xffffff:
                 continue

tests/test_connection.py

@@ -1,6 +1,7 @@
 import asyncio
 import gc
 import os
+import time
 
 import pytest
 
@@ -263,3 +264,17 @@ async def test_commit_during_multi_result(connection_creator):
     await cur.execute("SELECT 3;")
     resp = await cur.fetchone()
     assert resp[0] == 3
+
+
+@pytest.mark.run_loop
+async def test_connect_timeout_error(mysql_params, loop):
+    async def mock_block_code():
+        # block a little longer then default connect_timeout variable (10s)
+        time.sleep(11)
+
+    async def connect():
+        mysql_params.pop("ssl", None)
+        await aiomysql.connect(loop=loop, **mysql_params)
+
+    with pytest.raises(aiomysql.OperationalError):
+        await asyncio.gather(connect(), mock_block_code())