Merge pull request #314 from methane/fix/bytes

Stricter test for blob

pymysql/connections.py

@@ -760,7 +760,10 @@ class Connection(object):
         #if DEBUG:
         #    print("DEBUG: sending query:", sql)
         if isinstance(sql, text_type) and not (JYTHON or IRONPYTHON):
-            sql = sql.encode(self.encoding)
+            if PY2:
+                sql = sql.encode(self.encoding)
+            else:
+                sql = sql.encode(self.encoding, 'surrogateescape')
         self._execute_command(COMMAND.COM_QUERY, sql)
         self._affected_rows = self._read_query_result(unbuffered=unbuffered)
         return self._affected_rows

pymysql/tests/test_basic.py

@@ -86,19 +86,20 @@ class TestConversion(base.PyMySQLTestCase):
         finally:
             c.execute("drop table test_dict")
 
-
+    def test_blob(self):
-    def test_big_blob(self):
+        """test binary data"""
-        """ test tons of data """
+        data = bytes(bytearray(range(256)) * 4)
         conn = self.connections[0]
+        self.safe_create_table(
+            conn, "test_blob", "create table test_blob (b blob)")
+
         c = conn.cursor()
-        c.execute("create table test_big_blob (b blob)")
         try:
-            data = "pymysql" * 1024
+            c.execute("insert into test_blob (b) values (%s)", (data,))
-            c.execute("insert into test_big_blob (b) values (%s)", (data,))
+            c.execute("select b from test_blob")
-            c.execute("select b from test_big_blob")
+            self.assertEqual(data, c.fetchone()[0])
-            self.assertEqual(data.encode(conn.charset), c.fetchone()[0])
         finally:
-            c.execute("drop table test_big_blob")
+            c.close()
 
     def test_untyped(self):
         """ test conversion of null, empty string """