feat: add categorize test when invalid mode

Author: mjs1995

pymysqlreplication/tests/test_basic.py

@@ -591,6 +591,35 @@ def test_categorize_none(self):
         self.assertIsInstance(write_rows_event, WriteRowsEvent)
         self.assertEqual(write_rows_event.rows[0]['category_of_none']['col1'], 'null')
 
+    def test_categorize_none_invalid(self):
+        self.execute("SET SESSION SQL_MODE='ALLOW_INVALID_DATES'")
+        self.execute("CREATE TABLE test_table (col0 INT, col1 VARCHAR(10), col2 DATETIME, col3 DATE, col4 SET('a', 'b', 'c'))")
+        self.execute("INSERT INTO test_table VALUES (NULL, NULL, '0000-00-00 00:00:00', NULL, NULL)")
+        self.resetBinLog()
+        self.execute("UPDATE test_table SET col1 = NULL, col2 = NULL, col3='0000-00-00',col4 = 'd' WHERE col0 IS NULL")
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
+
+        event = self.stream.fetchone()
+        if self.isMySQL56AndMore():
+            self.assertEqual(event.event_type, UPDATE_ROWS_EVENT_V2)
+        else:
+            self.assertEqual(event.event_type, UPDATE_ROWS_EVENT_V1)
+        self.assertIsInstance(event, UpdateRowsEvent)
+        self.assertEqual(event.rows[0]["before_category_of_none"]["col0"], 'null')
+        self.assertEqual(event.rows[0]["before_category_of_none"]["col1"], 'null')
+        self.assertEqual(event.rows[0]["before_category_of_none"]["col2"], 'out of datetime2 range')
+        self.assertEqual(event.rows[0]["before_category_of_none"]["col3"], 'null')
+        self.assertEqual(event.rows[0]["before_category_of_none"]["col4"], 'null')
+        self.assertEqual(event.rows[0]["after_category_of_none"]["col0"], 'null')
+        self.assertEqual(event.rows[0]["after_category_of_none"]["col1"], 'null')
+        self.assertEqual(event.rows[0]["after_category_of_none"]["col2"], 'null')
+        self.assertEqual(event.rows[0]["after_category_of_none"]["col3"], 'out of date range')
+        self.assertEqual(event.rows[0]["after_category_of_none"]["col4"], 'empty set')
 
 
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):