在我们的接口自动化框架中,链接数据库是很重要的一个部分,我们可以从数据库中查一些数据作为用例中的预期结果,也可以在自动化执行结束的时候清理数据
设计思路:
类中分三部分
初始化:链接数据库
执行sql 有commit操作
获取数据库数据,select语句
其中加入日志,数据库的参数从read_config里面的config中获取
详细如下:
import pymysql from DBUtils.PooledDB import PooledDB from common.logger import Log from common.read_config import config from time import sleep class ConnectMysql(object): def __init__(self): parm = config.read_config_parm('mysql') self.logger = Log('连接mysql数据库').get_logger() try: pool = PooledDB(pymysql, 1, host=parm.get('host'), user=parm.get('user'), passwd=parm.get('password'), db=parm.get('database'), port=parm.get('port')) # 1为连接池里的最少连接数 self.conn = pool.connection() self.logger.info('mysql数据库连接成功') except Exception as e: self.logger.error('mysql数据库连接失败:{}'.format(e)) def execute_sql(self, sql): """ 执行sql语句 :param sql: 需要执行的sql语句 :return: """ cur = self.conn.cursor() # 获取游标 try: cur.execute(sql) self.conn.commit() self.logger.info('sql执行成功:{}'.format(sql)) cur.close() except Exception as e: self.logger.error('sql执行失败: %s' % e) self.conn.rollback() # sql执行失败进行回滚 def get_string(self, sql): """ 通过sql获取某个字段的值 :param sql: 需要执行的sql语句 :return: """ sleep(3) cur = self.conn.cursor() cur.execute(sql) results = cur.fetchall() cur.close() self.logger.info('数据库获取结果为:{}'.format(results)) if results is not None: for result in results: for data in result: return data
这样我们后面调用数据库时就只要直接调用方法,传sql语句就行,为了方便调用,我们可以在后面加上mysql = ConnectMysql()
这样只需要mysql.方法就ok了
如果系统有用到Redis,而我们自动化执行时需要从Redis中获取数据
也可以如下写:
class ConnectRedis(object): def __init__(self): parm = config.read_config_parm('redis') logger = Log('连接redis数据库').get_logger() try: self.pool = redis.ConnectionPool(host=parm['host'], port=parm['port'], auth=parm['auth']) self.redis = redis.StrictRedis(connection_pool=self.pool) logger.info('redis数据库连接成功') except Exception as e: logger.error('redis数据库连接失败: %s' % e) def get_one(self, key): """ 获取redis的一个值 :param key: :return: """ result = self.redis.get(key) return result def get_two(self, name, key): result = self.redis.hget(name, key) return result