Tests gtid

Arthur Gautier · Arthur Gautier · commit a805ccd490b1 · 2014-09-01T11:16:42.000+02:00
diff --git a/pymysqlreplication/tests/test_basic.py b/pymysqlreplication/tests/test_basic.py
@@ -6,7 +6,7 @@
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
 
-__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader"]
+__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestGtidBinLogStreamReader"]
 
 
 class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
@@ -355,6 +355,83 @@ def test_delete_multiple_row_event(self):
         self.assertEqual(event.rows[1]["values"]["data"], "World")
 
 
+class TestGtidBinLogStreamReader(base.PyMySQLReplicationTestCase):
+    def test_read_query_event(self):
+        query = "CREATE TABLE test (id INT NOT NULL, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "SELECT @@global.gtid_executed;"
+        gtid = self.execute(query).fetchone()[0]
+
+        self.stream.close()
+        self.stream = BinLogStreamReader(connection_settings=self.database,
+                                         blocking=True,
+                                         auto_position=gtid)
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+
+        # Insert first event
+        query = "BEGIN;"
+        self.execute(query)
+        query = "INSERT INTO test (id, data) VALUES(1, 'Hello');"
+        self.execute(query)
+        query = "COMMIT;"
+        self.execute(query)
+
+        firstevent = self.stream.fetchone()
+        self.assertIsInstance(firstevent, GtidEvent)
+
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
+        self.assertIsInstance(self.stream.fetchone(), WriteRowsEvent)
+        self.assertIsInstance(self.stream.fetchone(), XidEvent)
+
+        # Insert second event
+        query = "BEGIN;"
+        self.execute(query)
+        query = "INSERT INTO test (id, data) VALUES(2, 'Hello');"
+        self.execute(query)
+        query = "COMMIT;"
+        self.execute(query)
+
+        secondevent = self.stream.fetchone()
+        self.assertIsInstance(secondevent, GtidEvent)
+
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), TableMapEvent)
+        self.assertIsInstance(self.stream.fetchone(), WriteRowsEvent)
+        self.assertIsInstance(self.stream.fetchone(), XidEvent)
+
+        self.assertEqual(secondevent.gno, firstevent.gno + 1)
+
+    def test_position_gtid(self):
+        query = "CREATE TABLE test (id INT NOT NULL, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "BEGIN;"
+        self.execute(query)
+        query = "INSERT INTO test (id, data) VALUES(1, 'Hello');"
+        self.execute(query)
+        query = "COMMIT;"
+        self.execute(query)
+
+        query = "CREATE TABLE test2 (id INT NOT NULL, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "SELECT @@global.gtid_executed;"
+        gtid = self.execute(query).fetchone()[0]
+
+        self.stream.close()
+        self.stream = BinLogStreamReader(connection_settings=self.database,
+                                         blocking=True,
+                                         auto_position=gtid)
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+        self.assertIsInstance(self.stream.fetchone(), GtidEvent)
+        event = self.stream.fetchone()
+
+        self.assertEqual(event.query, 'CREATE TABLE test2 (id INT NOT NULL, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))');
+
+
 if __name__ == "__main__":
     import unittest
     unittest.main()
