Fix SSL connection handshake charset not respecting client configuration

CHANGES.txt

@@ -1,6 +1,11 @@
 Changes
 -------
 
+next (unreleased)
+^^^^^^^^^^^^^^^^^
+
+* Fix SSL connection handshake charset not respecting client configuration #776
+
 0.1.0 (2022-04-11)
 ^^^^^^^^^^^^^^^^^^
 

aiomysql/connection.py

@@ -733,12 +733,12 @@ async def _request_authentication(self):
         if self.user is None:
             raise ValueError("Did not specify a username")
 
-        if self._ssl_context and self.server_capabilities & CLIENT.SSL:
-            # capablities, max packet, charset
-            data = struct.pack('<IIB', self.client_flag, 16777216, 33)
-            data += b'\x00' * (32 - len(data))
+        charset_id = charset_by_name(self.charset).id
+        data_init = struct.pack('<iIB23s', self.client_flag, MAX_PACKET_LEN,
+                                charset_id, b'')
 
-            self.write_packet(data)
+        if self._ssl_context and self.server_capabilities & CLIENT.SSL:
+            self.write_packet(data_init)
 
             # Stop sending events to data_received
             self._writer.transport.pause_reading()
@@ -762,15 +762,11 @@ async def _request_authentication(self):
 
             self._secure = True
 
-        charset_id = charset_by_name(self.charset).id
         if isinstance(self.user, str):
             _user = self.user.encode(self.encoding)
         else:
             _user = self.user
 
-        data_init = struct.pack('<iIB23s', self.client_flag, MAX_PACKET_LEN,
-                                charset_id, b'')
-
         data = data_init + _user + b'\0'
 
         authresp = b''

tests/test_basic.py

@@ -106,6 +106,20 @@ async def test_string(cursor, table_cleanup):
     assert (test_value,) == r
 
 
+@pytest.mark.run_loop
+async def test_string_with_emoji(cursor, table_cleanup):
+    await cursor.execute("DROP TABLE IF EXISTS test_string_with_emoji;")
+    await cursor.execute("CREATE TABLE test_string_with_emoji (a text) "
+                         "DEFAULT CHARACTER SET=\"utf8mb4\"")
+    test_value = "I am a test string with emoji 😄"
+    table_cleanup('test_string_with_emoji')
+    await cursor.execute("INSERT INTO test_string_with_emoji (a) VALUES (%s)",
+                         test_value)
+    await cursor.execute("SELECT a FROM test_string_with_emoji")
+    r = await cursor.fetchone()
+    assert (test_value,) == r
+
+
 @pytest.mark.run_loop
 async def test_integer(cursor, table_cleanup):
     await cursor.execute("CREATE TABLE test_integer (a INTEGER)")