From 9ba26bca6bafea3f0d04d9f9cbe4d8078df9a716 Mon Sep 17 00:00:00 2001 From: Lubimov Igor Date: Wed, 13 Jul 2016 10:28:22 +0300 Subject: [PATCH] added test test_encoding_latin1_statement --- pymysqlreplication/tests/test_data_type.py | 30 ++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/pymysqlreplication/tests/test_data_type.py b/pymysqlreplication/tests/test_data_type.py index 5ed48732..0c04a963 100644 --- a/pymysqlreplication/tests/test_data_type.py +++ b/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()