feat: add version8 test in test_data_type.py

related PR: julien-duponchelle#355

pymysqlreplication/tests/base.py

@@ -161,9 +161,10 @@ def bin_log_basename(self):
         bin_log_basename = bin_log_basename.split("/")[-1]
         return bin_log_basename
 
+
 class PyMySQLReplicationPercona8TestCase(PyMySQLReplicationTestCase):
     def setUp(self):
-        super().setUp()        
+        super().setUp()
         # default
         self.database = {
             "host": os.environ.get("MYSQL_8_0") or "localhost",

pymysqlreplication/tests/test_data_type.py

@@ -19,7 +19,7 @@
 from pymysqlreplication._compat import text_type
 
 
-__all__ = ["TestDataType"]
+__all__ = ["TestDataType", "TestDataTypeVersion8"]
 
 
 def to_binary_dict(d):
@@ -894,5 +894,23 @@ def test_mariadb_only_status_vars(self):
         self.assertEqual(event.query, create_query)
 
 
+class TestDataTypeVersion8(base.PyMySQLReplicationPercona8TestCase):
+    def test_partition_id(self):
+        if not self.isMySQL80AndMore():
+            self.skipTest("Not supported in this version of MySQL")
+        create_query = "CREATE TABLE test (id INTEGER) \
+            PARTITION BY RANGE (id) ( \
+                PARTITION p0 VALUES LESS THAN (1),   \
+                PARTITION p1 VALUES LESS THAN (2),   \
+                PARTITION p2 VALUES LESS THAN (3),   \
+                PARTITION p3 VALUES LESS THAN (4),   \
+                PARTITION p4 VALUES LESS THAN (5)    \
+            )"
+        insert_query = "INSERT INTO test (id) VALUES(3)"
+        event = self.create_and_insert_value(create_query, insert_query)
+        self.assertEqual(event.extra_data_type, 1)
+        self.assertEqual(event.partition_id, 3)
+
+
 if __name__ == "__main__":
     unittest.main()