response

上一篇说的

client.py

来发送请求,这里就来看另一个

response.py

,该文件主要是完成测试断言方法

可用资料

jmespath[json数据取值处理]: https://github.com/jmespath/jmespath.py

导包

from typing import Dict, Text, Any, NoReturnimport jmespathimport requestsfrom jmespath.exceptions import JMESPathErrorfrom loguru import loggerfrom httprunner import exceptionsfrom httprunner.exceptions import ValidationFailure, ParamsErrorfrom 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 comparatordef 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]}Returnsdict: 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:# format1check_item = validator["check"]expect_value = validator["expect"]message = validator.get("message", "")comparator = validator.get("comparator", "eq")elif len(validator) == 1:# format2comparator = 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) == 2message = ""else:raise ParamsError(f"invalid validator: {validator}")# uniform comparator, e.g. lt => less_than, eq => equalsassert_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 objectArgs:resp_obj (instance): requests.Response instance"""self.resp_obj = resp_objself.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.contentelif 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] = valuereturn valuedef _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 exprtry: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}")raisereturn check_valuedef 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_valuelogger.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:returnvalidate_pass = Truefailures = []for v in validators:if "validate_extractor" not in self.validation_results:self.validation_results["validate_extractor"] = []u_validator = uniform_validator(v)# check itemcheck_item = u_validator["check"]if "$" in check_item:# 需要检查的元素是 变量或者函数# check_item is variable or functioncheck_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 textcheck_value = check_item# comparatorassert_method = u_validator["assert"]assert_func = get_mapping_function(assert_method, functions_mapping)# expect itemexpect_item = u_validator["expect"]# parse expected value with config/teststep/extracted variablesexpect_value = parse_data(expect_item, variables_mapping, functions_mapping)# messagemessage = u_validator["message"]# parse message with config/teststep/extracted variablesmessage = 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 = Falsevalidator_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)