Add test for rows_query_log_event

Author: soulee-dev

docker-compose.yml

@@ -15,3 +15,11 @@ services:
     ports:
       - 3307:3307
     command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+
+  percona-5.7-binlog-rows-query-events:
+    image: percona:5.7
+    environment:
+        MYSQL_ALLOW_EMPTY_PASSWORD: true
+    ports:
+        - 3308:3308
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --binlog-rows-query-log-events=on --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3308

pymysqlreplication/tests/base.py

@@ -18,15 +18,15 @@ class PyMySQLReplicationTestCase(base):
     def ignoredEvents(self):
         return []
 
-    def setUp(self):
+    def setUp(self, port=3306):
 
         db = os.environ.get('DB')
         # default
         self.database = {
             "host": "localhost",
             "user": "root",
             "passwd": "",
-            "port": 3306,
+            "port": port,
             "use_unicode": True,
             "charset": "utf8",
             "db": "pymysqlreplication_test"

pymysqlreplication/tests/test_basic.py

@@ -1003,6 +1003,24 @@ def test_parsing(self):
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac::1")
 
 
+class TestRowsQueryLogEvents(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestRowsQueryLogEvents, self).setUp(port=3308)
+
+    def test_rows_query_log_event(self):
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=[RowsQueryLogEvent],
+        )
+        self.execute("CREATE TABLE IF NOT EXISTS test (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255))")
+        self.execute("INSERT INTO test (name) VALUES ('Soul Lee')")
+        self.execute("COMMIT")
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, RowsQueryLogEvent)
+
+
 if __name__ == "__main__":
     import unittest
     unittest.main()