add: type hints for PyMyReplTestCase

Author: soulee-dev

pymysqlreplication/tests/base.py

@@ -5,6 +5,8 @@
 import sys
 from pymysql.cursors import Cursor
 import pymysql
+from pymysql import Connection
+from typing import Optional
 
 from pymysqlreplication import BinLogStreamReader
 
@@ -22,7 +24,7 @@ def ignoredEvents(self) -> list:
 
     def setUp(self) -> None:
         # default
-        self.database = {
+        self.database: dict = {
             "host": os.environ.get("MYSQL_5_7") or "localhost",
             "user": "root",
             "passwd": "",
@@ -32,18 +34,18 @@ def setUp(self) -> None:
             "db": "pymysqlreplication_test",
         }
 
-        self.conn_control = None
+        self.conn_control: Optional[Connection] = 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.stream: Optional[BinLogStreamReader] = None
         self.resetBinLog()
         self.isMySQL56AndMore()
-        self.__is_mariaDB = None
+        self.__is_mariaDB: Optional[bool] = None
 
     def getMySQLVersion(self) -> str:
         """Return the MySQL version of the server
@@ -67,7 +69,7 @@ def isMySQL80AndMore(self) -> bool:
 
     def isMariaDB(self) -> bool:
         if self.__is_mariaDB is None:
-            self.__is_mariaDB = (
+            self.__is_mariaDB: bool = (
                 "MariaDB" in self.execute("SELECT VERSION()").fetchone()[0]
             )
         return self.__is_mariaDB
@@ -81,13 +83,13 @@ def supportsGTID(self) -> bool:
     def connect_conn_control(self, db) -> None:
         if self.conn_control is not None:
             self.conn_control.close()
-        self.conn_control = pymysql.connect(**db)
+        self.conn_control: Connection = pymysql.connect(**db)
 
     def tearDown(self) -> None:
         self.conn_control.close()
-        self.conn_control = None
+        self.conn_control: Optional[Connection] = None
         self.stream.close()
-        self.stream = None
+        self.stream: Optional[BinLogStreamReader] = None
 
     def execute(self, query: str) -> Cursor:
         c = self.conn_control.cursor()
@@ -103,7 +105,7 @@ def resetBinLog(self) -> None:
         self.execute("RESET MASTER")
         if self.stream is not None:
             self.stream.close()
-        self.stream = BinLogStreamReader(
+        self.stream: BinLogStreamReader = BinLogStreamReader(
             self.database, server_id=1024, ignored_events=self.ignoredEvents()
         )
 
@@ -139,15 +141,15 @@ def setUp(self) -> None:
             "db": "pymysqlreplication_test",
         }
 
-        self.conn_control = None
+        self.conn_control: Optional[Connection] = 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.stream: Optional[BinLogStreamReader] = None
         self.resetBinLog()
 
     def bin_log_basename(self) -> str: