Refactor test method in TestMariadbBinlogStreamReader class

Author: [email protected]

pymysqlreplication/tests/base.py

@@ -121,3 +121,32 @@ def bin_log_basename(self):
         bin_log_basename = cursor.fetchone()[0]
         bin_log_basename = bin_log_basename.split("/")[-1]
         return bin_log_basename
+
+
+class PyMySQLReplicationMariaDbTestCase(PyMySQLReplicationTestCase):
+    def setUp(self):
+        # default
+        self.database = {
+            "host": "localhost",
+            "user": "root",
+            "passwd": "",
+            "port": 3308,
+            "use_unicode": True,
+            "charset": "utf8",
+            "db": "pymysqlreplication_test"
+        }
+
+        self.conn_control = None
+        db = copy.copy(self.database)
+        db["db"] = None
+        self.connect_conn_control(db)
+        self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
+        self.execute("CREATE DATABASE pymysqlreplication_test")
+        db = copy.copy(self.database)
+        self.connect_conn_control(db)
+        self.stream = None
+        self.resetBinLog()
+
+        
+    
+    

pymysqlreplication/tests/test_basic.py

@@ -1002,19 +1002,7 @@ def test_parsing(self):
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac:1-:1")
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac::1")
 
-class TestMariadbBinlogStreaReader(base.PyMySQLReplicationTestCase):
-    def setUp(self):
-        super(TestMariadbBinlogStreaReader,self).setUp()
-        self.stream.close()
-        maria_db = copy.copy(self.database)
-        maria_db["db"] = None
-        maria_db["port"] = 3308
-        self.connect_conn_control(maria_db)
-        self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
-        self.execute("CREATE DATABASE pymysqlreplication_test")
-        self.isMariaDB()
-        self.stream = BinLogStreamReader(maria_db, server_id=1024,
-                                         ignored_events=self.ignoredEvents())
+class TestMariadbBinlogStreaReader(base.PyMySQLReplicationMariaDbTestCase):
 
     def test_gtid_list_event(self):
         event = self.stream.fetchone()