Add unit tests for 'MariaDbGtidListEvent' replication class

Author: [email protected]

pymysqlreplication/tests/test_basic.py

@@ -17,7 +17,7 @@
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
 
-__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings", "TestGtidBinLogStreamReader"]
+__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings", "TestGtidBinLogStreamReader","TestMariadbBinlogStreaReader"]
 
 
 class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
@@ -1002,6 +1002,25 @@ def test_parsing(self):
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac:1-:1")
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac::1")
 
+class TestMariadbBinlogStreaReader(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestMariadbBinlogStreaReader,self).setUp()
+        if not self.isMariaDB():
+            raise unittest.SkipTest("Skipping test: Not a MariaDB instance")
+        
+    def test_gtid_list_event(self):
+        event = self.stream.fetchone()
+        self.assertEqual(event.position, 4)  
+        
+        #FormatDescriptionEvent
+        event = self.stream.fetchone()
+        self.assertEqual(event.event_type,15)
+        self.assertIsInstance(event,FormatDescriptionEvent)
+
+        #MariadbAnnotateRowsEvent
+        event = self.stream.fetchone()
+        self.assertEqual(event.event_type,163)
+        self.assertIsInstance(event,MariadbGtidListEvent)
 
 if __name__ == "__main__":
     import unittest