Support test

Author: julien-duponchelle

pymysqlreplication/binlogstream.py

@@ -13,7 +13,15 @@ def __init__(self, connection, resume_stream = False, blocking = False):
         blocking: Read on stream is blocking
         '''
         self.__connection = connection
+        self.__connected = False
+        self.__resume_stream = resume_stream
+        self.__blocking = blocking
+        
+        #Store table meta informations
+        self.table_map = {}
+
 
+    def __connect_to_stream(self):
         cur = self.__connection.cursor()
         cur.execute("SHOW MASTER STATUS")
         (log_file, log_pos) = cur.fetchone()[:2]
@@ -27,22 +35,22 @@ def __init__(self, connection, resume_stream = False, blocking = False):
         command = COM_BINLOG_DUMP
         prelude = struct.pack('<i', len(log_file) + 11) \
                 + int2byte(command)
-        if resume_stream:
+        if self.__resume_stream:
             prelude += struct.pack('<I', log_pos)            
         else:
             prelude += struct.pack('<I', 4)
-        if blocking:
+        if self.__blocking:
             prelude += struct.pack('<h', 0)
         else:
             prelude += struct.pack('<h', 1)        
         prelude += struct.pack('<I', 3)
         self.__connection.wfile.write(prelude + log_file)
         self.__connection.wfile.flush()
-
-        #Store table meta informations
-        self.table_map = {}
+        self.__connected = True
 
     def fetchone(self):
+        if self.__connected == False:
+            self.__connect_to_stream()
         pkt = self.__connection.read_packet()
         if not pkt.is_ok_packet():
             return None

pymysqlreplication/tests/base.py

@@ -1,15 +1,21 @@
 import pymysql
 import unittest
+import copy
+from pymysqlreplication import BinLogStreamReader
 
 class PyMySQLReplicationTestCase(unittest.TestCase):
     '''Test the module. Be carefull it will reset your MySQL server'''
     database =  {"host":"localhost","user":"root", "passwd":"","use_unicode": True}
 
     def setUp(self):
         self.conn_control = pymysql.connect(**self.database)
-        self.conn_test = pymysql.connect(**self.database)
         self.execute("CREATE DATABASE pymysqlreplication_test")
         self.resetBinLog()
+        db = copy.copy(self.database)
+        db["db"] = "pymysqlreplication_test"
+        self.conn_control = pymysql.connect(**db)
+        self.conn_test = pymysql.connect(**self.database)
+        self.stream = BinLogStreamReader(self.conn_test)     
 
     def tearDown(self):
         self.execute("DROP DATABASE pymysqlreplication_test")

pymysqlreplication/tests/test_basic.py

@@ -1,11 +1,19 @@
 import base
-from pymysqlreplication import BinLogStreamReader
+from pymysqlreplication.event import *
 
 class TestBinLogStreamReader(base.PyMySQLReplicationTestCase):
-    def test_open_stream(self):
-        """ test opening stream"""
-        stream = BinLogStreamReader(self.conn_test)
+    def test_read_query_event(self):
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
 
+        #RotateEvent
+        self.stream.fetchone()
+        #FormatDescription
+        self.stream.fetchone()
+
+        event = self.stream.fetchone()
+        self.assertIsInstance(event.event, BinLogQueryEvent)
+        self.assertEqual(event.event.query, query)
 
 __all__ = ["TestBinLogStreamReader"]