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)