|
接到的测试任务为，对外联调的查询接口优化，由之前的查库改为查缓存数据，需要比对接口返回的字段值与数据库是否一致，涉及字段70个
脚本思路如下：
1、封装获取接口返参的方法
def get_text(res, key):
"""
获取接口指定字段的返回值
:param res: 请求结果
:param key: 需获取的字段名
:return: key的value
"""
if res is not None:
try:
txt = json.loads(res)
value = jsonpath.jsonpath(txt, '$..{}'.format(key))
# jsonpath获取成功会返回list，获取失败会返回False
if value:
if len(value) == 1:
return value[0]
return value
except Exception as e:
return e
else:
return None
2、封装获取数据库查询结果的方法
'''
数据库新增、查询封装
'''
import pymysql,traceback,sys
class MysqlUtil:
def __init__(self):
'''
初始化方法，链接数据库
'''
self.host = 'XXX' # 主机名
self.user = 'XXX' # 数据库用户名
self.password = 'XXX' # 数据库密码
self.port = 3306 # 数据库端口
self.charset = 'utf8'
def insert(self,database,sql):
'''
插入数据库
:param sql: 插入数据库的SQL语句
:param database： 数据库名称
:return:
'''
self.database = database
self.db = pymysql.connect(
host = self.host,user=self.user,password=self.password,db=self.database,
port=self.port,charset=self.charset) #建立链接
self.cursor = self.db.cursor(
cursor=pymysql.cursors.DictCursor) #设置游标，并将游标设置为字典类型
try:
#执行SQL语句
self.cursor.execute(sql)
#提交到数据库执行
self.db.commit()
except Exception: #方法一：捕获所有异常
#如果发生异常，则回滚
print('发生异常',Exception)
self.db.rollback()
finally:
#最终关闭数据库连接
self.db.close()
def fetchone(self,database,sql):
'''
查询数据库，单个结果集
fetchone（） 该方法获取下一个查询结果集，结果集是一个对象
'''
self.database = database #数据库名称
self.db = pymysql.connect(
host = self.host,user=self.user,password=self.password,db=self.database,
port=self.port,charset=self.charset) #建立链接
self.cursor = self.db.cursor(
cursor=pymysql.cursors.DictCursor) #设置游标，并将游标设置为字典类型
try:
#执行SQL语句
self.cursor.execute(sql)
resulist = self.cursor.fetchone() #获取查询的结果
return resulist
except Exception: #方法二： 采用traceback模块查看异常
#输出异常信息
traceback.print_exc()
#如果发生异常，则回滚
self.db.rollback()
finally:
#最终关闭数据库连接
self.db.close()
def fetchall(self,database,sql):
'''
查询数据库，多个结果集
fetchall()：接收全部的返回结果行
'''
self.database = database #数据库名称
self.db = pymysql.connect(
host = self.host,user=self.user,password=self.password,db=self.database,
port=self.port,charset=self.charset) #建立链接
self.cursor = self.db.cursor(
cursor=pymysql.cursors.DictCursor) #设置游标，并将游标设置为字典类型
try:
# 执行SQL语句
self.cursor.execute(sql)
resulit = self.cursor.fetchall()
return resulit #获取查询的结果
except Exception: #方法一：捕获所有异常
#如果发生异常，则回滚
print('发生异常',Exception)
self.db.rollback()
finally:
# 最终关闭数据库连接
self.db.close()
def delect(self,database,sql):
'''
删除结果集
:param database: 数据库名称
:param sql: sql语句
:return:
'''
self.database = database #数据库名称
self.db = pymysql.connect(
host = self.host,user=self.user,password=self.password,db=self.database,
port=self.port,charset=self.charset) #建立链接
self.cursor = self.db.cursor(
cursor=pymysql.cursors.DictCursor) #设置游标，并将游标设置为字典类型
# 执行SQL语句
try:
self.cursor.execute(sql)
self.db.commit()
except Exception: # 方法一：捕获所有异常
# 如果发生异常，则回滚
print('发生异常', Exception)
self.db.rollback()
finally:
# 最终关闭数据库连接
self.db.close()
def update(self,database,sql):
'''
更新结果集
'''
self.database = database #数据库名称
self.db = pymysql.connect(
host = self.host,user=self.user,password=self.password,db=self.database,
port=self.port,charset=self.charset) #建立链接
self.cursor = self.db.cursor(
cursor=pymysql.cursors.DictCursor) #设置游标，并将游标设置为字典类型
# 执行SQL语句
try:
self.cursor.execute(sql)
self.db.commit()
except Exception: # 方法一：捕获所有异常
# 如果发生异常，则回滚
print('发生异常', Exception)
self.db.rollback()
finally:
# 最终关闭数据库连接
self.db.close()
3、数据库返回结果数据类型和接口返回数据类型不一致，需将数据类型属性转换成一致才可比对
举例SQL查询结果为：sort_datebase，接口提取参数返回结果为：response
①获取数据类型属性方法：
for column in sort_datebase:
print(column,sort_datebase.get(column),type(sort_datebase.get(column)) )
②转换数据类型方法：
for column in sort_datebase:
if type(sort_datebase.get(column)) == datetime.datetime:
sort_datebase[column] = str(sort_datebase.get(column)) #将查询结果的数据类型转换为str
elif type(sort_datebase.get(column)) == decimal.Decimal:
sort_datebase[column] = float(sort_datebase.get(column))
elif type(sort_datebase.get(column)) == 'NoneType':
sort_datebase[column] = ' '
elif type(sort_datebase.get(column)) == int:
sort_datebase[column] = str(sort_datebase[column])
4、将key名转换一致，数据库字段名与接口返回的字段名不一致，需替换
def str2Hump(text):
"""
下划线转驼峰
"""
arr = filter(None, text.lower().split('_'))
res = ''
j = 0
for i in arr:
if j == 0:
res = i
else:
res = res + i[0].upper() + i[1:]
j += 1
return res
5、字典key排序(不一定需要排序，此处是为了直观查看print的结果)
def dict_order(dict_name): #字典排序
sort_dict = sorted(dict_name.items())
dict_value = {}
for dict in sort_dict:
key = dict[0]
value = dict[1]
dict_value[key] = value
return dict_value
6、字典元素比较方法
def compare(dict_file, dict_db):
for k, v in dict_file.items():
if v == None:
v == ''
elif v != dict_db[k]:
print(k,'接口返回为',v, '数据库为',dict_db[k])
|