added test test_encoding_latin1_statement

Author: mastak

pymysqlreplication/tests/test_data_type.py

@@ -451,6 +451,36 @@ def test_encoding_utf8(self):
         event = self.create_and_insert_value(create_query, insert_query)
         self.assertMultiLineEqual(event.rows[0]["values"]["test"], string)
 
+    def test_encoding_latin1_statement(self):
+        binlog_format = self.execute("show variables like 'binlog_format'").fetchone()[1]
+        if binlog_format.upper() != "STATEMENT":
+            raise unittest.SkipTest("Test test_encoding_latin1_statement is skipped, binlog_format"
+                                    " is %s. It test require STATEMENT" % binlog_format)
+
+        db = copy.copy(self.database)
+        db["charset"] = "latin1"
+        self.connect_conn_control(db)
+
+        create_query = "CREATE TABLE test (test CHAR(12)) CHARACTER SET latin1 COLLATE latin1_bin;"
+        insert_query = b"INSERT INTO test VALUES('\x96')"
+
+        self.execute(create_query)
+        self.execute(insert_query)
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        # QueryEvent for the Create Table
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        # QueryEvent for the BEGIN
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, QueryEvent)
+
+        self.assertEqual(event.query, insert_query.decode("latin-1"))
+
 
 if __name__ == "__main__":
     unittest.main()