用python做自动化测试--数据库的自动化测试（2）

          上一篇中我们分析了数据库测试的几种应用类型，不知道其他公司还有什么数据库应用测试类型，可以分享出来，大家一起探讨下测试方案。具体的案列代码我就不在这里贴出来了，这个和具体业务高度相关。这里贴两个mysql, vertical数据库的连接和SQL 操作的类。大家以后应用可以直接用了。

MySQL 数据库连接与SQL 操作

import MySQLdb,time,traceback,logging

class DBOperateAction:
    def __init__(self,dbhost,dbaccount,dbpasswd,dbname,port=3306,charset="utf8"):
        self.dbhost=dbhost
        self.dbaccount=dbaccount
        self.dbpasswd=dbpasswd
        self.dbname=dbname
        self.charset=charset
        self.port=port
        self.db_conn=""
        self.db_cursor=""
    
    def connect(self):
        try:
            self.db_conn=MySQLdb.connect(host=self.dbhost,user=self.dbaccount,passwd=self.dbpasswd,
                                         db=self.dbname,port=self.port,charset=self.charset)
            self.db_cursor=self.db_conn.cursor()
            return True
        except MySQLdb.OperationalError:
            logging.error("Connect to "+self.dbhost+" Failed")
            logging.exception("exception message:")
            return False

    def re_connect(self):
        logging.error("connect to mysql server failed, reconnect")
        try:
            self.db_conn=MySQLdb.connect(host=self.dbhost,user=self.dbaccount,passwd=self.dbpasswd,
                                         db=self.dbname,port=self.port,charset="utf8")
            self.db_cursor=self.db_conn.cursor()
            return True
        except MySQLdb.OperationalError:
            logging.error("Reconnect MySQL failed")
            return False

        
    def get_all_result(self,sql):
        try:
            logging.info("Execute sql: "+sql)
            self.db_cursor.execute(sql)
            self.db_conn.commit()
            result=self.db_cursor.fetchall()
            return result
        except MySQLdb.OperationalError:
            logging.exception("Execure SQL except"+sql)
            for i in range(0,3):
                time.sleep(5)
                if self.re_connect():
                    logging.info("reconnect database successfully")
                    return False
    
    def get_one_result(self,sql):
        try:
           logging.info("Execute sql: "+sql)
           self.db_cursor.execute(sql)
           self.db_conn.commit()
           result=self.db_cursor.fetchone()
           return result
        except MySQLdb.OperationalError:
            traceback.print_exc()
            for i in range(0,3):
                time.sleep(5)
                if self.re_connect():
                    break

    def close_connection(self):
        self.db_conn.close()

Vertica 数据库连接与SQL 操作. 需要注意的是, Vertical里面schema 和database 不个东西。在mysql里面schema和database 是一回事。

import vertica_python

class VerticaDBOperateAction:

    def __init__(self,dbhost,dbaccount,dbpasswd,dbname,port=5433):
        #self._conn_info=conn_info
        self.connection=""
        self.cursorcur =""
        self.dbhost=dbhost
        self.dbaccount=dbaccount
        self.dbpasswd=dbpasswd
        self.dbname=dbname
        self.port=port

    def connect(self):
        try:
            self.connection = vertica_python.connect(host=self.dbhost,user=self.dbaccount,password=self.dbpasswd,
                                         database=self.dbname,port=self.port)
            self.cursorcur=self.connection.cursor()
            return True
        except:
            logging.exception("Connect vertica database error with address:"+self.dbhost)
            logging.exception("exception message:")
            return False

    def get_all_result(self,sql):
        '''
        :return: # [ [1, 'something'], [2, 'something_else'] ]
        '''
        self.cursorcur.execute(sql)
        return self.cursorcur.fetchall()

    def close_connection(self):
        self.connection.close()