提取请求参数的bug
This commit is contained in:
chenyongzhiaaron
parent
48bc2f0907
commit
6536cb5f41
|
|
@ -5,7 +5,7 @@
|
|||
<sourceFolder url="file://$MODULE_DIR$" isTestSource="false" />
|
||||
<excludeFolder url="file://$MODULE_DIR$/venv" />
|
||||
</content>
|
||||
<orderEntry type="jdk" jdkName="Pipenv (apitest)" jdkType="Python SDK" />
|
||||
<orderEntry type="inheritedJdk" />
|
||||
<orderEntry type="sourceFolder" forTests="false" />
|
||||
</component>
|
||||
<component name="PackageRequirementsSettings">
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="dataSourceStorageLocal" created-in="PY-231.9011.38">
|
||||
<component name="dataSourceStorageLocal" created-in="PY-231.9161.41">
|
||||
<data-source name="@localhost" uuid="49b6f686-3676-4df5-9645-cd7a2fe91d80">
|
||||
<database-info product="MySQL" version="8.0.26" jdbc-version="4.2" driver-name="MySQL Connector/J" driver-version="mysql-connector-java-8.0.25 (Revision: 08be9e9b4cba6aa115f9b27b215887af40b159e0)" dbms="MYSQL" exact-version="8.0.26" exact-driver-version="8.0">
|
||||
<extra-name-characters>#@</extra-name-characters>
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Pipenv (apitest)" project-jdk-type="Python SDK" />
|
||||
<component name="ProjectRootManager" version="2" project-jdk-name="Pipenv (api-test-project)" project-jdk-type="Python SDK" />
|
||||
<component name="PyCharmProfessionalAdvertiser">
|
||||
<option name="shown" value="true" />
|
||||
</component>
|
||||
|
|
|
|||
|
|
@ -69,3 +69,25 @@ HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /
|
|||
2023-07-14 00:46:05 | ERROR | 被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: None
|
||||
2023-07-14 00:46:05 | ERROR | 异常用例: safe_5_劳务基础配置_查询配置
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/api/getTvInfo?t=1689266765926&projectId=3333 (Caused by ProxyError('Unable to connect to proxy', SSLError(SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:1123)'))))
|
||||
2023-07-14 17:56:48 | ERROR | 异常用例: safe_2_登录_非BIP用户登录
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/login (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x00000279F94EBE80>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
2023-07-14 17:56:50 | ERROR | key:<re.Match object; span=(127, 140), match='{{BSP_TOKEN}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
|
||||
|
||||
2023-07-14 17:56:56 | ERROR | 异常用例: safe_4_劳务基础配置_绑定TV
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/api/addTvSequence?t=1689328610847 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x00000279F952B0A0>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
2023-07-14 17:56:57 | ERROR | key:<re.Match object; span=(138, 151), match='{{BSP_TOKEN}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
|
||||
|
||||
2023-07-14 17:56:57 | ERROR | 被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: None
|
||||
2023-07-14 17:57:03 | ERROR | 异常用例: safe_5_劳务基础配置_查询配置
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/api/getTvInfo?t=1689328617901&projectId=3333 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x00000279F94C9280>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
2023-07-14 18:05:55 | ERROR | 异常用例: safe_2_登录_非BIP用户登录
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/login (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x000002B958C01370>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
2023-07-14 18:05:56 | ERROR | key:<re.Match object; span=(127, 140), match='{{BSP_TOKEN}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
|
||||
|
||||
2023-07-14 18:06:05 | ERROR | 异常用例: safe_4_劳务基础配置_绑定TV
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/api/addTvSequence?t=1689329156326 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x000002B958CC8400>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
2023-07-14 18:06:06 | ERROR | key:<re.Match object; span=(138, 151), match='{{BSP_TOKEN}}'>,在关联参数表中查询不到,请检查关联参数字段提取及填写是否正常
|
||||
|
||||
2023-07-14 18:06:06 | ERROR | 被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: None
|
||||
2023-07-14 18:06:20 | ERROR | 异常用例: safe_5_劳务基础配置_查询配置
|
||||
HTTPSConnectionPool(host='xxxx.com', port=443): Max retries exceeded with url: /test/api/getTvInfo?t=1689329166696&projectId=3333 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x000002B958C69D90>: Failed to establish a new connection: [WinError 10061] 由于目标计算机积极拒绝，无法连接。'))
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -10,54 +10,54 @@
|
|||
import ast
|
||||
|
||||
from common.crypto.encrypt_data import EncryptData
|
||||
from common.utils.load_and_execute_script import LoadScript
|
||||
from common.utils.decorators import singleton
|
||||
from common.utils.load_and_execute_script import LoadScript
|
||||
from common.validation.extractor import Extractor
|
||||
from common.validation.validator import Validator
|
||||
|
||||
|
||||
@singleton
|
||||
class Action(Extractor, LoadScript, Validator):
|
||||
def __init__(self, initialize_data=None, bif_functions=None):
|
||||
super().__init__()
|
||||
self.encrypt = EncryptData()
|
||||
self.__variables = {}
|
||||
self.set_environments(initialize_data)
|
||||
self.set_bif_fun(bif_functions)
|
||||
|
||||
def execute_dynamic_code(self, item, code):
|
||||
self.set_variables(item)
|
||||
if code is not None:
|
||||
try:
|
||||
ast_obj = ast.parse(code, mode='exec')
|
||||
compiled = compile(ast_obj, '<string>', 'exec')
|
||||
exec(compiled, {"pm": self})
|
||||
except SyntaxError as e:
|
||||
error_message = f'Syntax error in dynamic code: {e}'
|
||||
self._handle_error(error_message)
|
||||
except Exception as e:
|
||||
error_message = f"Error executing dynamic code: {e}"
|
||||
self._handle_error(error_message)
|
||||
finally:
|
||||
return self.__variables
|
||||
return item
|
||||
|
||||
def _handle_error(self, error_message):
|
||||
print(f'Error occurred: {error_message}')
|
||||
|
||||
def set_variables(self, item):
|
||||
self.__variables = item
|
||||
|
||||
def update_variables(self, key, value):
|
||||
self.__variables[f"{{{{{key}}}}}"] = value
|
||||
|
||||
def get_variables(self, key=None):
|
||||
"""获取依赖表 或 依赖表中key对应的值"""
|
||||
return self.__variables if not key else self.__variables.get(key)
|
||||
|
||||
def analysis_request(self, request_data, jp_dict, headers_crypto, headers, request_crypto):
|
||||
# 提取请求参数信息
|
||||
self.substitute_data(request_data, jp_dict=jp_dict)
|
||||
# 请求头及body加密或者加签
|
||||
headers, request_data = self.encrypt.encrypts(headers_crypto, headers, request_crypto, request_data)
|
||||
return headers, request_data
|
||||
def __init__(self, initialize_data=None, bif_functions=None):
|
||||
super().__init__()
|
||||
self.encrypt = EncryptData()
|
||||
self.__variables = {}
|
||||
self.set_environments(initialize_data)
|
||||
self.set_bif_fun(bif_functions)
|
||||
|
||||
def execute_dynamic_code(self, item, code):
|
||||
self.set_variables(item)
|
||||
if code is not None:
|
||||
try:
|
||||
ast_obj = ast.parse(code, mode='exec')
|
||||
compiled = compile(ast_obj, '<string>', 'exec')
|
||||
exec(compiled, {"pm": self})
|
||||
except SyntaxError as e:
|
||||
error_message = f'Syntax error in dynamic code: {e}'
|
||||
self._handle_error(error_message)
|
||||
except Exception as e:
|
||||
error_message = f"Error executing dynamic code: {e}"
|
||||
self._handle_error(error_message)
|
||||
finally:
|
||||
return self.__variables
|
||||
return item
|
||||
|
||||
def _handle_error(self, error_message):
|
||||
print(f'Error occurred: {error_message}')
|
||||
|
||||
def set_variables(self, item):
|
||||
self.__variables = item
|
||||
|
||||
def update_variables(self, key, value):
|
||||
self.__variables[f"{{{{{key}}}}}"] = value
|
||||
|
||||
def get_variables(self, key=None):
|
||||
"""获取依赖表 或 依赖表中key对应的值"""
|
||||
return self.__variables if not key else self.__variables.get(key)
|
||||
|
||||
def analysis_request(self, request_data, headers_crypto, headers, request_crypto, extract_request_data):
|
||||
# 请求头及body加密或者加签
|
||||
headers, request_data = self.encrypt.encrypts(headers_crypto, headers, request_crypto, request_data)
|
||||
# 提取请求参数信息
|
||||
self.substitute_data(request_data, jp_dict=extract_request_data)
|
||||
return headers, request_data
|
||||
|
|
|
|||
|
|
@ -15,114 +15,112 @@ from common.validation.loaders import Loaders
|
|||
|
||||
|
||||
class Validator(Loaders):
|
||||
"""
|
||||
校验器
|
||||
主要功能：
|
||||
1、格式化校验变量
|
||||
2、校验期望结果与实际结果与预期一致，并返回校验结果
|
||||
"""
|
||||
validate_variables_list = []
|
||||
built_in_comparators = Loaders.load_built_in_comparators()
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def uniform_validate(self, validate_variables):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate
|
||||
Args:
|
||||
validate_variables: 参数格式 list、dict
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
or {"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}
|
||||
"""
|
||||
校验器
|
||||
主要功能：
|
||||
1、格式化校验变量
|
||||
2、校验期望结果与实际结果与预期一致，并返回校验结果
|
||||
"""
|
||||
validate_variables_list = []
|
||||
built_in_comparators = Loaders.load_built_in_comparators()
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def uniform_validate(self, validate_variables):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate
|
||||
Args:
|
||||
validate_variables: 参数格式 list、dict
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
or {"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}
|
||||
|
||||
Returns: 返回数据格式 list
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
Returns: 返回数据格式 list
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
|
||||
"""
|
||||
if isinstance(validate_variables, str):
|
||||
validate_variables = json.loads(validate_variables)
|
||||
if isinstance(validate_variables, list):
|
||||
for item in validate_variables:
|
||||
self.uniform_validate(item)
|
||||
elif isinstance(validate_variables, dict):
|
||||
if "check" in validate_variables.keys() and "expect" in validate_variables.keys():
|
||||
# 如果验证mapping中不包含comparator时，默认为{"comparator": "eq"}
|
||||
check_item = validate_variables.get("check")
|
||||
expect_value = validate_variables.get("expect")
|
||||
comparator = validate_variables.get("comparator", "eq")
|
||||
self.validate_variables_list.append({
|
||||
"check": check_item,
|
||||
"expect": expect_value,
|
||||
"comparator": comparator
|
||||
})
|
||||
else:
|
||||
logger.error("参数格式错误！")
|
||||
|
||||
def validate(self, resp_obj=None):
|
||||
"""
|
||||
校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
resp_obj: ResponseObject对象实例
|
||||
"""
|
||||
if isinstance(validate_variables, str):
|
||||
validate_variables = json.loads(validate_variables)
|
||||
if isinstance(validate_variables, list):
|
||||
for item in validate_variables:
|
||||
self.uniform_validate(item)
|
||||
elif isinstance(validate_variables, dict):
|
||||
if "check" in validate_variables.keys() and "expect" in validate_variables.keys():
|
||||
# 如果验证mapping中不包含comparator时，默认为{"comparator": "eq"}
|
||||
check_item = validate_variables.get("check")
|
||||
expect_value = validate_variables.get("expect")
|
||||
comparator = validate_variables.get("comparator", "eq")
|
||||
self.validate_variables_list.append({
|
||||
"check": check_item,
|
||||
"expect": expect_value,
|
||||
"comparator": comparator
|
||||
})
|
||||
else:
|
||||
logger.error("参数格式错误！")
|
||||
|
||||
def validate(self, resp_obj=None):
|
||||
"""
|
||||
校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
resp_obj: ResponseObject对象实例
|
||||
|
||||
Returns:
|
||||
Returns:
|
||||
|
||||
"""
|
||||
|
||||
validate_pass = "PASS"
|
||||
|
||||
# 记录校验失败的原因
|
||||
failure_reason = []
|
||||
for validate_variable in self.validate_variables_list:
|
||||
check_item = validate_variable['check']
|
||||
expect_value = validate_variable['expect']
|
||||
comparator = validate_variable['comparator']
|
||||
actual_value = Extractor.extract_value_by_jsonpath(resp_obj=resp_obj, expr=check_item)
|
||||
try:
|
||||
# 获取比较器
|
||||
fun = self.built_in_comparators[comparator]
|
||||
fun(actual_value=actual_value, expect_value=expect_value)
|
||||
except (AssertionError, TypeError):
|
||||
validate_pass = "FAIL"
|
||||
failure_reason.append({
|
||||
'检查项': check_item,
|
||||
'期望值': expect_value,
|
||||
'实际值': actual_value,
|
||||
'断言方法': comparator_dict.get(comparator),
|
||||
})
|
||||
return validate_pass, failure_reason
|
||||
|
||||
def run_validate(self, validate_variables, resp_obj=None):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate，然后校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
validate_variables:参数格式 list、dict
|
||||
resp_obj:ResponseObject对象实例
|
||||
"""
|
||||
|
||||
validate_pass = "PASS"
|
||||
|
||||
# 记录校验失败的原因
|
||||
failure_reason = []
|
||||
for validate_variable in self.validate_variables_list:
|
||||
check_item = validate_variable['check']
|
||||
expect_value = validate_variable['expect']
|
||||
comparator = validate_variable['comparator']
|
||||
actual_value = Extractor.extract_value_by_jsonpath(resp_obj=resp_obj, expr=check_item)
|
||||
try:
|
||||
# 获取比较器
|
||||
fun = self.built_in_comparators[comparator]
|
||||
fun(actual_value=actual_value, expect_value=expect_value)
|
||||
except (AssertionError, TypeError):
|
||||
validate_pass = "FAIL"
|
||||
failure_reason.append({
|
||||
'检查项': check_item,
|
||||
'期望值': expect_value,
|
||||
'实际值': actual_value,
|
||||
'断言方法': comparator_dict.get(comparator),
|
||||
})
|
||||
return validate_pass, failure_reason
|
||||
|
||||
def run_validate(self, validate_variables, resp_obj=None):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate，然后校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
validate_variables:参数格式 list、dict
|
||||
resp_obj:ResponseObject对象实例
|
||||
|
||||
Returns:返回校验结果
|
||||
Returns:返回校验结果
|
||||
|
||||
"""
|
||||
if not validate_variables:
|
||||
return ""
|
||||
# 清空校验变量
|
||||
self.validate_variables_list.clear()
|
||||
self.uniform_validate(validate_variables)
|
||||
if not self.validate_variables_list:
|
||||
raise "uniform_validate 执行失败，无法进行 validate 校验"
|
||||
return self.validate(resp_obj)
|
||||
"""
|
||||
if not validate_variables:
|
||||
return ""
|
||||
# 清空校验变量
|
||||
self.validate_variables_list.clear()
|
||||
self.uniform_validate(validate_variables)
|
||||
if not self.validate_variables_list:
|
||||
raise "uniform_validate 执行失败，无法进行 validate 校验"
|
||||
return self.validate(resp_obj)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# validate_variables1 = {"check": "$.result.user.name", "comparator": "eq", "expect": "chenyongzhi"}
|
||||
validate_variables2 = [
|
||||
{"check": "code", "comparator": "eq", "expect": "200"},
|
||||
{"check": "result.user.name", "comparator": "eq", "expect": "chenyongzhi"}
|
||||
]
|
||||
resp_obj = {"code": "200", "result": {"user": {"name": "chenyongzhi"}}}
|
||||
validator = Validator()
|
||||
print(validator.run_validate(validate_variables2, resp_obj))
|
||||
# logger.info("---")
|
||||
# validator.run_validate(validate_variables2, resp_obj)
|
||||
# logger.info("---")
|
||||
# validator.run_validate(validate_variables2, resp_obj)
|
||||
validate_variables1 = {"check": "$.result.user.name", "comparator": "eq", "expect": "chenyongzhi"}
|
||||
validate_variables2 = [
|
||||
{"check": "code", "comparator": "eq", "expect": "200"},
|
||||
{"check": "result.user.name", "comparator": "eq", "expect": "chenyongzhi"}
|
||||
]
|
||||
resp_obj = {"code": "200", "result": {"user": {"name": "chenyongzhi"}}}
|
||||
validator = Validator()
|
||||
print(validator.run_validate(validate_variables1, resp_obj))
|
||||
print(validator.run_validate(validate_variables2, resp_obj))
|
||||
print(validator.run_validate(validate_variables2, resp_obj))
|
||||
|
|
|
|||
|
|
@ -25,173 +25,173 @@ host = init_data.get('host', "") + init_data.get("path", "")
|
|||
|
||||
@ddt
|
||||
class TestProjectApi(unittest.TestCase):
|
||||
maxDiff = None
|
||||
action = Action(initialize_data, bif_functions)
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
pass
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.action.set_bif_fun(dynamic_scaling_methods)
|
||||
|
||||
@data(*test_case)
|
||||
def test_api(self, item):
|
||||
|
||||
sheet, iid, condition, st, name, desc, h_crypto, r_crypto, method = self.__base_info(item)
|
||||
regex, keys, deps, jp_dict, extract_request_data = self.__extractor_info(item)
|
||||
setup_script, teardown_script = self.script(item)
|
||||
|
||||
if self.__is_run(condition):
|
||||
return
|
||||
|
||||
self.__pause_execution(st)
|
||||
|
||||
# 首执行 sql
|
||||
self.__exc_sql(item, method)
|
||||
|
||||
# 执行动态代码
|
||||
item = self.action.execute_dynamic_code(item, setup_script)
|
||||
|
||||
# prepost_script = f"prepost_script_{sheet}_{iid}.py"
|
||||
# item = self.action.load_and_execute_script(Config.SCRIPTS_DIR, prepost_script, "setup", item)
|
||||
|
||||
# 修正参数
|
||||
item = self.action.replace_dependent_parameter(item)
|
||||
url, query_str, request_data, headers, expected, request_data_type = self.__request_info(item)
|
||||
|
||||
# 分析请求参数信息
|
||||
headers, request_data = self.action.analysis_request(request_data, jp_dict, h_crypto, headers, r_crypto)
|
||||
result_tuple = None
|
||||
result = "PASS"
|
||||
response = None
|
||||
|
||||
try:
|
||||
# 执行请求操作
|
||||
kwargs = {request_data_type: request_data, 'headers': headers, "params": query_str}
|
||||
response = self.action.http_client(host, url, method, **kwargs)
|
||||
|
||||
# 执行后置代码片段
|
||||
self.action.execute_dynamic_code(response, teardown_script)
|
||||
|
||||
# 执行断言 返回结果元组
|
||||
result_tuple = self.action.run_validate(expected, response.json())
|
||||
self.assertNotIn("FAIL", result_tuple, "FAIL 存在结果元组中")
|
||||
try:
|
||||
# 提取响应
|
||||
self.action.substitute_data(response.json(), regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
|
||||
|
||||
except Exception as err:
|
||||
logger.error(f"提取响应失败：{sheet}_{iid}_{name}_{desc}"
|
||||
f"\nregex={regex};"
|
||||
f" \nkeys={keys};"
|
||||
f"\ndeps={deps};"
|
||||
f"\njp_dict={jp_dict}"
|
||||
f"\n{err}")
|
||||
except Exception as e:
|
||||
result = "FAIL"
|
||||
logger.error(f'异常用例: {sheet}_{iid}_{name}_{desc}\n{e}')
|
||||
raise e
|
||||
finally:
|
||||
response = response.text if response is not None else str(response)
|
||||
# 响应结果及测试结果回写 excel
|
||||
excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
|
||||
assert_log=str(result_tuple))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
excel.close_excel()
|
||||
logger.info(f"所有用例执行完毕")
|
||||
|
||||
@staticmethod
|
||||
def __base_info(item):
|
||||
"""
|
||||
获取基础信息
|
||||
"""
|
||||
sheet = item.pop("sheet")
|
||||
item_id = item.pop("Id")
|
||||
condition = item.pop("Run")
|
||||
sleep_time = item.pop("Time")
|
||||
name = item.pop("Name")
|
||||
desc = item.pop("Description")
|
||||
headers_crypto = item.pop("Headers Crypto")
|
||||
request_data_crypto = item.pop("Request Data Crypto")
|
||||
method = item.pop("Method")
|
||||
return sheet, item_id, condition, sleep_time, name, desc, headers_crypto, request_data_crypto, method
|
||||
|
||||
@staticmethod
|
||||
def __sql_info(item):
|
||||
sql = item.pop("SQL")
|
||||
sql_params_dict = item.pop("Sql Params Dict")
|
||||
return sql, sql_params_dict
|
||||
|
||||
@staticmethod
|
||||
def __extractor_info(item):
|
||||
"""
|
||||
获取提取参数的基本字段信息
|
||||
Args:
|
||||
item:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
regex = item.pop("Regex")
|
||||
keys = item.pop("Regex Params List")
|
||||
deps = item.pop("Retrieve Value")
|
||||
jp_dict = item.pop("Jsonpath")
|
||||
extract_request_data = item.pop("Extract Request Data")
|
||||
return regex, keys, deps, jp_dict, extract_request_data
|
||||
|
||||
@staticmethod
|
||||
def __request_info(item):
|
||||
"""
|
||||
请求数据
|
||||
"""
|
||||
url = item.pop("Url")
|
||||
query_str = item.pop("Query Str")
|
||||
request_data = item.pop("Request Data")
|
||||
headers = item.pop("Headers")
|
||||
expected = item.pop("Expected")
|
||||
request_data_type = item.pop("Request Data Type") if item.get("Request Data Type") else 'params'
|
||||
|
||||
return url, query_str, request_data, headers, expected, request_data_type
|
||||
|
||||
@staticmethod
|
||||
def script(item):
|
||||
setup_script = item.pop("Setup Script")
|
||||
teardown_script = item.pop("Teardown Script")
|
||||
return setup_script, teardown_script
|
||||
|
||||
@staticmethod
|
||||
def __is_run(condition):
|
||||
is_run = condition
|
||||
if not is_run or is_run.upper() != "YES":
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def __pause_execution(sleep_time):
|
||||
if sleep_time:
|
||||
try:
|
||||
time.sleep(sleep_time)
|
||||
except Exception as e:
|
||||
logger.error("暂时时间必须是数字")
|
||||
raise e
|
||||
|
||||
def __exc_sql(self, item, method):
|
||||
sql, sql_params_dict = self.__sql_info(item)
|
||||
sql = self.action.replace_dependent_parameter(sql)
|
||||
if sql:
|
||||
try:
|
||||
execute_sql_results = mysql.do_mysql(sql)
|
||||
if execute_sql_results and sql_params_dict:
|
||||
self.action.extract_request_data(execute_sql_results, jp_dict=sql_params_dict)
|
||||
if method == "SQL" and mysql:
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f'执行 sql 失败:{sql},异常信息:{e}')
|
||||
raise e
|
||||
return sql
|
||||
maxDiff = None
|
||||
action = Action(initialize_data, bif_functions)
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
pass
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.action.set_bif_fun(dynamic_scaling_methods)
|
||||
|
||||
@data(*test_case)
|
||||
def test_api(self, item):
|
||||
|
||||
sheet, iid, condition, st, name, desc, h_crypto, r_crypto, method = self.__base_info(item)
|
||||
regex, keys, deps, jp_dict, extract_request_data = self.__extractor_info(item)
|
||||
setup_script, teardown_script = self.script(item)
|
||||
|
||||
if self.__is_run(condition):
|
||||
return
|
||||
|
||||
self.__pause_execution(st)
|
||||
|
||||
# 首执行 sql
|
||||
self.__exc_sql(item, method)
|
||||
|
||||
# 执行动态代码
|
||||
item = self.action.execute_dynamic_code(item, setup_script)
|
||||
|
||||
# prepost_script = f"prepost_script_{sheet}_{iid}.py"
|
||||
# item = self.action.load_and_execute_script(Config.SCRIPTS_DIR, prepost_script, "setup", item)
|
||||
|
||||
# 修正参数
|
||||
item = self.action.replace_dependent_parameter(item)
|
||||
url, query_str, request_data, headers, expected, request_data_type = self.__request_info(item)
|
||||
|
||||
# 分析请求参数信息
|
||||
headers, request_data = self.action.analysis_request(request_data, h_crypto, headers, r_crypto,extract_request_data)
|
||||
result_tuple = None
|
||||
result = "PASS"
|
||||
response = None
|
||||
|
||||
try:
|
||||
# 执行请求操作
|
||||
kwargs = {request_data_type: request_data, 'headers': headers, "params": query_str}
|
||||
response = self.action.http_client(host, url, method, **kwargs)
|
||||
|
||||
# 执行后置代码片段
|
||||
self.action.execute_dynamic_code(response, teardown_script)
|
||||
|
||||
# 执行断言 返回结果元组
|
||||
result_tuple = self.action.run_validate(expected, response.json())
|
||||
self.assertNotIn("FAIL", result_tuple, "FAIL 存在结果元组中")
|
||||
try:
|
||||
# 提取响应
|
||||
self.action.substitute_data(response.json(), regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
|
||||
|
||||
except Exception as err:
|
||||
logger.error(f"提取响应失败：{sheet}_{iid}_{name}_{desc}"
|
||||
f"\nregex={regex};"
|
||||
f" \nkeys={keys};"
|
||||
f"\ndeps={deps};"
|
||||
f"\njp_dict={jp_dict}"
|
||||
f"\n{err}")
|
||||
except Exception as e:
|
||||
result = "FAIL"
|
||||
logger.error(f'异常用例: {sheet}_{iid}_{name}_{desc}\n{e}')
|
||||
raise e
|
||||
finally:
|
||||
response = response.text if response is not None else str(response)
|
||||
# 响应结果及测试结果回写 excel
|
||||
excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
|
||||
assert_log=str(result_tuple))
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
excel.close_excel()
|
||||
logger.info(f"所有用例执行完毕")
|
||||
|
||||
@staticmethod
|
||||
def __base_info(item):
|
||||
"""
|
||||
获取基础信息
|
||||
"""
|
||||
sheet = item.pop("sheet")
|
||||
item_id = item.pop("Id")
|
||||
condition = item.pop("Run")
|
||||
sleep_time = item.pop("Time")
|
||||
name = item.pop("Name")
|
||||
desc = item.pop("Description")
|
||||
headers_crypto = item.pop("Headers Crypto")
|
||||
request_data_crypto = item.pop("Request Data Crypto")
|
||||
method = item.pop("Method")
|
||||
return sheet, item_id, condition, sleep_time, name, desc, headers_crypto, request_data_crypto, method
|
||||
|
||||
@staticmethod
|
||||
def __sql_info(item):
|
||||
sql = item.pop("SQL")
|
||||
sql_params_dict = item.pop("Sql Params Dict")
|
||||
return sql, sql_params_dict
|
||||
|
||||
@staticmethod
|
||||
def __extractor_info(item):
|
||||
"""
|
||||
获取提取参数的基本字段信息
|
||||
Args:
|
||||
item:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
regex = item.pop("Regex")
|
||||
keys = item.pop("Regex Params List")
|
||||
deps = item.pop("Retrieve Value")
|
||||
jp_dict = item.pop("Jsonpath")
|
||||
extract_request_data = item.pop("Extract Request Data")
|
||||
return regex, keys, deps, jp_dict, extract_request_data
|
||||
|
||||
@staticmethod
|
||||
def __request_info(item):
|
||||
"""
|
||||
请求数据
|
||||
"""
|
||||
url = item.pop("Url")
|
||||
query_str = item.pop("Query Str")
|
||||
request_data = item.pop("Request Data")
|
||||
headers = item.pop("Headers")
|
||||
expected = item.pop("Expected")
|
||||
request_data_type = item.pop("Request Data Type") if item.get("Request Data Type") else 'params'
|
||||
|
||||
return url, query_str, request_data, headers, expected, request_data_type
|
||||
|
||||
@staticmethod
|
||||
def script(item):
|
||||
setup_script = item.pop("Setup Script")
|
||||
teardown_script = item.pop("Teardown Script")
|
||||
return setup_script, teardown_script
|
||||
|
||||
@staticmethod
|
||||
def __is_run(condition):
|
||||
is_run = condition
|
||||
if not is_run or is_run.upper() != "YES":
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def __pause_execution(sleep_time):
|
||||
if sleep_time:
|
||||
try:
|
||||
time.sleep(sleep_time)
|
||||
except Exception as e:
|
||||
logger.error("暂时时间必须是数字")
|
||||
raise e
|
||||
|
||||
def __exc_sql(self, item, method):
|
||||
sql, sql_params_dict = self.__sql_info(item)
|
||||
sql = self.action.replace_dependent_parameter(sql)
|
||||
if sql:
|
||||
try:
|
||||
execute_sql_results = mysql.do_mysql(sql)
|
||||
if execute_sql_results and sql_params_dict:
|
||||
self.action.extract_request_data(execute_sql_results, jp_dict=sql_params_dict)
|
||||
if method == "SQL" and mysql:
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f'执行 sql 失败:{sql},异常信息:{e}')
|
||||
raise e
|
||||
return sql
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Reference in New Issue