python 接口自动化测试--代码实现（七）

python 接口自动化测试--代码实现（七） - 涛哥爱吃面

hito 2021-08-13 原文

python 接口自动化测试–代码实现（七）

数据库还原脚本：

#! /usr/bin/python
# coding:utf-8
import time
from Engine.SqlEngine import MSSQL
COUNT=1
def restoreRelease():
    global COUNT
    checkConSql = "select spid from sysprocesses where dbid in (select dbid from sysdatabases where name=\'SOATest\')"
    restoreSql = "RESTORE DATABASE SOATest FROM DATABASE_SNAPSHOT = \'SOATest_ss\'"
    dbc=MSSQL(\'192.168.1.2\',\'yht\',\'yht\',\'Master\')
    conNum = dbc.ExecQuery(checkConSql)
    if COUNT < 5:
        if len(conNum) == 0:
            print(u\'%d条连接数，可以还原数据库，还原中...\'%len(conNum))
            dbc.ExecNonQuery(restoreSql)
            print(u\'数据库还原完成\')
            return True
        else:
            print(u\'%d条连接数占用无法还原数据库，5秒后重试\'%len(conNum))
            time.sleep(5)
            COUNT=COUNT+1
            restoreRelease()
    else:
        print(u\'%d条连接数始终占用，已试过5次依然无法还原数据库\'%len(conNum))
        return False

SqlEngine.py

#! /usr/bin/python
# coding:utf-8
import pymssql
import sys
from Engine.DataEngine import decoCatchExcep
reload(sys)
sys.setdefaultencoding(\'utf-8\')
class MSSQL(object):
    """
    对pymssql的简单封装
    pymssql库，该库到这里下载：http://www.lfd.uci.edu/~gohlke/pythonlibs/#pymssql
    使用该库时，需要在Sql Server Configuration Manager里面将TCP/IP协议开启

    用法：

    """

    def __init__(self,host,user,pwd,db):
        self.host = host
        self.user = user
        self.pwd = pwd
        self.db = db

    def __GetConnect(self):
        """
        得到连接信息
        返回: conn.cursor()
        """
        if not self.db:
            raise(NameError,"没有设置数据库信息")
        self.conn = pymssql.connect(host=self.host,user=self.user,password=self.pwd,database=self.db,charset="utf8")
        cur = self.conn.cursor()
        if not cur:
            raise(NameError,"连接数据库失败")
        else:
            return cur
    @decoCatchExcep
    def ExecQuery(self,sql):

        cur = self.__GetConnect()
        cur.execute(sql)
        resList = cur.fetchall()
        self.conn.close()
        return resList
    @decoCatchExcep
    def ExecNonQuery(self,sql):
        cur = self.__GetConnect()
        self.conn.autocommit(True)
        cur.execute(sql)
        self.conn.autocommit(False)
        self.conn.commit()
        self.conn.close()

装饰器decoCatchExcep：

def decoCatchExcep(func):
    def _decoCatchExcep(*args, **kwargs):
        try:
            ret = func(*args, **kwargs)
            return ret
        except Exception,e:
            print(func.__name__+\' : \'+str(e).encode(\'gb18030\'))
            LogPro.writeException(str(e).encode(\'gb18030\'))
    return _decoCatchExcep