兼容断言检查项不使用jsonpath提取参数
This commit is contained in:
chenyongzhiaaron
parent
737624746f
commit
ffe6fb2553
|
|
@ -15,6 +15,9 @@ OutPut/log
|
|||
*.ipa
|
||||
*.apk
|
||||
*.log
|
||||
*.xhtml
|
||||
*.xml
|
||||
.idea
|
||||
__pycache__
|
||||
# Project files, i.e. `.project`, `.actionScriptProperties` and `.flexProperties`
|
||||
# should NOT be excluded as they contain compiler settings and other important
|
||||
|
|
|
|||
|
|
@ -4,10 +4,7 @@
|
|||
<option name="autoReloadType" value="SELECTIVE" />
|
||||
</component>
|
||||
<component name="ChangeListManager">
|
||||
<list default="true" id="eded71e8-6551-463d-9bd4-cdbb3ffc536c" name="更改" comment="增加处理函数调用链变量以及修复动态函数中传参失效的问题">
|
||||
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
|
||||
<change beforePath="$PROJECT_DIR$/common/data_extraction/dependent_parameter.py" beforeDir="false" afterPath="$PROJECT_DIR$/common/data_extraction/dependent_parameter.py" afterDir="false" />
|
||||
</list>
|
||||
<list default="true" id="eded71e8-6551-463d-9bd4-cdbb3ffc536c" name="更改" comment="增加处理函数调用链变量以及修复动态函数中传参失效的问题" />
|
||||
<option name="SHOW_DIALOG" value="false" />
|
||||
<option name="HIGHLIGHT_CONFLICTS" value="true" />
|
||||
<option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
|
||||
|
|
@ -69,7 +66,29 @@
|
|||
<recent name="D:\app\apitest\common\parsing" />
|
||||
</key>
|
||||
</component>
|
||||
<component name="RunManager" selected="Python.dependent_parameter">
|
||||
<component name="RunManager" selected="Python 测试.Python 测试 (test_executor.py 内)">
|
||||
<configuration name="assert_dict" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
<envs>
|
||||
<env name="PYTHONUNBUFFERED" value="1" />
|
||||
</envs>
|
||||
<option name="SDK_HOME" value="" />
|
||||
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/common/data_extraction" />
|
||||
<option name="IS_MODULE_SDK" value="true" />
|
||||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/common/data_extraction/assert_dict.py" />
|
||||
<option name="PARAMETERS" value="" />
|
||||
<option name="SHOW_COMMAND_LINE" value="false" />
|
||||
<option name="EMULATE_TERMINAL" value="false" />
|
||||
<option name="MODULE_MODE" value="false" />
|
||||
<option name="REDIRECT_INPUT" value="false" />
|
||||
<option name="INPUT_FILE" value="" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<configuration name="dependent_parameter" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
|
|
@ -92,7 +111,7 @@
|
|||
<option name="INPUT_FILE" value="" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<configuration name="load_modules_from_folder" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<configuration name="extractor" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
|
|
@ -105,7 +124,7 @@
|
|||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/common/validation/load_modules_from_folder.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/common/validation/extractor.py" />
|
||||
<option name="PARAMETERS" value="" />
|
||||
<option name="SHOW_COMMAND_LINE" value="false" />
|
||||
<option name="EMULATE_TERMINAL" value="false" />
|
||||
|
|
@ -114,7 +133,7 @@
|
|||
<option name="INPUT_FILE" value="" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<configuration name="loaders" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<configuration name="validator" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
|
|
@ -127,7 +146,7 @@
|
|||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/common/validation/loaders.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/common/validation/validator.py" />
|
||||
<option name="PARAMETERS" value="" />
|
||||
<option name="SHOW_COMMAND_LINE" value="false" />
|
||||
<option name="EMULATE_TERMINAL" value="false" />
|
||||
|
|
@ -136,57 +155,28 @@
|
|||
<option name="INPUT_FILE" value="" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<configuration name="method_chain" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<configuration name="Python 测试 (test_executor.py 内)" type="tests" factoryName="Autodetect" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
<envs>
|
||||
<env name="PYTHONUNBUFFERED" value="1" />
|
||||
</envs>
|
||||
<option name="SDK_HOME" value="" />
|
||||
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/debug" />
|
||||
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/test_script" />
|
||||
<option name="IS_MODULE_SDK" value="true" />
|
||||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/debug/method_chain.py" />
|
||||
<option name="PARAMETERS" value="" />
|
||||
<option name="SHOW_COMMAND_LINE" value="false" />
|
||||
<option name="EMULATE_TERMINAL" value="false" />
|
||||
<option name="MODULE_MODE" value="false" />
|
||||
<option name="REDIRECT_INPUT" value="false" />
|
||||
<option name="INPUT_FILE" value="" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<configuration name="re_chain" type="PythonConfigurationType" factoryName="Python" temporary="true" nameIsGenerated="true">
|
||||
<module name="api_project" />
|
||||
<option name="INTERPRETER_OPTIONS" value="" />
|
||||
<option name="PARENT_ENVS" value="true" />
|
||||
<envs>
|
||||
<env name="PYTHONUNBUFFERED" value="1" />
|
||||
</envs>
|
||||
<option name="SDK_HOME" value="" />
|
||||
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/debug" />
|
||||
<option name="IS_MODULE_SDK" value="true" />
|
||||
<option name="ADD_CONTENT_ROOTS" value="true" />
|
||||
<option name="ADD_SOURCE_ROOTS" value="true" />
|
||||
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
|
||||
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/debug/re_chain.py" />
|
||||
<option name="PARAMETERS" value="" />
|
||||
<option name="SHOW_COMMAND_LINE" value="false" />
|
||||
<option name="EMULATE_TERMINAL" value="false" />
|
||||
<option name="MODULE_MODE" value="false" />
|
||||
<option name="REDIRECT_INPUT" value="false" />
|
||||
<option name="INPUT_FILE" value="" />
|
||||
<option name="_new_additionalArguments" value="""" />
|
||||
<option name="_new_target" value=""$PROJECT_DIR$/test_script/test_executor.py"" />
|
||||
<option name="_new_targetType" value=""PATH"" />
|
||||
<method v="2" />
|
||||
</configuration>
|
||||
<recent_temporary>
|
||||
<list>
|
||||
<item itemvalue="Python 测试.Python 测试 (test_executor.py 内)" />
|
||||
<item itemvalue="Python.assert_dict" />
|
||||
<item itemvalue="Python.validator" />
|
||||
<item itemvalue="Python.extractor" />
|
||||
<item itemvalue="Python.dependent_parameter" />
|
||||
<item itemvalue="Python.method_chain" />
|
||||
<item itemvalue="Python.re_chain" />
|
||||
<item itemvalue="Python.loaders" />
|
||||
<item itemvalue="Python.load_modules_from_folder" />
|
||||
</list>
|
||||
</recent_temporary>
|
||||
</component>
|
||||
|
|
@ -218,7 +208,8 @@
|
|||
<workItem from="1690855086910" duration="749000" />
|
||||
<workItem from="1690855882311" duration="1981000" />
|
||||
<workItem from="1690873614157" duration="4828000" />
|
||||
<workItem from="1690937027137" duration="26771000" />
|
||||
<workItem from="1690937027137" duration="37912000" />
|
||||
<workItem from="1691034024300" duration="1062000" />
|
||||
</task>
|
||||
<task id="LOCAL-00001" summary="优化代码">
|
||||
<option name="closed" value="true" />
|
||||
|
|
@ -316,7 +307,15 @@
|
|||
<option name="project" value="LOCAL" />
|
||||
<updated>1690969473478</updated>
|
||||
</task>
|
||||
<option name="localTasksCounter" value="13" />
|
||||
<task id="LOCAL-00013" summary="增加处理函数调用链变量以及修复动态函数中传参失效的问题">
|
||||
<option name="closed" value="true" />
|
||||
<created>1690970941519</created>
|
||||
<option name="number" value="00013" />
|
||||
<option name="presentableId" value="LOCAL-00013" />
|
||||
<option name="project" value="LOCAL" />
|
||||
<updated>1690970941519</updated>
|
||||
</task>
|
||||
<option name="localTasksCounter" value="14" />
|
||||
<servers />
|
||||
</component>
|
||||
<component name="TypeScriptGeneratedFilesManager">
|
||||
|
|
@ -334,6 +333,17 @@
|
|||
<MESSAGE value="增加处理函数调用链变量以及修复动态函数中传参失效的问题" />
|
||||
<option name="LAST_COMMIT_MESSAGE" value="增加处理函数调用链变量以及修复动态函数中传参失效的问题" />
|
||||
</component>
|
||||
<component name="XDebuggerManager">
|
||||
<breakpoint-manager>
|
||||
<breakpoints>
|
||||
<line-breakpoint enabled="true" suspend="THREAD" type="python-line">
|
||||
<url>file://$PROJECT_DIR$/common/validation/extractor.py</url>
|
||||
<line>10</line>
|
||||
<option name="timeStamp" value="12" />
|
||||
</line-breakpoint>
|
||||
</breakpoints>
|
||||
</breakpoint-manager>
|
||||
</component>
|
||||
<component name="com.github.evgenys91.machinet.common.dslhistory.DslHistoryState">
|
||||
<option name="historyDtoById">
|
||||
<map>
|
||||
|
|
@ -356,10 +366,15 @@
|
|||
<SUITE FILE_PATH="coverage/apitest$Unittest__test_executor_py__.coverage" NAME="Unittest (test_executor.py 内) 覆盖结果" MODIFIED="1690850721335" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/test_script" />
|
||||
<SUITE FILE_PATH="coverage/apitest$http_client.coverage" NAME="http_client 覆盖结果" MODIFIED="1690850772526" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/http_client" />
|
||||
<SUITE FILE_PATH="coverage/apitest$Unittest__test_api_py__.coverage" NAME="Unittest (test_api.py 内) 覆盖结果" MODIFIED="1689907531802" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/test_script" />
|
||||
<SUITE FILE_PATH="coverage/apitest$dependent_parameter.coverage" NAME="dependent_parameter 覆盖结果" MODIFIED="1690970103268" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/data_extraction" />
|
||||
<SUITE FILE_PATH="coverage/apitest$extractor.coverage" NAME="extractor 覆盖结果" MODIFIED="1691031544934" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/validation" />
|
||||
<SUITE FILE_PATH="coverage/apitest$dependent_parameter.coverage" NAME="dependent_parameter 覆盖结果" MODIFIED="1691028535712" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/data_extraction" />
|
||||
<SUITE FILE_PATH="coverage/apitest$assert_dict.coverage" NAME="assert_dict 覆盖结果" MODIFIED="1691034548959" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/data_extraction" />
|
||||
<SUITE FILE_PATH="coverage/apitest$re_chain.coverage" NAME="re_chain 覆盖结果" MODIFIED="1690962595251" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/debug" />
|
||||
<SUITE FILE_PATH="coverage/apitest$validator.coverage" NAME="validator 覆盖结果" MODIFIED="1691032283545" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/validation" />
|
||||
<SUITE FILE_PATH="coverage/apitest$loaders.coverage" NAME="loaders 覆盖结果" MODIFIED="1690942567666" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/validation" />
|
||||
<SUITE FILE_PATH="coverage/apitest$.coverage" NAME=" 覆盖结果" MODIFIED="1691035043685" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/test_script" />
|
||||
<SUITE FILE_PATH="coverage/apitest$requestRecord__1_.coverage" NAME="requestRecord (1) 覆盖结果" MODIFIED="1690531988570" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/debug" />
|
||||
<SUITE FILE_PATH="coverage/apitest$__init__.coverage" NAME="__init__ 覆盖结果" MODIFIED="1691028135541" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/debug" />
|
||||
<SUITE FILE_PATH="coverage/apitest$method_chain.coverage" NAME="method_chain 覆盖结果" MODIFIED="1690967090148" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/debug" />
|
||||
<SUITE FILE_PATH="coverage/apitest$action.coverage" NAME="action 覆盖结果" MODIFIED="1689907783681" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common" />
|
||||
<SUITE FILE_PATH="coverage/apitest$encryption_str.coverage" NAME="encryption_str 覆盖结果" MODIFIED="1690796164466" SOURCE_PROVIDER="com.intellij.coverage.DefaultCoverageFileProvider" RUNNER="coverage.py" COVERAGE_BY_TEST_ENABLED="true" COVERAGE_TRACING_ENABLED="false" WORKING_DIRECTORY="$PROJECT_DIR$/common/crypto" />
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -16,135 +16,133 @@ from common.utils.environments import Environments
|
|||
from common.data_extraction import logger
|
||||
|
||||
REPLACE_DICT = {
|
||||
"null": None,
|
||||
"True": True,
|
||||
"false": False
|
||||
"null": None,
|
||||
"True": True,
|
||||
"false": False
|
||||
}
|
||||
|
||||
|
||||
class DataExtractor(Environments):
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
@logger.log_decorator("提取参数出现了意想不到的错误！！")
|
||||
def substitute_data(self, response, regex=None, keys=None, deps=None, jp_dict=None):
|
||||
"""
|
||||
方法接收一个正则表达式 regex 和一个关联参数表 deps，用于从接口返回的数据中提取关联参数。
|
||||
它会从接口返回的数据中使用正则表达式 regex 和正则表达式返回结果的键列表 keys 提取数据，并将其更新到关联参数表中。
|
||||
然后，它会使用 subs_deps 和 subs_lists 方法提取更多的关联参数。最后，它将更新后的关联参数表设置为 Environments 类的静态变量，并将其返回
|
||||
Args:
|
||||
response: 被提取数据对象
|
||||
regex: 正则表达式： r'"id": (\d+), "name": "(\w+)",'
|
||||
keys: 接收正则表达式返回结果的key： ["a", "b"]
|
||||
deps: "name=data[0].name;ok=data[0].id;an=data[0].age[3].a"
|
||||
jp_dict: jsonpath 提取方式入参：{"k": "$.data", "x": "$.data[0].age[3].a"}
|
||||
Returns:
|
||||
"""
|
||||
|
||||
response = response
|
||||
if not isinstance(response, (dict, str, list)):
|
||||
logger.error(f"被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: {response}")
|
||||
return {}
|
||||
if regex and keys:
|
||||
self.substitute_regex(response, regex, keys)
|
||||
response = response if isinstance(response, (dict, list)) else json.loads(response)
|
||||
if deps:
|
||||
self.substitute_route(response, deps)
|
||||
if jp_dict:
|
||||
self.substitute_jsonpath(response, jp_dict)
|
||||
|
||||
def substitute_regex(self, response, regex, keys):
|
||||
"""
|
||||
方法用于使用正则表达式 regex 和正则表达式返回结果的键列表 keys 从接口返回的数据中提取数据，并将其更新到关联参数表中。
|
||||
Args:
|
||||
response:
|
||||
regex: 正则表达式：r'"id": (\d+), "name": "(\w+)",'
|
||||
keys:结果键列表：["a", "b"],
|
||||
Returns:
|
||||
|
||||
"""
|
||||
response = json.dumps(response) if isinstance(response, (dict, list)) else response
|
||||
match = re.search(regex, response)
|
||||
if not match:
|
||||
return {}
|
||||
groups = match.groups()
|
||||
for i, key in enumerate(keys):
|
||||
try:
|
||||
self.update_environments(key, groups[i])
|
||||
except:
|
||||
self.update_environments(key, None)
|
||||
|
||||
def substitute_route(self, response, route_str):
|
||||
"""
|
||||
想字典一样取值:name=data[0].name;ok=data[0].id;an=data[0].age
|
||||
Args:
|
||||
response:
|
||||
route_str:
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
Returns:
|
||||
@logger.log_decorator("提取参数出现了意想不到的错误！！")
|
||||
def substitute_data(self, response, regex=None, keys=None, deps=None, jp_dict=None):
|
||||
"""
|
||||
数据提取
|
||||
Args:
|
||||
response: 被提取数据对象
|
||||
regex: 正则表达式： r'"id": (\d+), "name": "(\w+)",'
|
||||
keys: 接收正则表达式返回结果的key： ["a", "b"]
|
||||
deps: "name=data[0].name;ok=data[0].id;an=data[0].age[3].a"
|
||||
jp_dict: jsonpath 提取方式入参：{"k": "$.data", "x": "$.data[0].age[3].a"}
|
||||
Returns:
|
||||
"""
|
||||
|
||||
"""
|
||||
deps_list = re.sub(f"[\r\n]+", "", route_str).split(";")
|
||||
for dep_item in deps_list:
|
||||
key, value_path = dep_item.split("=")
|
||||
value_path_parts = re.findall(r'\w+', value_path)
|
||||
temp = response
|
||||
for part in value_path_parts:
|
||||
if isinstance(temp, dict):
|
||||
temp = temp.get(part)
|
||||
elif isinstance(temp, list):
|
||||
if part.isdigit():
|
||||
index = int(part)
|
||||
if index < len(temp):
|
||||
temp = temp[index]
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
if isinstance(temp, (dict, list)):
|
||||
continue
|
||||
else:
|
||||
break
|
||||
if temp is not None:
|
||||
self.update_environments(key, temp)
|
||||
|
||||
def substitute_jsonpath(self, response, json_path_dict):
|
||||
"""
|
||||
jsonpath 提取参数
|
||||
Args:
|
||||
response: 响应结果
|
||||
json_path_dict: {"k": "$.data", "x": "$.data[0].age[3].a"}
|
||||
response = response
|
||||
if not isinstance(response, (dict, str, list)):
|
||||
logger.error(f"| 被提取对象非字典、非字符串、非列表，不执行jsonpath提取，被提取对象: {response}")
|
||||
return {}
|
||||
if regex and keys:
|
||||
self.substitute_regex(response, regex, keys)
|
||||
response = response if isinstance(response, (dict, list)) else json.loads(response)
|
||||
if deps:
|
||||
self.substitute_route(response, deps)
|
||||
if jp_dict:
|
||||
self.substitute_jsonpath(response, jp_dict)
|
||||
|
||||
Returns: 字符串或者list
|
||||
def substitute_regex(self, response, regex, keys):
|
||||
"""
|
||||
使用正则表达式
|
||||
Args:
|
||||
response:
|
||||
regex: 正则表达式：r'"id": (\d+), "name": "(\w+)",'
|
||||
keys:结果键列表：["a", "b"],
|
||||
Returns:
|
||||
|
||||
"""
|
||||
json_path_dict = json_path_dict if isinstance(json_path_dict, dict) else json.loads(json_path_dict)
|
||||
for key, expression in json_path_dict.items():
|
||||
try:
|
||||
parsed_expression = parse(expression)
|
||||
data = response
|
||||
# 使用解析器对象进行匹配和提取
|
||||
match = parsed_expression.find(data)
|
||||
result = [m.value for m in match]
|
||||
self.update_environments(key, result[0]) if len(result) == 1 else self.update_environments(key,
|
||||
result)
|
||||
except Exception as e:
|
||||
logger.error(f"jsonpath表达式错误'{expression}': {e}")
|
||||
"""
|
||||
response = json.dumps(response) if isinstance(response, (dict, list)) else response
|
||||
match = re.search(regex, response)
|
||||
if not match:
|
||||
return {}
|
||||
groups = match.groups()
|
||||
for i, key in enumerate(keys):
|
||||
try:
|
||||
self.update_environments(key, groups[i])
|
||||
except:
|
||||
self.update_environments(key, None)
|
||||
|
||||
def substitute_route(self, response, route_str):
|
||||
"""
|
||||
字典取值
|
||||
Args:
|
||||
response:
|
||||
route_str:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
deps_list = re.sub(f"[\r\n]+", "", route_str).split(";")
|
||||
for dep_item in deps_list:
|
||||
key, value_path = dep_item.split("=")
|
||||
value_path_parts = re.findall(r'\w+', value_path)
|
||||
temp = response
|
||||
for part in value_path_parts:
|
||||
if isinstance(temp, dict):
|
||||
temp = temp.get(part)
|
||||
elif isinstance(temp, list):
|
||||
if part.isdigit():
|
||||
index = int(part)
|
||||
if index < len(temp):
|
||||
temp = temp[index]
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
else:
|
||||
temp = None
|
||||
break
|
||||
if isinstance(temp, (dict, list)):
|
||||
continue
|
||||
else:
|
||||
break
|
||||
if temp is not None:
|
||||
self.update_environments(key, temp)
|
||||
|
||||
def substitute_jsonpath(self, response, json_path_dict):
|
||||
"""
|
||||
jsonpath取值
|
||||
Args:
|
||||
response:
|
||||
json_path_dict: {"k": "$.data", "x": "$.data[0].age[3].a"}
|
||||
|
||||
Returns: 字符串或者list
|
||||
|
||||
"""
|
||||
json_path_dict = json_path_dict if isinstance(json_path_dict, dict) else json.loads(json_path_dict)
|
||||
for key, expression in json_path_dict.items():
|
||||
try:
|
||||
parsed_expression = parse(expression)
|
||||
data = response
|
||||
# 使用解析器对象进行匹配和提取
|
||||
match = parsed_expression.find(data)
|
||||
result = [m.value for m in match]
|
||||
self.update_environments(key, result[0]) if len(result) == 1 else self.update_environments(key,
|
||||
result)
|
||||
except Exception as e:
|
||||
logger.error(f"| jsonpath表达式错误'{expression}': {e}")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# 测试subs函数
|
||||
res = '{"code": 1,"data": [{"id": 1, "name": "Alice", "age": [20, 21, 22, {"a": "b"}]}]}'
|
||||
lists = {"k": "$..code", "x": "$.data[0].age[3].a"}
|
||||
dep_str = "name=data[0].name;ok=data[0].id;an=data[0].age"
|
||||
regex_str = r'"id": (\d+), "name": "(\w+)",'
|
||||
regex_key = ["a", "b"]
|
||||
t = DataExtractor()
|
||||
t.substitute_data(res, regex=regex_str, keys=regex_key, deps=dep_str, jp_dict=lists)
|
||||
print("-------->res:", t.get_environments())
|
||||
# 测试subs函数
|
||||
res = '{"code": 1,"data": [{"id": 1, "name": "Alice", "age": [20, 21, 22, {"a": "b"}]}]}'
|
||||
lists = {"k": "$..code", "x": "$.data[0].age[3].a"}
|
||||
dep_str = "name=data[0].name;ok=data[0].id;an=data[0].age"
|
||||
regex_str = r'"id": (\d+), "name": "(\w+)",'
|
||||
regex_key = ["a", "b"]
|
||||
t = DataExtractor()
|
||||
t.substitute_data(res, regex=regex_str, keys=regex_key, deps=dep_str, jp_dict=lists)
|
||||
print("-------->res:", t.get_environments())
|
||||
|
|
|
|||
|
|
@ -14,10 +14,7 @@ from common.data_extraction.data_extractor import DataExtractor
|
|||
|
||||
|
||||
class DependentParameter(DataExtractor):
|
||||
"""
|
||||
该类用于替换接口参数。它会从字符串中寻找需要替换的参数，并将其替换为关联参数表中对应的值。
|
||||
然后，它将替换后的字符串转化为字典并返回。如果找不到需要替换的参数，则直接返回原始字符串。
|
||||
"""
|
||||
"""数据更换"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
|
|
|||
|
|
@ -35,7 +35,7 @@ class MysqlClient:
|
|||
self.conn = self.pool.connection()
|
||||
self.cursor = self.conn.cursor(DictCursor)
|
||||
except Exception as e:
|
||||
logger.error(f"数据库链接失败: {e}")
|
||||
logger.error(f"| 数据库链接失败: {e}")
|
||||
# raise
|
||||
|
||||
# @logger.log_decorator()
|
||||
|
|
@ -72,9 +72,9 @@ class MysqlClient:
|
|||
for method, sql_data in sql.items():
|
||||
execute_method = getattr(self, f"_execute_{method}", None)
|
||||
if not execute_method:
|
||||
logger.error("sql字典集编写格式不符合规范")
|
||||
raise ValueError("Invalid SQL method")
|
||||
logger.info(f"执行 sql 语句集: {sql_data}")
|
||||
logger.error("| sql字典集编写格式不符合规范")
|
||||
raise ValueError("| Invalid SQL method")
|
||||
logger.info(f"| 执行 sql 语句集: {sql_data}")
|
||||
execute_method(sql_data)
|
||||
|
||||
self.cursor.close()
|
||||
|
|
@ -83,7 +83,7 @@ class MysqlClient:
|
|||
return self.result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"数据库操作异常: {e}")
|
||||
logger.error(f"| 数据库操作异常: {e}")
|
||||
raise
|
||||
|
||||
def _execute_write(self, sql_data):
|
||||
|
|
@ -94,7 +94,7 @@ class MysqlClient:
|
|||
try:
|
||||
self.cursor.execute(str(sql_))
|
||||
except Exception as err:
|
||||
logger.error(f"执行 SQL 异常: {sql_}")
|
||||
logger.error(f"| 执行 SQL 异常: {sql_}")
|
||||
raise err
|
||||
self.cursor.connection.commit()
|
||||
|
||||
|
|
@ -118,7 +118,7 @@ class MysqlClient:
|
|||
self.cursor.execute(sql_)
|
||||
self.result[sql_name] = self.cursor.fetchall()
|
||||
except Exception as err:
|
||||
logger.error(f"查询异常 sql: {sql_}")
|
||||
logger.error(f"| 查询异常 sql: {sql_}")
|
||||
raise err
|
||||
|
||||
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -15,12 +15,12 @@ from common.file_handling import logger
|
|||
|
||||
@singleton
|
||||
class DoExcel:
|
||||
|
||||
|
||||
def __init__(self, file_name):
|
||||
self.file_name = file_name
|
||||
self.wb = load_workbook(self.file_name)
|
||||
self.init_sheet = self.wb["init"]
|
||||
|
||||
|
||||
# def __enter__(self):
|
||||
# self.wb = load_workbook(self.file_name)
|
||||
# self.init_sheet = self.wb['init']
|
||||
|
|
@ -29,7 +29,7 @@ class DoExcel:
|
|||
# def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
# self.wb.save(self.file_name)
|
||||
# self.wb.close()
|
||||
|
||||
|
||||
def do_excel_yield(self):
|
||||
"""
|
||||
读取excel数据
|
||||
|
|
@ -50,7 +50,7 @@ class DoExcel:
|
|||
sub_data[first_header[k - 1]] = sheet.cell(i, k).value
|
||||
sub_data["sheet"] = sheet_name
|
||||
yield sub_data
|
||||
|
||||
|
||||
@logger.log_decorator()
|
||||
def write_back(self, sheet_name, i, **kwargs):
|
||||
"""
|
||||
|
|
@ -71,7 +71,7 @@ class DoExcel:
|
|||
sheet.cell(i + 1, 25).value = test_result
|
||||
sheet.cell(i + 1, 26).value = assert_log
|
||||
self.wb.save(self.file_name)
|
||||
|
||||
|
||||
@logger.log_decorator()
|
||||
def clear_date(self):
|
||||
"""
|
||||
|
|
@ -80,7 +80,7 @@ class DoExcel:
|
|||
|
||||
"""
|
||||
sheets = eval(self.get_excel_init().get("sheets"))
|
||||
|
||||
|
||||
for sheet_name in sheets:
|
||||
sheet = self.wb[sheet_name]
|
||||
max_row = sheet.max_row # 获取最大行
|
||||
|
|
@ -90,7 +90,7 @@ class DoExcel:
|
|||
sheet.cell(i, 26).value = ""
|
||||
self.wb.save(self.file_name)
|
||||
return f"清空指定 {sheets} 中的单元格成功"
|
||||
|
||||
|
||||
@logger.log_decorator()
|
||||
def get_excel_init(self):
|
||||
"""
|
||||
|
|
@ -109,7 +109,7 @@ class DoExcel:
|
|||
if init.get("run").upper() == "YES":
|
||||
break
|
||||
return init
|
||||
|
||||
|
||||
def get_excel_init_and_cases(self):
|
||||
"""
|
||||
|
||||
|
|
@ -122,18 +122,20 @@ class DoExcel:
|
|||
init_data = self.get_excel_init()
|
||||
databases = init_data.get('databases')
|
||||
initialize_data = eval(init_data.get("initialize_data"))
|
||||
host = init_data.get('host', "") + init_data.get("path", "")
|
||||
host = init_data.get('host', "")
|
||||
path = init_data.get("path", "")
|
||||
host_path = host if host is not None else "" + path if path is not None else ""
|
||||
except Exception as e:
|
||||
raise e
|
||||
return test_case, databases, initialize_data, host
|
||||
|
||||
return test_case, databases, initialize_data, host_path
|
||||
|
||||
def close_excel(self):
|
||||
self.wb.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from config import Config
|
||||
|
||||
|
||||
file_n = Config.test_case
|
||||
excel = DoExcel(file_n)
|
||||
# excel.get_excel_init()
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -17,13 +17,13 @@ from common.utils.decorators import request_retry_on_exception
|
|||
|
||||
class Pyt(LoadModulesFromFolder):
|
||||
session = requests.Session()
|
||||
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.request = None
|
||||
self.response = None
|
||||
self.response_json = None
|
||||
|
||||
|
||||
@request_retry_on_exception()
|
||||
def http_client(self, host, url, method, **kwargs):
|
||||
"""
|
||||
|
|
@ -36,20 +36,21 @@ class Pyt(LoadModulesFromFolder):
|
|||
"""
|
||||
# 关闭 https 警告信息
|
||||
urllib3.disable_warnings()
|
||||
|
||||
self.request = None
|
||||
self.response = None
|
||||
if not url:
|
||||
raise ValueError("URL cannot be None")
|
||||
__url = f'{host}{url}' if not re.match(r"https?", url) else url
|
||||
|
||||
|
||||
# 增加兼容
|
||||
# 处理 headers 参数为字符串类型的情况
|
||||
if 'headers' in kwargs and isinstance(kwargs['headers'], str):
|
||||
kwargs['headers'] = json.loads(kwargs['headers'])
|
||||
|
||||
|
||||
# 处理 json 参数为字符串类型的情况
|
||||
if 'json' in kwargs and isinstance(kwargs['json'], str):
|
||||
kwargs['json'] = json.loads(kwargs['json'])
|
||||
|
||||
|
||||
# 处理 files 参数的情况
|
||||
fs = []
|
||||
if 'files' in kwargs:
|
||||
|
|
@ -67,7 +68,7 @@ class Pyt(LoadModulesFromFolder):
|
|||
('file', (f'{file_path}', f, file_type))
|
||||
)
|
||||
kwargs['files'] = files
|
||||
|
||||
|
||||
# 发送请求
|
||||
self.request = requests.Request(method, __url, **kwargs)
|
||||
self.response = self.session.send(self.request.prepare(), timeout=30, verify=True)
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -11,111 +11,111 @@ from time import perf_counter
|
|||
|
||||
from loguru import logger
|
||||
|
||||
from config import Config
|
||||
from common.utils.decorators import singleton
|
||||
from config import Config
|
||||
|
||||
LOG_DIR = Config.log_path
|
||||
|
||||
|
||||
@singleton
|
||||
class MyLogger:
|
||||
"""
|
||||
根据时间、文件大小切割日志
|
||||
"""
|
||||
|
||||
def __init__(self, log_dir=LOG_DIR, max_size=20, retention='7 days'):
|
||||
self.log_dir = log_dir
|
||||
self.max_size = max_size
|
||||
self.retention = retention
|
||||
self.logger = self.configure_logger()
|
||||
|
||||
def configure_logger(self):
|
||||
"""
|
||||
Returns:
|
||||
"""
|
||||
# 创建日志目录
|
||||
os.makedirs(self.log_dir, exist_ok=True)
|
||||
|
||||
shared_config = {
|
||||
"level": "ERROR",
|
||||
"enqueue": True,
|
||||
"backtrace": False,
|
||||
"format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}",
|
||||
}
|
||||
|
||||
# 添加按照日期和大小切割的文件 handler
|
||||
logger.add(
|
||||
sink=f"{self.log_dir}/{{time:YYYY-MM-DD}}.log",
|
||||
rotation=f"{self.max_size} MB",
|
||||
retention=self.retention,
|
||||
**shared_config
|
||||
)
|
||||
|
||||
# 配置按照等级划分的文件 handler 和控制台输出
|
||||
logger.add(sink=self.get_log_path, **shared_config)
|
||||
|
||||
return logger
|
||||
|
||||
def get_log_path(self, message: str) -> str:
|
||||
"""
|
||||
根据等级返回日志路径
|
||||
Args:
|
||||
message:
|
||||
Returns:
|
||||
"""
|
||||
log_level = message.record["level"].name.lower()
|
||||
log_file = f"{log_level}.log"
|
||||
log_path = os.path.join(self.log_dir, log_file)
|
||||
|
||||
return log_path
|
||||
|
||||
def __getattr__(self, level: str):
|
||||
return getattr(self.logger, level)
|
||||
|
||||
def log_decorator(self, msg="快看，异常了，别唧唧哇哇，快排查！！"):
|
||||
"""
|
||||
日志装饰器，记录函数的名称、参数、返回值、运行时间和异常信息
|
||||
Args:
|
||||
logger: 日志记录器对象
|
||||
Returns:
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
# self.logger.info(f'-----------分割线-----------')
|
||||
self.logger.info(f'| 调用函数： {func.__name__} | args: {args} kwargs:{kwargs}')
|
||||
start = perf_counter() # 开始时间
|
||||
try:
|
||||
result = func(*args, **kwargs)
|
||||
end = perf_counter() # 结束时间
|
||||
duration = end - start
|
||||
self.logger.info(f"| 结束调用函数： {func.__name__}, duration：{duration:4f}s")
|
||||
return result
|
||||
except Exception as e:
|
||||
self.logger.error(f"| called {func.__name__} | error: {msg}: {e}")
|
||||
|
||||
# self.logger.info(f"-----------分割线-----------")
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
"""
|
||||
根据时间、文件大小切割日志
|
||||
"""
|
||||
|
||||
def __init__(self, log_dir=LOG_DIR, max_size=20, retention='7 days'):
|
||||
self.log_dir = log_dir
|
||||
self.max_size = max_size
|
||||
self.retention = retention
|
||||
self.logger = self.configure_logger()
|
||||
|
||||
def configure_logger(self):
|
||||
"""
|
||||
Returns:
|
||||
"""
|
||||
# 创建日志目录
|
||||
os.makedirs(self.log_dir, exist_ok=True)
|
||||
|
||||
shared_config = {
|
||||
"level": "ERROR",
|
||||
"enqueue": True,
|
||||
"backtrace": False,
|
||||
"format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}",
|
||||
}
|
||||
|
||||
# 添加按照日期和大小切割的文件 handler
|
||||
logger.add(
|
||||
sink=f"{self.log_dir}/{{time:YYYY-MM-DD}}.log",
|
||||
rotation=f"{self.max_size} MB",
|
||||
retention=self.retention,
|
||||
**shared_config
|
||||
)
|
||||
|
||||
# 配置按照等级划分的文件 handler 和控制台输出
|
||||
logger.add(sink=self.get_log_path, **shared_config)
|
||||
|
||||
return logger
|
||||
|
||||
def get_log_path(self, message: str) -> str:
|
||||
"""
|
||||
根据等级返回日志路径
|
||||
Args:
|
||||
message:
|
||||
Returns:
|
||||
"""
|
||||
log_level = message.record["level"].name.lower()
|
||||
log_file = f"{log_level}.log"
|
||||
log_path = os.path.join(self.log_dir, log_file)
|
||||
|
||||
return log_path
|
||||
|
||||
def __getattr__(self, level: str):
|
||||
return getattr(self.logger, level)
|
||||
|
||||
def log_decorator(self, msg="快看，异常了，别唧唧哇哇，快排查！！"):
|
||||
"""
|
||||
日志装饰器，记录函数的名称、参数、返回值、运行时间和异常信息
|
||||
Args:
|
||||
logger: 日志记录器对象
|
||||
Returns:
|
||||
"""
|
||||
|
||||
def decorator(func):
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
# self.logger.info(f'-----------分割线-----------')
|
||||
# self.logger.info(f'| 调用函数： {func.__name__} | args: {args} kwargs:{kwargs}')
|
||||
start = perf_counter() # 开始时间
|
||||
try:
|
||||
result = func(*args, **kwargs)
|
||||
end = perf_counter() # 结束时间
|
||||
duration = end - start
|
||||
# self.logger.info(f"| 结束调用函数： {func.__name__}, duration：{duration:4f}s")
|
||||
return result
|
||||
except Exception as e:
|
||||
self.logger.error(f"| called {func.__name__} | error: {msg}: {e}")
|
||||
|
||||
# self.logger.info(f"-----------分割线-----------")
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
log = MyLogger()
|
||||
|
||||
|
||||
@log.log_decorator("勇哥也不知道错在哪里")
|
||||
def test_zero_division_error(a, b):
|
||||
return a / b
|
||||
|
||||
|
||||
for i in range(1000):
|
||||
log.error('错误信息')
|
||||
log.critical('严重错误信息')
|
||||
test_zero_division_error(1, 0)
|
||||
log.debug('调试信息')
|
||||
log.info('普通信息')
|
||||
log.success('成功信息')
|
||||
log.warning('警告信息')
|
||||
log = MyLogger()
|
||||
|
||||
|
||||
@log.log_decorator("勇哥也不知道错在哪里")
|
||||
def test_zero_division_error(a, b):
|
||||
return a / b
|
||||
|
||||
|
||||
for i in range(1000):
|
||||
log.error('错误信息')
|
||||
log.critical('严重错误信息')
|
||||
test_zero_division_error(1, 0)
|
||||
log.debug('调试信息')
|
||||
log.info('普通信息')
|
||||
log.success('成功信息')
|
||||
log.warning('警告信息')
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -41,7 +41,9 @@ class Action(Extractor, LoadScript, Validator, MysqlClient):
|
|||
exec(compiled, {"pm": self})
|
||||
except SyntaxError as e:
|
||||
ExecuteDynamiCodeError(code, e)
|
||||
except ExecuteDynamiCodeError as e:
|
||||
except TypeError as e:
|
||||
ExecuteDynamiCodeError(code, e)
|
||||
except Exception as e:
|
||||
ExecuteDynamiCodeError(code, e)
|
||||
|
||||
return self.variables
|
||||
|
|
@ -54,22 +56,43 @@ class Action(Extractor, LoadScript, Validator, MysqlClient):
|
|||
def variables(self, item):
|
||||
self.__variables = item
|
||||
|
||||
# def update_variables(self, key, value):
|
||||
# self.__variables[f"{{{{{key}}}}}"] = value
|
||||
|
||||
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)
|
||||
# 提取请求参数信息
|
||||
if extract_request_data is not None and request_data is not None:
|
||||
self.substitute_data(request_data, jp_dict=extract_request_data)
|
||||
return headers, request_data
|
||||
|
||||
def send_request(self, host, url, method, teardown_script, **kwargs):
|
||||
self.http_client(host, url, method, **kwargs)
|
||||
self.update_environments("responseTime", self.response.elapsed.total_seconds() * 1000) # 存响应时间
|
||||
self.update_environments("responseStatusCode", self.response.status_code)
|
||||
self.update_environments("responseTime", round(self.response.elapsed.total_seconds() * 1000, 2))
|
||||
self.execute_dynamic_code(self.response, teardown_script)
|
||||
|
||||
def analysis_response(self, sheet, iid, name, desc, regex, keys, deps, jp_dict):
|
||||
try:
|
||||
self.substitute_data(self.response_json, regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
|
||||
except Exception as err:
|
||||
self.logger.error(f"| 分析响应失败：{sheet}_{iid}_{name}_{desc}"
|
||||
f"\nregex={regex};"
|
||||
f" \nkeys={keys};"
|
||||
f"\ndeps={deps};"
|
||||
f"\njp_dict={jp_dict}"
|
||||
f"\n{err}")
|
||||
|
||||
def execute_validation(self, excel, sheet, iid, name, desc, expected):
|
||||
try:
|
||||
self.run_validate(expected, self.response_json)
|
||||
result = "PASS"
|
||||
except Exception as e:
|
||||
result = "FAIL"
|
||||
self.logger.error(f"| exception case:**{sheet}_{iid}_{name}_{desc}**\n{e}")
|
||||
raise AssertionFailedError(self.assertions)
|
||||
finally:
|
||||
print(f'| Assertion Result-->{self.assertions}')
|
||||
response = self.response.text if self.response is not None else str(self.response)
|
||||
# excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
|
||||
# assert_log=str(self.assertions))
|
||||
|
||||
@staticmethod
|
||||
def base_info(item):
|
||||
"""
|
||||
|
|
@ -139,7 +162,7 @@ class Action(Extractor, LoadScript, Validator, MysqlClient):
|
|||
if sleep_time:
|
||||
try:
|
||||
time.sleep(sleep_time)
|
||||
except MyBaseException as e:
|
||||
except Exception as e:
|
||||
raise MyBaseException(f"暂停时间必须是数字!")
|
||||
|
||||
def exc_sql(self, item):
|
||||
|
|
@ -154,9 +177,9 @@ class Action(Extractor, LoadScript, Validator, MysqlClient):
|
|||
if execute_sql_results and sql_params_dict:
|
||||
try:
|
||||
self.substitute_data(execute_sql_results, jp_dict=sql_params_dict)
|
||||
except ParameterExtractionError as e:
|
||||
except Exception as e:
|
||||
ParameterExtractionError(sql_params_dict, str(e))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print(Action.__mro__)
|
||||
print(Action())
|
||||
|
|
|
|||
|
|
@ -37,20 +37,21 @@ def request_retry_on_exception(retries=2, delay=1.5):
|
|||
nonlocal e
|
||||
for i in range(retries):
|
||||
try:
|
||||
print(f"第{i + 1}次发送请求的参数： {kwargs}")
|
||||
print(f"| 第{i + 1}次发送请求的参数： {kwargs}")
|
||||
response = func(*args, **kwargs)
|
||||
print(f"请求地址 --> {response.request.url}")
|
||||
print(f"请求头 --> {response.request.headers}")
|
||||
print(f"请求 body --> {response.request.body}")
|
||||
print(f"接口状态--> {response.status_code}")
|
||||
print(f"接口耗时--> {response.elapsed}")
|
||||
print(f"接口响应--> {response.text}")
|
||||
print(f"| 请求地址 --> {response.request.url}")
|
||||
print(f"| 请求头 --> {response.request.headers}")
|
||||
print(f"| 请求 body --> {response.request.body}")
|
||||
print(f"| 接口状态--> {response.status_code}")
|
||||
print(f"| 接口耗时--> {response.elapsed}")
|
||||
print(f"| 接口响应--> {response.text}")
|
||||
|
||||
except Exception as error:
|
||||
e = error
|
||||
time.sleep(delay)
|
||||
else:
|
||||
return response
|
||||
raise Exception(f"请求重试**{retries}**次失败，请检查！！{e}")
|
||||
raise Exception(f"| 请求重试**{retries}**次失败，请检查！！{e}")
|
||||
|
||||
return wrapper
|
||||
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@
|
|||
class MyBaseException(Exception):
|
||||
def __init__(self, msg):
|
||||
self.msg = msg
|
||||
|
||||
|
||||
def __str__(self):
|
||||
return self.msg
|
||||
|
||||
|
|
@ -20,7 +20,7 @@ class MyBaseException(Exception):
|
|||
class RequestSendingError(MyBaseException):
|
||||
"""请求异常"""
|
||||
ERROR_CODE = 1001
|
||||
|
||||
|
||||
def __init__(self, url, reason):
|
||||
msg = f"请求异常：URL={url}, 原因={reason}"
|
||||
super().__init__(msg)
|
||||
|
|
@ -29,7 +29,7 @@ class RequestSendingError(MyBaseException):
|
|||
class DatabaseExceptionError(MyBaseException):
|
||||
"""数据库异常"""
|
||||
ERROR_CODE = 1002
|
||||
|
||||
|
||||
def __init__(self, operation, reason):
|
||||
msg = f"数据库异常：操作={operation}, 原因={reason}"
|
||||
super().__init__(msg)
|
||||
|
|
@ -38,7 +38,7 @@ class DatabaseExceptionError(MyBaseException):
|
|||
class ParameterExtractionError(MyBaseException):
|
||||
"""参数提取异常"""
|
||||
ERROR_CODE = 1003
|
||||
|
||||
|
||||
def __init__(self, parameter_path, reason):
|
||||
msg = f"参数提取异常：参数路径={parameter_path}, 原因={reason}"
|
||||
super().__init__(msg)
|
||||
|
|
@ -47,7 +47,7 @@ class ParameterExtractionError(MyBaseException):
|
|||
class ParameterReplacementError(MyBaseException):
|
||||
"""参数替换异常"""
|
||||
ERROR_CODE = 1004
|
||||
|
||||
|
||||
def __init__(self, parameter_name, reason):
|
||||
msg = f"参数替换异常：参数名称={parameter_name}, 原因={reason}"
|
||||
super().__init__(msg)
|
||||
|
|
@ -56,16 +56,18 @@ class ParameterReplacementError(MyBaseException):
|
|||
class AssertionFailedError(MyBaseException):
|
||||
"""断言异常"""
|
||||
ERROR_CODE = 1005
|
||||
|
||||
def __init__(self, assertion_name, actual_value, expected_value):
|
||||
msg = f"断言失败：断言名称={assertion_name}, 实际值={actual_value}, 期望值={expected_value}"
|
||||
|
||||
def __init__(self, assertion):
|
||||
msg = f"断言失败：{assertion}"
|
||||
print(msg)
|
||||
super().__init__(msg)
|
||||
|
||||
|
||||
class ExecuteDynamiCodeError(MyBaseException):
|
||||
"""执行动态代码异常"""
|
||||
ERROR_CODE = 1006
|
||||
|
||||
|
||||
def __init__(self, code, reason):
|
||||
msg = f"执行动态代码异常：动态代码={code}, 原因={reason}"
|
||||
print(msg)
|
||||
super().__init__(msg)
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -4,7 +4,7 @@
|
|||
# Name: comparator_dict.py
|
||||
# Description: 比较器名词释义
|
||||
# Author: kira
|
||||
# EMAIL: 2419352654@qq.com
|
||||
# EMAIL: 262667641@qq.com
|
||||
# Date: 2020/10/29 16:51
|
||||
# -------------------------------------------------------------------------------
|
||||
|
||||
|
|
|
|||
|
|
@ -12,274 +12,274 @@ import re
|
|||
|
||||
|
||||
def eq(actual_value, expect_value):
|
||||
"""
|
||||
实际值与期望值相等
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value == expect_value
|
||||
"""
|
||||
实际值与期望值相等
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value == expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def lt(actual_value, expect_value):
|
||||
"""
|
||||
实际值小于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value < expect_value
|
||||
"""
|
||||
实际值小于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value < expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def lte(actual_value, expect_value):
|
||||
"""
|
||||
实际值小于或等于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value <= expect_value
|
||||
"""
|
||||
实际值小于或等于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value <= expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def gt(actual_value, expect_value):
|
||||
"""
|
||||
实际值大于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value > expect_value
|
||||
"""
|
||||
实际值大于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value > expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def gte(actual_value, expect_value):
|
||||
"""
|
||||
实际值大于或等于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value >= expect_value
|
||||
"""
|
||||
实际值大于或等于期望值
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value >= expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def neq(actual_value, expect_value):
|
||||
"""
|
||||
实际值与期望值不相等
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value != expect_value
|
||||
"""
|
||||
实际值与期望值不相等
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert actual_value != expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def str_eq(actual_value, expect_value):
|
||||
"""
|
||||
字符串实际值与期望值相同
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value) == str(expect_value)
|
||||
"""
|
||||
字符串实际值与期望值相同
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value) == str(expect_value), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def length_eq(actual_value, expect_value):
|
||||
"""
|
||||
实际值的长度等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,))
|
||||
assert len(actual_value) == expect_value
|
||||
"""
|
||||
实际值的长度等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert len(actual_value) == expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def length_gt(actual_value, expect_value):
|
||||
"""
|
||||
实际值的长度大于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,))
|
||||
assert len(actual_value) > expect_value
|
||||
"""
|
||||
实际值的长度大于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert len(actual_value) > expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def length_gte(actual_value, expect_value):
|
||||
"""
|
||||
实际值的长度大于或等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,))
|
||||
assert len(actual_value) >= expect_value
|
||||
"""
|
||||
实际值的长度大于或等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert len(actual_value) >= expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def length_lt(actual_value, expect_value):
|
||||
"""
|
||||
实际值的长度小于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,))
|
||||
assert len(actual_value) < expect_value
|
||||
"""
|
||||
实际值的长度小于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert len(actual_value) < expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def length_lte(actual_value, expect_value):
|
||||
"""
|
||||
实际值的长度小于或等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,))
|
||||
assert len(actual_value) <= expect_value
|
||||
"""
|
||||
实际值的长度小于或等于期望长度
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (int,)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert len(actual_value) <= expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def contains(actual_value, expect_value):
|
||||
"""
|
||||
期望值包含在实际值中
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(actual_value, (list, tuple, dict, str, bytes))
|
||||
assert expect_value in actual_value
|
||||
"""
|
||||
期望值包含在实际值中
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(actual_value, (list, tuple, dict, str, bytes)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert expect_value in actual_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def contained_by(actual_value, expect_value):
|
||||
"""
|
||||
实际值被包含在期望值中
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (list, tuple, dict, str, bytes))
|
||||
assert actual_value in expect_value
|
||||
"""
|
||||
实际值被包含在期望值中
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert isinstance(expect_value, (list, tuple, dict, str, bytes)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
assert actual_value in expect_value, f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def type_match(actual_value, expect_value):
|
||||
"""
|
||||
实际值的类型与期望值的类型相匹配
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
|
||||
def get_type(name):
|
||||
if isinstance(name, type):
|
||||
return name
|
||||
elif isinstance(name, (str, bytes)):
|
||||
try:
|
||||
return __builtins__[name]
|
||||
except KeyError:
|
||||
raise ValueError(name)
|
||||
else:
|
||||
raise ValueError(name)
|
||||
|
||||
assert isinstance(actual_value, get_type(expect_value))
|
||||
"""
|
||||
实际值的类型与期望值的类型相匹配
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
|
||||
def get_type(name):
|
||||
if isinstance(name, type):
|
||||
return name
|
||||
elif isinstance(name, (str, bytes)):
|
||||
try:
|
||||
return __builtins__[name]
|
||||
except KeyError:
|
||||
raise ValueError(name)
|
||||
else:
|
||||
raise ValueError(name)
|
||||
|
||||
assert isinstance(actual_value, get_type(expect_value)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def regex_match(actual_value, expect_value):
|
||||
"""
|
||||
正则匹配(从字符串的起始位置匹配)
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if not isinstance(actual_value, str):
|
||||
actual_value = json.dumps(actual_value, ensure_ascii=False)
|
||||
if not isinstance(expect_value, str):
|
||||
expect_value = json.dumps(expect_value, ensure_ascii=False)
|
||||
assert re.match(expect_value, actual_value)
|
||||
"""
|
||||
正则匹配(从字符串的起始位置匹配)
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if not isinstance(actual_value, str):
|
||||
actual_value = json.dumps(actual_value, ensure_ascii=False)
|
||||
if not isinstance(expect_value, str):
|
||||
expect_value = json.dumps(expect_value, ensure_ascii=False)
|
||||
assert re.match(expect_value, actual_value), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def regex_search(actual_value, expect_value):
|
||||
"""
|
||||
正则匹配(从字符串的任意位置匹配)
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if not isinstance(actual_value, str):
|
||||
actual_value = json.dumps(actual_value, ensure_ascii=False)
|
||||
if not isinstance(expect_value, str):
|
||||
expect_value = json.dumps(expect_value, ensure_ascii=False)
|
||||
assert re.search(expect_value, actual_value)
|
||||
"""
|
||||
正则匹配(从字符串的任意位置匹配)
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if not isinstance(actual_value, str):
|
||||
actual_value = json.dumps(actual_value, ensure_ascii=False)
|
||||
if not isinstance(expect_value, str):
|
||||
expect_value = json.dumps(expect_value, ensure_ascii=False)
|
||||
assert re.search(expect_value, actual_value), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def startswith(actual_value, expect_value):
|
||||
"""
|
||||
实际值是以期望值开始
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value).startswith(str(expect_value))
|
||||
"""
|
||||
实际值是以期望值开始
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value).startswith(str(expect_value)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
||||
|
||||
def endswith(actual_value, expect_value):
|
||||
"""
|
||||
实际值是以期望值结束
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value).endswith(str(expect_value))
|
||||
"""
|
||||
实际值是以期望值结束
|
||||
Args:
|
||||
actual_value: 实际值
|
||||
expect_value: 期望值
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
assert str(actual_value).endswith(str(expect_value)), f"预期结果：{actual_value},实际结果：{expect_value}"
|
||||
|
|
|
|||
|
|
@ -73,8 +73,8 @@ class Extractor:
|
|||
Returns:
|
||||
|
||||
"""
|
||||
logger.debug(f'正在执行数据提取：提取数据源内容：{resp_obj},{type(resp_obj)}')
|
||||
logger.debug('正在执行数据提取：提取表达式：{expr}'.format(expr=expr))
|
||||
# logger.debug(f'正在执行数据提取：提取数据源内容：{resp_obj},{type(resp_obj)}')
|
||||
# logger.debug('正在执行数据提取：提取表达式：{expr}'.format(expr=expr))
|
||||
try:
|
||||
result = jsonpath.jsonpath(resp_obj if isinstance(resp_obj, (dict, list)) else json.dumps(resp_obj), expr)
|
||||
except Exception as e:
|
||||
|
|
@ -86,7 +86,7 @@ class Extractor:
|
|||
elif isinstance(result, list):
|
||||
if len(result) == 1:
|
||||
result = result[0]
|
||||
logger.info(f'提取成功，输出结果，提取表达式：{expr}，提取结果：{result}')
|
||||
# logger.info(f'提取成功，输出结果，提取表达式：{expr}，提取结果：{result}')
|
||||
return result
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -27,17 +27,17 @@ class LoadModulesFromFolder(DependentParameter):
|
|||
|
||||
if isinstance(folder_or_mnodule, str):
|
||||
folder_path = folder_or_mnodule
|
||||
if not os.path.exists(folder_path): # 检查文件夹路径是否存在
|
||||
if not os.path.exists(folder_path):
|
||||
raise ValueError("Folder path does not exist.")
|
||||
|
||||
for file_name in os.listdir(folder_path): # 遍历指定文件夹下的所有文件
|
||||
for file_name in os.listdir(folder_path):
|
||||
module_name, ext = os.path.splitext(file_name)
|
||||
if ext == '.py' and module_name != '__init__':
|
||||
module_path = os.path.join(folder_path, file_name) # 获取模块文件的完整路径
|
||||
module_path = os.path.join(folder_path, file_name)
|
||||
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
try:
|
||||
spec.loader.exec_module(module) # 加载模块文件并执行其中的代码，将函数定义添加到 module 对象中
|
||||
spec.loader.exec_module(module)
|
||||
except:
|
||||
continue
|
||||
for name, obj in vars(module).items():
|
||||
|
|
|
|||
|
|
@ -15,56 +15,56 @@ from common.validation import logger
|
|||
|
||||
|
||||
class Loaders(Pyt):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
@logger.log_decorator()
|
||||
def load_built_in_functions(self, model):
|
||||
"""
|
||||
加载bif_functions包中的内建方法
|
||||
Returns:
|
||||
"""
|
||||
built_in_functions = {}
|
||||
for name, item in vars(model).items():
|
||||
if isinstance(item, types.FunctionType):
|
||||
built_in_functions[name] = item
|
||||
return built_in_functions
|
||||
|
||||
@staticmethod
|
||||
@logger.log_decorator()
|
||||
def load_built_in_comparators() -> object:
|
||||
"""
|
||||
加载包中的内建比较器
|
||||
Returns:
|
||||
|
||||
"""
|
||||
built_in_comparators = {}
|
||||
for name, item in vars(comparators).items():
|
||||
if isinstance(item, types.FunctionType):
|
||||
built_in_comparators[name] = item
|
||||
|
||||
return built_in_comparators
|
||||
|
||||
@logger.log_decorator()
|
||||
def set_bif_fun(self, model):
|
||||
"""
|
||||
将所有内置方法加载到依赖表中
|
||||
Returns:
|
||||
|
||||
"""
|
||||
for k, v in self.load_built_in_functions(model).items():
|
||||
self.update_environments(f"{k}()", v)
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
@logger.log_decorator()
|
||||
def load_built_in_functions(self, model):
|
||||
"""
|
||||
加载bif_functions包中的内建方法
|
||||
Returns:
|
||||
"""
|
||||
built_in_functions = {}
|
||||
for name, item in vars(model).items():
|
||||
if isinstance(item, types.FunctionType):
|
||||
built_in_functions[name] = item
|
||||
return built_in_functions
|
||||
|
||||
@staticmethod
|
||||
@logger.log_decorator()
|
||||
def load_built_in_comparators() -> object:
|
||||
"""
|
||||
加载包中的内建比较器
|
||||
Returns:
|
||||
|
||||
"""
|
||||
built_in_comparators = {}
|
||||
for name, item in vars(comparators).items():
|
||||
if isinstance(item, types.FunctionType):
|
||||
built_in_comparators[name] = item
|
||||
|
||||
return built_in_comparators
|
||||
|
||||
@logger.log_decorator()
|
||||
def set_bif_fun(self, model):
|
||||
"""
|
||||
加载内置方法
|
||||
Returns:
|
||||
|
||||
"""
|
||||
for k, v in self.load_built_in_functions(model).items():
|
||||
self.update_environments(f"{k}()", v)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
from common.bif_functions import bif_faker
|
||||
import extensions
|
||||
|
||||
print()
|
||||
loaders = Loaders()
|
||||
loaders.load_built_in_comparators()
|
||||
loaders.set_bif_fun(bif_faker)
|
||||
print(loaders.get_environments())
|
||||
|
||||
loaders.set_bif_fun(extensions)
|
||||
print(loaders.get_environments())
|
||||
from common.bif_functions import bif_faker
|
||||
import extensions
|
||||
|
||||
print()
|
||||
loaders = Loaders()
|
||||
loaders.load_built_in_comparators()
|
||||
loaders.set_bif_fun(bif_faker)
|
||||
print(loaders.get_environments())
|
||||
|
||||
loaders.set_bif_fun(extensions)
|
||||
print(loaders.get_environments())
|
||||
|
|
|
|||
|
|
@ -15,116 +15,116 @@ from common.validation.loaders import Loaders
|
|||
|
||||
|
||||
class Validator(Loaders):
|
||||
"""
|
||||
校验器
|
||||
主要功能：
|
||||
1、格式化校验变量
|
||||
2、校验期望结果与实际结果与预期一致，并返回校验结果
|
||||
"""
|
||||
validate_variables_list = []
|
||||
assertions = []
|
||||
|
||||
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 = []
|
||||
assertions = []
|
||||
|
||||
Returns: 返回数据格式 list
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
"""
|
||||
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,
|
||||
# "check_rt": check_rt
|
||||
})
|
||||
else:
|
||||
logger.error("参数格式错误！")
|
||||
|
||||
def validate(self, resp=None):
|
||||
"""
|
||||
校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
resp: ResponseObject对象实例
|
||||
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:
|
||||
Returns: 返回数据格式 list
|
||||
示例：
|
||||
[{"check":"result.user.name","comparator":"eq","expect":"chenyongzhi"}]
|
||||
|
||||
"""
|
||||
|
||||
validate_result = "通过"
|
||||
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, expr=check_item)
|
||||
try:
|
||||
# 获取比较器
|
||||
fun = self.load_built_in_comparators()[comparator]
|
||||
fun(actual_value=actual_value, expect_value=expect_value)
|
||||
except (AssertionError, TypeError) as e:
|
||||
validate_result = "失败"
|
||||
raise e
|
||||
finally:
|
||||
self.assertions.append({
|
||||
'检查项': check_item,
|
||||
'期望值': expect_value,
|
||||
'实际值': actual_value,
|
||||
'断言方法': comparator_dict.get(comparator),
|
||||
"断言结果": validate_result
|
||||
})
|
||||
|
||||
def run_validate(self, validate_variables, resp=None):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate，然后校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
validate_variables:参数格式 list、dict
|
||||
resp: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("参数格式错误！")
|
||||
|
||||
Returns:返回校验结果
|
||||
def validate(self, resp=None):
|
||||
"""
|
||||
校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
resp: ResponseObject对象实例
|
||||
|
||||
"""
|
||||
if not validate_variables:
|
||||
return ""
|
||||
# 清空校验变量
|
||||
self.validate_variables_list.clear()
|
||||
self.assertions.clear()
|
||||
self.uniform_validate(validate_variables)
|
||||
if not self.validate_variables_list:
|
||||
raise "uniform_validate 执行失败，无法进行 validate 校验"
|
||||
self.validate(resp)
|
||||
Returns:
|
||||
|
||||
"""
|
||||
|
||||
for validate_variable in self.validate_variables_list:
|
||||
check_item = validate_variable['check']
|
||||
expect_value = validate_variable['expect']
|
||||
comparator = validate_variable['comparator']
|
||||
if not check_item.startswith("$"):
|
||||
actual_value = check_item
|
||||
else:
|
||||
actual_value = Extractor.extract_value_by_jsonpath(resp_obj=resp, expr=check_item)
|
||||
try:
|
||||
fun = self.load_built_in_comparators()[comparator]
|
||||
fun(actual_value=actual_value, expect_value=expect_value)
|
||||
validate_result = "通过"
|
||||
except (AssertionError, TypeError) as e:
|
||||
validate_result = "失败"
|
||||
raise e
|
||||
finally:
|
||||
self.assertions.append({
|
||||
'检查项': check_item,
|
||||
'期望值': expect_value,
|
||||
'实际值': actual_value,
|
||||
'断言方法': comparator_dict.get(comparator),
|
||||
"断言结果": validate_result
|
||||
})
|
||||
|
||||
def run_validate(self, validate_variables, resp=None):
|
||||
"""
|
||||
统一格式化测试用例的验证变量validate，然后校验期望结果与实际结果与预期一致
|
||||
Args:
|
||||
validate_variables:参数格式 list、dict
|
||||
resp:ResponseObject对象实例
|
||||
|
||||
Returns:返回校验结果
|
||||
|
||||
"""
|
||||
if not validate_variables:
|
||||
return ""
|
||||
self.validate_variables_list.clear()
|
||||
self.assertions.clear()
|
||||
self.uniform_validate(validate_variables)
|
||||
if not self.validate_variables_list:
|
||||
raise "uniform_validate 执行失败，无法进行 validate 校验"
|
||||
self.validate(resp)
|
||||
|
||||
|
||||
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()
|
||||
validator.run_validate(validate_variables1, resp_obj)
|
||||
print(validator.assertions)
|
||||
validator.run_validate(validate_variables2, resp_obj)
|
||||
print(validator.assertions)
|
||||
validator.run_validate(validate_variables2, resp_obj)
|
||||
print(validator.assertions)
|
||||
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()
|
||||
validator.run_validate(validate_variables1, resp_obj)
|
||||
print(validator.assertions)
|
||||
validator.run_validate(validate_variables2, resp_obj)
|
||||
print(validator.assertions)
|
||||
validator.run_validate(validate_variables2, resp_obj)
|
||||
print(validator.assertions)
|
||||
|
|
|
|||
|
|
@ -9,8 +9,8 @@ class Config:
|
|||
# 测试数据所在路径
|
||||
# *****************************************************************
|
||||
templates = os.path.join(base_path, "cases", "templates", "template.xlsx") # 模板文件
|
||||
test_case = os.path.join(base_path, "cases", "cases", "test_api.xlsx")
|
||||
# test_case = os.path.join(base_path, "cases", "cases", "test_cases.xlsx")
|
||||
# test_case = os.path.join(base_path, "cases", "cases", "test_api.xlsx")
|
||||
test_case = os.path.join(base_path, "cases", "cases", "test_cases.xlsx")
|
||||
test_files = os.path.join(base_path, 'cases', 'files')
|
||||
# *****************************************************************
|
||||
|
||||
|
|
|
|||
|
|
@ -7,3 +7,5 @@
|
|||
@time: 2023/7/13 16:38
|
||||
@desc:
|
||||
"""
|
||||
a = {"key": "value", "c": ""}
|
||||
print(a.get("c", ""))
|
||||
|
|
|
|||
35
debug/pm.py
35
debug/pm.py
|
|
@ -1,6 +1,9 @@
|
|||
# 前置脚本代码
|
||||
import json
|
||||
|
||||
|
||||
def setup(pm):
|
||||
print("pm---------------->", pm.get_variables())
|
||||
print("pm---------------->", pm.variables())
|
||||
# request_data = pm.get_variables() # 获取得到请求数据
|
||||
"""
|
||||
request_data 的值: {'Url': '/login',
|
||||
|
|
@ -11,10 +14,10 @@ def setup(pm):
|
|||
'Expected': None, 'Response': '', 'Assertion': '', 'Error Log': ''
|
||||
}
|
||||
"""
|
||||
BSP_TOKEN = pm.get_environments("{{BSP_TOKEN}}") # 获取环境变量
|
||||
pm.update_environments("BSP_TOKEN_NEWS", BSP_TOKEN + "修改了环境变量") # 设置环境变量
|
||||
print("---->pm.get_environments", pm.get_environments("{{BSP_TOKEN_NEWS}}"))
|
||||
print("---->pm.get_variables", pm.get_variables())
|
||||
request = pm.variables()
|
||||
email = json.loads(request.get("Request Data")).get("email")
|
||||
pm.update_environments("email", email) # 设置环境变量
|
||||
print("---->pm.get_environments", pm.get_environments("{{email}}"))
|
||||
|
||||
|
||||
setup(pm)
|
||||
|
|
@ -22,19 +25,19 @@ setup(pm)
|
|||
|
||||
# 后置脚本代码
|
||||
def tear_down(pm):
|
||||
vars_data = pm.get_environments("{{变量名称}}") # 获取环境变量
|
||||
response = pm.get_variables() # 获取得到响应结果对象
|
||||
# vars_data = pm.get_environments("{{变量名称}}") # 获取环境变量
|
||||
response = pm.variables() # 获取得到响应结果对象
|
||||
response.json()
|
||||
print(f"请求地址 --> {response.request.url}")
|
||||
print(f"请求头 --> {response.request.headers}")
|
||||
print(f"请求 body --> {response.request.body}")
|
||||
print(f"接口状态--> {response.status_code}")
|
||||
print(f"接口耗时--> {response.elapsed}")
|
||||
print(f"接口响应--> {response.text}")
|
||||
# print(f"请求地址 --> {response.request.url}")
|
||||
# print(f"请求头 --> {response.request.headers}")
|
||||
# print(f"请求 body --> {response.request.body}")
|
||||
# print(f"接口状态--> {response.status_code}")
|
||||
# print(f"接口耗时--> {response.elapsed}")
|
||||
# print(f"接口响应--> {response.text}")
|
||||
token = response.json()['token']
|
||||
pm.update_environments("BSP_TOKEN_NEWS", token + vars_data) # 重新设置环境变量
|
||||
print("---->pm.get_environments", pm.get_environments("{{BSP_TOKEN_NEWS}}"))
|
||||
print("---->pm.get_variables", pm.get_variables())
|
||||
pm.update_environments("token", token) # 重新设置环境变量
|
||||
# print("---->pm.get_environments", pm.get_environments("{{BSP_TOKEN_NEWS}}"))
|
||||
# print("---->pm.get_variables", pm.get_variables())
|
||||
|
||||
|
||||
tear_down(pm)
|
||||
|
|
|
|||
|
|
@ -42,57 +42,22 @@ class TestProjectApi(unittest.TestCase):
|
|||
regex, keys, deps, jp_dict, ex_request_data = self.action.extractor_info(item)
|
||||
setup_script, teardown_script = self.action.script(item)
|
||||
self.action.pause_execution(st)
|
||||
|
||||
# 首执行 sql
|
||||
self.action.exc_sql(item)
|
||||
if method.upper() == 'SQL':
|
||||
self.skipTest("这条测试用例被 SQL 吃了，所以放弃执行了！！")
|
||||
|
||||
# 执行动态代码
|
||||
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, request_data_type = self.action.request_info(item)
|
||||
|
||||
# 分析请求参数信息
|
||||
headers, request_data = self.action.analysis_request(request_data, h_crypto, headers, r_crypto, ex_request_data)
|
||||
result = "PASS"
|
||||
|
||||
# 执行请求操作
|
||||
kwargs = {request_data_type: request_data, 'headers': headers, "params": query_str}
|
||||
self.action.send_request(host, url, method, teardown_script, **kwargs)
|
||||
|
||||
try:
|
||||
# 提取响应
|
||||
self.action.substitute_data(self.action.response_json, regex=regex, keys=keys, deps=deps, jp_dict=jp_dict)
|
||||
except Exception as err:
|
||||
self.action.logger.error(f"提取响应失败：{sheet}_{iid}_{name}_{desc}"
|
||||
f"\nregex={regex};"
|
||||
f" \nkeys={keys};"
|
||||
f"\ndeps={deps};"
|
||||
f"\njp_dict={jp_dict}"
|
||||
f"\n{err}")
|
||||
|
||||
# 修正断言
|
||||
self.action.analysis_response(sheet, iid, name, desc, regex, keys, deps, jp_dict)
|
||||
expected = self.action.replace_dependent_parameter(expected)
|
||||
try:
|
||||
# print(f"期望结果--> {expected}")
|
||||
# 执行断言 返回结果元组
|
||||
self.action.run_validate(expected, self.action.response_json)
|
||||
except Exception as e:
|
||||
result = "FAIL"
|
||||
self.action.logger.error(f'异常用例: **{sheet}_{iid}_{name}_{desc}**\n{e}')
|
||||
raise e
|
||||
finally:
|
||||
print(f"断言结果-->", self.action.assertions)
|
||||
response = self.action.response.text if self.action.response is not None else str(self.action.response)
|
||||
# 响应结果及测试结果回写 excel
|
||||
excel.write_back(sheet_name=sheet, i=iid, response=response, test_result=result,
|
||||
assert_log=str(self.action.assertions))
|
||||
self.action.execute_validation(excel, sheet, iid, name, desc, expected)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls) -> None:
|
||||
|
|
|
|||
Loading…
Reference in New Issue