199 lines
8.2 KiB
Python
199 lines
8.2 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
# @Time : 2022/3/28 14:18
|
||
# @Author : 郭林莉
|
||
|
||
|
||
import jsonpath
|
||
from tools.allure_control import SqlSwitch
|
||
from tools.log_control import ERROR, WARNING
|
||
|
||
|
||
class Assert:
|
||
|
||
def __init__(self, assert_data):
|
||
self.assert_data = assert_data
|
||
|
||
@staticmethod
|
||
def _check_params(response_data, sql_data):
|
||
"""
|
||
|
||
:param response_data: 响应数据
|
||
:param sql_data: 数据库数据
|
||
:return:
|
||
"""
|
||
# 用例如果不执行，接口返回的相应数据和数据库断言都是 False，这里则判断跳过断言判断
|
||
if response_data is False and sql_data is False:
|
||
return False
|
||
else:
|
||
# 判断断言的数据类型
|
||
if isinstance(response_data, dict) and isinstance(sql_data, dict):
|
||
pass
|
||
elif isinstance(response_data, str):
|
||
pass
|
||
else:
|
||
# pass
|
||
raise ValueError("response_data、sql_data、assert_data的数据类型必须要是字典或者str类型")
|
||
|
||
@staticmethod
|
||
def _assert_type(key: any, types: str, value: any):
|
||
# 是否相等
|
||
if str(types) == "==":
|
||
assert key == value
|
||
# 判断实际结果小于预期结果
|
||
elif str(types) == "lt":
|
||
assert key < value
|
||
# 判断实际结果小于等于预期结果
|
||
elif str(types) == "le":
|
||
assert key < value
|
||
# 判断实际结果大于预期结果
|
||
elif str(types) == "gt":
|
||
assert key > value
|
||
# 判断实际结果大于等于预期结果
|
||
elif str(types) == "ge":
|
||
assert key >= value
|
||
# 判断实际结果不等于预期结果
|
||
elif str(types) == "not_eq":
|
||
assert key != value
|
||
# 判断字符串是否相等
|
||
elif str(types) == "str_eq":
|
||
assert key == value
|
||
# 判断长度是否相等
|
||
elif str(types) == "len_eq":
|
||
# assertion isinstance(value, int)
|
||
assert len(key) == value
|
||
# 判断长度大于
|
||
elif str(types) == "len_gt":
|
||
assert isinstance(value, int)
|
||
assert len(str(key)) > value
|
||
# 判断长度大于等于
|
||
elif str(types) == 'len_ge':
|
||
assert isinstance(value, int)
|
||
assert len(key) >= value
|
||
elif str(types) == "len_lt":
|
||
assert isinstance(value, int)
|
||
assert len(key) < value
|
||
# 判断长度小于等于
|
||
elif str(types) == 'len_le':
|
||
assert isinstance(value, int)
|
||
assert len(key) <= value
|
||
# 判断期望结果内容包含在实际结果中
|
||
elif str(types) == "contains":
|
||
assert str(value) in str(key)
|
||
# 判断实际结果包含在期望结果中
|
||
elif str(types) == 'contained_by':
|
||
assert str(value) in str(key)
|
||
# 检查响应内容的开头是否和预期结果内容的开头相等
|
||
elif str(types) == 'startswith':
|
||
assert str(value).startswith(str(key))
|
||
# 检查响应内容的结尾是否和预期结果内容相等
|
||
elif str(types) == 'endswith':
|
||
assert str(key).endswith(str(value))
|
||
elif str(types) == 'listdata':
|
||
assert key
|
||
elif str(types) == 'null_list':
|
||
assert not key
|
||
elif str(types) == 'listdata_len':
|
||
# print(len(key))
|
||
assert len(key) >= value
|
||
elif str(types) == 'status_code':
|
||
print("KEY是", key)
|
||
print("value", value)
|
||
# assert (key['status_code'], 200)
|
||
elif str(types) == 'list_in':
|
||
# print(len(key))
|
||
|
||
if key in value:
|
||
assert True
|
||
else:
|
||
assert False
|
||
else:
|
||
raise ValueError(f"断言失败，目前不支持{types}断言类型，如需新增断言类型，请联系管理员")
|
||
|
||
def sql_switch_handle(self, sql_data, assert_value, key, values, resp_data) -> None:
|
||
"""
|
||
|
||
:param sql_data: 测试用例中的sql
|
||
:param assert_value: 断言内容
|
||
:param key:
|
||
:param values:
|
||
:param resp_data: 预期结果
|
||
:return:
|
||
"""
|
||
# 判断数据库为开关为关闭状态
|
||
if SqlSwitch() is False:
|
||
WARNING.logger.warning(f"检测到数据库状态为关闭状态，程序已为您跳过此断言，断言值:{values}")
|
||
# 数据库开关为开启
|
||
if SqlSwitch():
|
||
# 判断当用例走的数据数据库断言，但是用例中未填写SQL
|
||
if sql_data == {'sql': None}:
|
||
raise ValueError("请在用例中添加您要查询的SQL语句。")
|
||
# 走正常SQL断言逻辑
|
||
else:
|
||
res_sql_data = jsonpath.jsonpath(sql_data, assert_value)[0]
|
||
# 判断mysql查询出来的数据类型如果是bytes类型，转换成str类型
|
||
if isinstance(res_sql_data, bytes):
|
||
res_sql_data = res_sql_data.decode('utf=8')
|
||
self._assert_type(types=self.assert_data[key]['type'], key=resp_data[0], value=res_sql_data)
|
||
|
||
def assert_type_handle(self, assert_type, sql_data, assert_value, key, values, resp_data) -> None:
|
||
# 判断断言类型
|
||
if assert_type == 'SQL':
|
||
self.sql_switch_handle(sql_data, assert_value, key, values, resp_data)
|
||
# 判断assertType为空的情况下，则走响应断言
|
||
elif assert_type == 'str':
|
||
print('现在是str类型的断言')
|
||
print("预期值",assert_value)
|
||
print("实际值", resp_data)
|
||
self._assert_type(types=self.assert_data[key]['type'], key=resp_data, value=assert_value)
|
||
# elif assert_type == 'noResponse':
|
||
# print("预期值", assert_value)
|
||
# print("实际值", resp_data[0])
|
||
# self._assert_type(types=self.assert_data[key]['type'], key=resp_data, value=assert_value)
|
||
elif assert_type is None:
|
||
# print("预期值", assert_value)
|
||
# print("实际值" ,resp_data[0])
|
||
self._assert_type(types=self.assert_data[key]['type'], key=resp_data[0], value=assert_value)
|
||
else:
|
||
raise ValueError("断言失败，目前只支持数据库断言和响应断言")
|
||
|
||
def assert_equality(self, response_data, sql_data):
|
||
# 判断数据类型
|
||
if self._check_params(response_data, sql_data) is not False:
|
||
for key, values in self.assert_data.items():
|
||
assert_value = self.assert_data[key]['value'] # 获取 yaml 文件中的期望value值
|
||
assert_jsonpath = self.assert_data[key]['jsonpath'] # 获取到 yaml断言中的jsonpath的数据
|
||
|
||
|
||
assert_type = self.assert_data[key]['AssertType']
|
||
# 从yaml获取jsonpath，拿到对象的接口响应数据
|
||
if assert_type=='str':
|
||
resp_data = response_data
|
||
print("resp_dataresp_dataresp_data",resp_data)
|
||
else:
|
||
resp_data = jsonpath.jsonpath(response_data, assert_jsonpath)
|
||
|
||
# jsonpath 如果数据获取失败，会返回False，判断获取成功才会执行如下代码
|
||
if resp_data is not False:
|
||
# 判断断言类型
|
||
# print('断言',assert_type, sql_data, assert_value, key, values, resp_data)
|
||
self.assert_type_handle(assert_type, sql_data, assert_value, key, values, resp_data)
|
||
else:
|
||
ERROR.logger.error("JsonPath值获取失败{}".format(assert_jsonpath))
|
||
raise ValueError(f"JsonPath值获取失败{assert_jsonpath}")
|
||
else:
|
||
raise '断言失败'
|
||
pass
|
||
|
||
#
|
||
# if __name__ == '__main__':
|
||
# from tools.readfiletools.yaml_data_analysis import CaseData
|
||
# from config.setting import ConfigHandler
|
||
# #获取用例清洗后的数据
|
||
# GetCaseData = CaseData(ConfigHandler.merchant_data_path + r'test_dm\InFo.yaml').case_process()
|
||
# #获取用例列表里面第一个用例的 assert
|
||
# cc = GetCaseData[0]['assert']
|
||
# aa = {'status': 200, 'message': 'ok', 'data': '12312312'}
|
||
# bb = {'sql': None}
|
||
# Assert(cc).assert_equality(response_data=aa,sql_data=bb)
|