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)