BUG19500097: Fix string decoding with binary character set

peeyushgupta · peeyushgupta · commit dcd5b6301a1f · 2014-10-01T16:31:49.000+05:30
When connection character set is set to 'binary', the string data
failed to decode since it is not a valid Python character set.
We fix this by returning string data without decoding when the
character set is set to 'binary'.

A unit test has been added for BUG#19500097.
diff --git a/lib/mysql/connector/connection.py b/lib/mysql/connector/connection.py
@@ -1135,7 +1135,7 @@ def python_charset(self):
         Returns a string.
         """
         encoding = CharacterSet.get_info(self._charset_id)[0]
-        if encoding == 'utf8mb4':
+        if encoding in ('utf8mb4', 'binary'):
             return 'utf8'
         else:
             return encoding
diff --git a/lib/mysql/connector/conversion.py b/lib/mysql/connector/conversion.py
@@ -178,10 +178,15 @@ def _str_to_mysql(self, value):
 
     def _unicode_to_mysql(self, value):
         """Convert unicode"""
-        encoded = value.encode(self.charset)
-        if self.charset_id in CharacterSet.slash_charsets:
+        charset = self.charset
+        charset_id = self.charset_id
+        if charset == 'binary':
+            charset = 'utf8'
+            charset_id = CharacterSet.get_charset_info(charset)[0]
+        encoded = value.encode(charset)
+        if charset_id in CharacterSet.slash_charsets:
             if b'\x5c' in encoded:
-                return HexLiteral(value, self.charset)
+                return HexLiteral(value, charset)
         return encoded
 
     def _bytes_to_mysql(self, value):
@@ -537,6 +542,8 @@ def _STRING_to_python(self, value, dsc=None):  # pylint: disable=C0103
             if dsc[7] & FieldFlag.BINARY:
                 return value
 
+        if self.charset == 'binary':
+            return value
         if isinstance(value, (bytes, bytearray)) and self.use_unicode:
             return value.decode(self.charset)
 
diff --git a/tests/test_bugs.py b/tests/test_bugs.py
@@ -2960,3 +2960,48 @@ def test_row_to_python(self):
         cur.execute("INSERT INTO {0} (c1) VALUES (?)".format(self.tbl), (data,))
         self.cur.execute("SELECT * FROM {0}".format(self.tbl))
         self.assertEqual((data,), self.cur.fetchone())
+
+
+class BugOra19500097(tests.MySQLConnectorTests):
+    """BUG#19500097: BETTER SUPPORT FOR RAW/BINARY DATA
+    """
+    def setUp(self):
+        config = tests.get_mysql_config()
+        self.cnx = connection.MySQLConnection(**config)
+        self.cur = self.cnx.cursor()
+
+        self.tbl = 'Bug19500097'
+        self.cur.execute("DROP TABLE IF EXISTS {0}".format(self.tbl))
+
+        create = ("CREATE TABLE {0} (col1 VARCHAR(10), col2 INT) "
+                  "DEFAULT CHARSET latin1".format(self.tbl))
+        self.cur.execute(create)
+
+    def tearDown(self):
+        self.cur.execute("DROP TABLE IF EXISTS {0}".format(self.tbl))
+        self.cur.close()
+        self.cnx.close()
+
+    def test_binary_charset(self):
+
+        sql = "INSERT INTO {0} VALUES(%s, %s)".format(self.tbl)
+        self.cur.execute(sql, ('foo', 1))
+        self.cur.execute(sql, ('ëëë', 2))
+        self.cur.execute(sql, (u'ááá', 5))
+
+        self.cnx.set_charset_collation('binary')
+        self.cur.execute(sql, ('bar', 3))
+        self.cur.execute(sql, ('ëëë', 4))
+        self.cur.execute(sql, (u'ááá', 6))
+
+        exp = [
+            (bytearray(b'foo'), 1),
+            (bytearray(b'\xeb\xeb\xeb'), 2),
+            (bytearray(b'\xe1\xe1\xe1'), 5),
+            (bytearray(b'bar'), 3),
+            (bytearray(b'\xc3\xab\xc3\xab\xc3\xab'), 4),
+            (bytearray(b'\xc3\xa1\xc3\xa1\xc3\xa1'), 6)
+        ]
+
+        self.cur.execute("SELECT * FROM {0}".format(self.tbl))
+        self.assertEqual(exp, self.cur.fetchall())
