add log_pos tests

Author: lxyu

pymysqlreplication/tests/test_basic.py

@@ -30,17 +30,20 @@ def test_reading_rotate_event(self):
         query = "CREATE TABLE test_2 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)
 
-        rotate_event = self.stream.fetchone()
+        # Rotate event
+        self.stream.fetchone()
         self.stream.close()
 
         query = "CREATE TABLE test_3 (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)
 
-        rotate_event = self.stream.fetchone()
+        # Rotate event
+        self.stream.fetchone()
 
     def test_connection_stream_lost_event(self):
         self.stream.close()
-        self.stream = BinLogStreamReader(connection_settings = self.database, blocking = True)
+        self.stream = BinLogStreamReader(connection_settings=self.database,
+                                         blocking=True)
 
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)
@@ -66,7 +69,8 @@ def test_connection_stream_lost_event(self):
 
     def test_filtering_events(self):
         self.stream.close()
-        self.stream = BinLogStreamReader(connection_settings = self.database, only_events = [QueryEvent])
+        self.stream = BinLogStreamReader(connection_settings=self.database,
+                                         only_events=[QueryEvent])
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
         self.execute(query)
 
@@ -172,6 +176,48 @@ def test_update_row_event(self):
         self.assertEqual(event.rows[0]["after_values"]["id"], 1)
         self.assertEqual(event.rows[0]["after_values"]["data"], "World")
 
+    def test_log_pos(self):
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "INSERT INTO test (data) VALUES('Hello')"
+        self.execute(query)
+        self.execute("COMMIT")
+
+        for i in range(6):
+            self.stream.fetchone()
+        # record position after insert
+        log_file, log_pos = self.stream.log_file, self.stream.log_pos
+
+        query = "UPDATE test SET data = 'World' WHERE id = 1"
+        self.execute(query)
+        self.execute("COMMIT")
+
+        # resume stream from previous position
+        if self.stream is not None:
+            self.stream.close()
+        self.stream = BinLogStreamReader(
+            connection_settings=self.database,
+            resume_stream=True,
+            log_file=log_file,
+            log_pos=log_pos
+        )
+
+        # RotateEvent
+        self.stream.fetchone()
+        # FormatDescription
+        self.stream.fetchone()
+        # XvidEvent
+        self.stream.fetchone()
+        # QueryEvent for the BEGIN
+        self.stream.fetchone()
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, TableMapEvent)
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event, UpdateRowsEvent)
+
+
 class TestMultipleRowBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def test_insert_multiple_row_event(self):
         query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
@@ -234,7 +280,7 @@ def test_update_multiple_row_event(self):
 
         event = self.stream.fetchone()
         if self.isMySQL56AndMore():
-           self.assertEqual(event.event_type, UPDATE_ROWS_EVENT_V2)
+            self.assertEqual(event.event_type, UPDATE_ROWS_EVENT_V2)
         else:
             self.assertEqual(event.event_type, UPDATE_ROWS_EVENT_V1)
         self.assertIsInstance(event, UpdateRowsEvent)