标签:comparator py value item HttpRunner3 源码 expect check validator
response
上一篇说的
client.py
来发送请求,这里就来看另一个response.py
,该文件主要是完成测试断言方法
可用资料
jmespath[json数据取值处理]: https://github.com/jmespath/jmespath.py
导包
from typing import Dict, Text, Any, NoReturn
import jmespath
import requests
from jmespath.exceptions import JMESPathError
from loguru import logger
from httprunner import exceptions
from httprunner.exceptions import ValidationFailure, ParamsError
from httprunner.models import VariablesMapping, Validators, FunctionsMapping
# 数据解析,字符串解析,方法字典
from httprunner.parser import parse_data, parse_string_value, get_mapping_function
源码附注释
def get_uniform_comparator(comparator: Text):
""" convert comparator alias to uniform name
转换统一的比较器名称
"""
if comparator in ["eq", "equals", "equal"]:
return "equal"
elif comparator in ["lt", "less_than"]:
return "less_than"
elif comparator in ["le", "less_or_equals"]:
return "less_or_equals"
elif comparator in ["gt", "greater_than"]:
return "greater_than"
elif comparator in ["ge", "greater_or_equals"]:
return "greater_or_equals"
elif comparator in ["ne", "not_equal"]:
return "not_equal"
elif comparator in ["str_eq", "string_equals"]:
return "string_equals"
elif comparator in ["len_eq", "length_equal"]:
return "length_equal"
elif comparator in [
"len_gt",
"length_greater_than",
]:
return "length_greater_than"
elif comparator in [
"len_ge",
"length_greater_or_equals",
]:
return "length_greater_or_equals"
elif comparator in ["len_lt", "length_less_than"]:
return "length_less_than"
elif comparator in [
"len_le",
"length_less_or_equals",
]:
return "length_less_or_equals"
else:
return comparator
def uniform_validator(validator):
""" unify validator
统一验证器
Args:
validator (dict): validator maybe in two formats:
format1: this is kept for compatibility with the previous versions.
{"check": "status_code", "comparator": "eq", "expect": 201}
{"check": "$resp_body_success", "comparator": "eq", "expect": True}
format2: recommended new version, {assert: [check_item, expected_value]}
{'eq': ['status_code', 201]}
{'eq': ['$resp_body_success', True]}
Returns
dict: validator info
{
"check": "status_code",
"expect": 201,
"assert": "equals"
}
"""
if not isinstance(validator, dict):
raise ParamsError(f"invalid validator: {validator}")
if "check" in validator and "expect" in validator:
# format1
check_item = validator["check"]
expect_value = validator["expect"]
message = validator.get("message", "")
comparator = validator.get("comparator", "eq")
elif len(validator) == 1:
# format2
comparator = list(validator.keys())[0]
compare_values = validator[comparator]
if not isinstance(compare_values, list) or len(compare_values) not in [2, 3]:
raise ParamsError(f"invalid validator: {validator}")
check_item = compare_values[0]
expect_value = compare_values[1]
if len(compare_values) == 3:
message = compare_values[2]
else:
# len(compare_values) == 2
message = ""
else:
raise ParamsError(f"invalid validator: {validator}")
# uniform comparator, e.g. lt => less_than, eq => equals
assert_method = get_uniform_comparator(comparator)
return {
"check": check_item,
"expect": expect_value,
"assert": assert_method,
"message": message,
}
class ResponseObject(object):
def __init__(self, resp_obj: requests.Response):
""" initialize with a requests.Response object
Args:
resp_obj (instance): requests.Response instance
"""
self.resp_obj = resp_obj
self.validation_results: Dict = {}
def __getattr__(self, key):
# 魔术方法,查找属性时调用 实例对象.属性名
if key in ["json", "content", "body"]:
try:
value = self.resp_obj.json()
except ValueError:
value = self.resp_obj.content
elif key == "cookies":
value = self.resp_obj.cookies.get_dict()
else:
try:
value = getattr(self.resp_obj, key)
except AttributeError:
err_msg = "ResponseObject does not have attribute: {}".format(key)
logger.error(err_msg)
raise exceptions.ParamsError(err_msg)
self.__dict__[key] = value
return value
def _search_jmespath(self, expr: Text) -> Any:
# 根据jmespath语法搜索提取实际结果值
resp_obj_meta = {
"status_code": self.status_code,
"headers": self.headers,
"cookies": self.cookies,
"body": self.body,
}
if not expr.startswith(tuple(resp_obj_meta.keys())):
return expr
try:
check_value = jmespath.search(expr, resp_obj_meta)
except JMESPathError as ex:
logger.error(
f"failed to search with jmespath\n"
f"expression: {expr}\n"
f"data: {resp_obj_meta}\n"
f"exception: {ex}"
)
raise
return check_value
def extract(self, extractors: Dict[Text, Text]) -> Dict[Text, Any]:
# 根据jmespath 语法找到值 放入 提取参数字典中
if not extractors:
return {}
extract_mapping = {}
for key, field in extractors.items():
field_value = self._search_jmespath(field)
extract_mapping[key] = field_value
logger.info(f"extract mapping: {extract_mapping}")
return extract_mapping
# 验证&结果回写
def validate(
self,
validators: Validators,
variables_mapping: VariablesMapping = None,
functions_mapping: FunctionsMapping = None,
) -> NoReturn:
variables_mapping = variables_mapping or {}
functions_mapping = functions_mapping or {}
self.validation_results = {}
if not validators:
return
validate_pass = True
failures = []
for v in validators:
if "validate_extractor" not in self.validation_results:
self.validation_results["validate_extractor"] = []
u_validator = uniform_validator(v)
# check item
check_item = u_validator["check"]
if "$" in check_item:
# 需要检查的元素是 变量或者函数
# check_item is variable or function
check_item = parse_data(
check_item, variables_mapping, functions_mapping
)
check_item = parse_string_value(check_item)
if check_item and isinstance(check_item, Text):
check_value = self._search_jmespath(check_item)
else:
# variable or function evaluation result is "" or not text
check_value = check_item
# comparator
assert_method = u_validator["assert"]
assert_func = get_mapping_function(assert_method, functions_mapping)
# expect item
expect_item = u_validator["expect"]
# parse expected value with config/teststep/extracted variables
expect_value = parse_data(expect_item, variables_mapping, functions_mapping)
# message
message = u_validator["message"]
# parse message with config/teststep/extracted variables
message = parse_data(message, variables_mapping, functions_mapping)
validate_msg = f"assert {check_item} {assert_method} {expect_value}({type(expect_value).__name__})"
validator_dict = {
"comparator": assert_method,
"check": check_item,
"check_value": check_value,
"expect": expect_item,
"expect_value": expect_value,
"message": message,
}
try:
assert_func(check_value, expect_value, message)
validate_msg += "\t==> pass"
logger.info(validate_msg)
validator_dict["check_result"] = "pass"
except AssertionError as ex:
validate_pass = False
validator_dict["check_result"] = "fail"
validate_msg += "\t==> fail"
validate_msg += (
f"\n"
f"check_item: {check_item}\n"
f"check_value: {check_value}({type(check_value).__name__})\n"
f"assert_method: {assert_method}\n"
f"expect_value: {expect_value}({type(expect_value).__name__})"
)
message = str(ex)
if message:
validate_msg += f"\nmessage: {message}"
logger.error(validate_msg)
failures.append(validate_msg)
self.validation_results["validate_extractor"].append(validator_dict)
if not validate_pass:
failures_string = "\n".join([failure for failure in failures])
raise ValidationFailure(failures_string)
标签:comparator,py,value,item,HttpRunner3,源码,expect,check,validator 来源: https://www.cnblogs.com/zy7y/p/15117856.html
本站声明: 1. iCode9 技术分享网(下文简称本站)提供的所有内容,仅供技术学习、探讨和分享; 2. 关于本站的所有留言、评论、转载及引用,纯属内容发起人的个人观点,与本站观点和立场无关; 3. 关于本站的所有言论和文字,纯属内容发起人的个人观点,与本站观点和立场无关; 4. 本站文章均是网友提供,不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属;如您发现该文章侵犯了您的权益,可联系我们第一时间进行删除; 5. 本站为非盈利性的个人网站,所有内容不会用来进行牟利,也不会利用任何形式的广告来间接获益,纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。