Start implementing test

Author: julien-duponchelle

pymysqlreplication/tests/__init__.py

@@ -0,0 +1,6 @@
+from pymysqlreplication.tests.test_basic import *
+
+if __name__ == "__main__":
+    import unittest
+    unittest.main()
+

pymysqlreplication/tests/base.py

@@ -0,0 +1,24 @@
+import pymysql
+import unittest
+
+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()
+
+    def tearDown(self):
+        self.execute("DROP DATABASE pymysqlreplication_test")
+        self.conn_control.close()
+        self.conn_test.close()
+
+    def execute(self, query):
+        c = self.conn_control.cursor()
+        c.execute(query)
+
+    def resetBinLog(self):
+        self.execute("RESET MASTER")

pymysqlreplication/tests/test_basic.py

@@ -0,0 +1,14 @@
+import base
+from pymysqlreplication import BinLogStreamReader
+
+class TestBinLogStreamReader(base.PyMySQLReplicationTestCase):
+    def test_open_stream(self):
+        """ test opening stream"""
+        stream = BinLogStreamReader(self.conn_test)
+
+
+__all__ = ["TestBinLogStreamReader"]
+
+if __name__ == "__main__":
+    import unittest
+    unittest.main()