runner

HttpRunner的执行函数存在的位置，程序内部执行运行入口了，文件名称很明显了

runner.py

，其中最主要的为

run_testcase()

,

__run_step_request()

,

__run_step_testcase()

,方法

可用资料

无

导包

import osimport timeimport uuid  # 32个十六进制数字组成的字符串from datetime import datetimefrom typing import List, Dict, Text, NoReturntry:  # 导成功就用allure报告import allureUSE_ALLURE = Trueexcept ModuleNotFoundError:USE_ALLURE = Falsefrom loguru import loggerfrom httprunner import utils, exceptionsfrom httprunner.client import HttpSessionfrom httprunner.exceptions import ValidationFailure, ParamsErrorfrom httprunner.ext.uploader import prepare_upload_stepfrom httprunner.loader import load_project_meta, load_testcase_filefrom httprunner.parser import build_url, parse_data, parse_variables_mappingfrom httprunner.response import ResponseObjectfrom httprunner.testcase import Config, Stepfrom httprunner.utils import merge_variablesfrom httprunner.models import (TConfig,TStep,VariablesMapping,StepData,TestCaseSummary,TestCaseTime,TestCaseInOut,ProjectMeta,TestCase,Hooks,)

源码附注释

class HttpRunner(object):# 属性： Config 对象config: Config# 步骤列表： Stepteststeps: List[Step]# 测试结果success: bool = False  # indicate testcase execution result__config: TConfig__teststeps: List[TStep]__project_meta: ProjectMeta = None__case_id: Text = ""__export: List[Text] = []__step_datas: List[StepData] = []__session: HttpSession = None__session_variables: VariablesMapping = {}# time__start_at: float = 0__duration: float = 0# log__log_path: Text = ""def __init_tests__(self) -> NoReturn:self.__config = self.config.perform()self.__teststeps = []for step in self.teststeps:self.__teststeps.append(step.perform())@propertydef raw_testcase(self) -> TestCase:if not hasattr(self, "__config"):self.__init_tests__()# 对象模型return TestCase(config=self.__config, teststeps=self.__teststeps)def with_project_meta(self, project_meta: ProjectMeta) -> "HttpRunner":self.__project_meta = project_metareturn selfdef with_session(self, session: HttpSession) -> "HttpRunner":self.__session = sessionreturn selfdef with_case_id(self, case_id: Text) -> "HttpRunner":self.__case_id = case_idreturn selfdef with_variables(self, variables: VariablesMapping) -> "HttpRunner":self.__session_variables = variablesreturn selfdef with_export(self, export: List[Text]) -> "HttpRunner":self.__export = exportreturn selfdef __call_hooks(self, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text,) -> NoReturn:""" call hook actions.调用hooks 函数，结果写到 步骤变量中Args:hooks (list): each hook in hooks list maybe in two format.format1 (str): only call hook functions.${func()}format2 (dict): assignment, the value returned by hook function will be assigned to variable.{"var": "${func()}"}step_variables: current step variables to call hook, include two special variablesrequest: parsed request dictresponse: ResponseObject for current responsehook_msg: setup/teardown request/testcase"""logger.info(f"call hook actions: {hook_msg}")if not isinstance(hooks, List):logger.error(f"Invalid hooks format: {hooks}")returnfor hook in hooks:if isinstance(hook, Text):# format 1: ["${func()}"]logger.debug(f"call hook function: {hook}")parse_data(hook, step_variables, self.__project_meta.functions)elif isinstance(hook, Dict) and len(hook) == 1:# format 2: {"var": "${func()}"}var_name, hook_content = list(hook.items())[0]hook_content_eval = parse_data(hook_content, step_variables, self.__project_meta.functions)logger.debug(f"call hook function: {hook_content}, got value: {hook_content_eval}")logger.debug(f"assign variable: {var_name} = {hook_content_eval}")step_variables[var_name] = hook_content_evalelse:logger.error(f"Invalid hook format: {hook}")def __run_step_request(self, step: TStep) -> StepData:"""run teststep: request"""step_data = StepData(name=step.name)# parse# 准备文件上传步骤prepare_upload_step(step, self.__project_meta.functions)# 请求字典request_dict = step.request.dict()request_dict.pop("upload", None)# 解析变量&方法数据parsed_request_dict = parse_data(request_dict, step.variables, self.__project_meta.functions)# 设置请求头parsed_request_dict["headers"].setdefault("HRUN-Request-ID",f"HRUN-{self.__case_id}-{str(int(time.time() * 1000))[-6:]}",)# 步骤参数字典 加 key requeststep.variables["request"] = parsed_request_dict# setup hooks  setup 的钩子函数if step.setup_hooks:self.__call_hooks(step.setup_hooks, step.variables, "setup request")# prepare arguments# 移除字典中的method，返回对应的valuemethod = parsed_request_dict.pop("method")url_path = parsed_request_dict.pop("url")# 组装最终请求urlurl = build_url(self.__config.base_url, url_path)parsed_request_dict["verify"] = self.__config.verifyparsed_request_dict["json"] = parsed_request_dict.pop("req_json", {})# request 发起请求resp = self.__session.request(method, url, **parsed_request_dict)resp_obj = ResponseObject(resp)step.variables["response"] = resp_obj# teardown hooks 请求完成之后执行函数if step.teardown_hooks:self.__call_hooks(step.teardown_hooks, step.variables, "teardown request")def log_req_resp_details(): # 详细日志err_msg = "\\n{} DETAILED REQUEST & RESPONSE {}\\n".format("*" * 32, "*" * 32)# log requesterr_msg += "====== request details ======\\n"err_msg += f"url: {url}\\n"err_msg += f"method: {method}\\n"headers = parsed_request_dict.pop("headers", {})err_msg += f"headers: {headers}\\n"for k, v in parsed_request_dict.items():v = utils.omit_long_data(v)err_msg += f"{k}: {repr(v)}\\n"err_msg += "\\n"# log responseerr_msg += "====== response details ======\\n"err_msg += f"status_code: {resp.status_code}\\n"err_msg += f"headers: {resp.headers}\\n"err_msg += f"body: {repr(resp.text)}\\n"logger.error(err_msg)# extract 提取参数extractors = step.extractextract_mapping = resp_obj.extract(extractors)step_data.export_vars = extract_mappingvariables_mapping = step.variablesvariables_mapping.update(extract_mapping)# validate 验证validators = step.validatorssession_success = Falsetry:resp_obj.validate(validators, variables_mapping, self.__project_meta.functions)session_success = Trueexcept ValidationFailure:session_success = Falselog_req_resp_details()# log testcase duration before raise ValidationFailureself.__duration = time.time() - self.__start_atraisefinally:self.success = session_successstep_data.success = session_successif hasattr(self.__session, "data"):# httprunner.client.HttpSession, not locust.clients.HttpSession# save request & response meta dataself.__session.data.success = session_successself.__session.data.validators = resp_obj.validation_results# save step datastep_data.data = self.__session.data# 返回步骤数据return step_datadef __run_step_testcase(self, step: TStep) -> StepData:"""run teststep: referenced testcase 步骤中引入的其他测试用例"""step_data = StepData(name=step.name)step_variables = step.variablesstep_export = step.export# setup hooksif step.setup_hooks:self.__call_hooks(step.setup_hooks, step_variables, "setup testcase")# 运行测试用例if hasattr(step.testcase, "config") and hasattr(step.testcase, "teststeps"):testcase_cls = step.testcasecase_result = (testcase_cls().with_session(self.__session).with_case_id(self.__case_id).with_variables(step_variables).with_export(step_export).run())elif isinstance(step.testcase, Text):if os.path.isabs(step.testcase):ref_testcase_path = step.testcaseelse:ref_testcase_path = os.path.join(self.__project_meta.RootDir, step.testcase)case_result = (HttpRunner().with_session(self.__session).with_case_id(self.__case_id).with_variables(step_variables).with_export(step_export).run_path(ref_testcase_path))else:raise exceptions.ParamsError(f"Invalid teststep referenced testcase: {step.dict()}")# teardown hooksif step.teardown_hooks:self.__call_hooks(step.teardown_hooks, step.variables, "teardown testcase")step_data.data = case_result.get_step_datas()  # list of step datastep_data.export_vars = case_result.get_export_variables()step_data.success = case_result.successself.success = case_result.successif step_data.export_vars:logger.info(f"export variables: {step_data.export_vars}")return step_datadef __run_step(self, step: TStep) -> Dict:"""run teststep, teststep maybe a request or referenced testcase"""logger.info(f"run step begin: {step.name} >>>>>>")if step.request:step_data = self.__run_step_request(step)elif step.testcase:step_data = self.__run_step_testcase(step)else:raise ParamsError(f"teststep is neither a request nor a referenced testcase: {step.dict()}")self.__step_datas.append(step_data)logger.info(f"run step end: {step.name} <<<<<<\\n")return step_data.export_varsdef __parse_config(self, config: TConfig) -> NoReturn:"""解析配置"""config.variables.update(self.__session_variables)config.variables = parse_variables_mapping(config.variables, self.__project_meta.functions)config.name = parse_data(config.name, config.variables, self.__project_meta.functions)config.base_url = parse_data(config.base_url, config.variables, self.__project_meta.functions)def run_testcase(self, testcase: TestCase) -> "HttpRunner":"""run specified testcase， 运行单一测试用例Examples:>>> testcase_obj = TestCase(config=TConfig(...), teststeps=[TStep(...)])>>> HttpRunner().with_project_meta(project_meta).run_testcase(testcase_obj)"""self.__config = testcase.configself.__teststeps = testcase.teststeps# prepareself.__project_meta = self.__project_meta or load_project_meta(self.__config.path)self.__parse_config(self.__config)self.__start_at = time.time()self.__step_datas: List[StepData] = []self.__session = self.__session or HttpSession()# save extracted variables of teststepsextracted_variables: VariablesMapping = {}# run teststepsfor step in self.__teststeps:# override variables# step variables > extracted variables from previous steps 步骤变量step.variables = merge_variables(step.variables, extracted_variables)# step variables > testcase config variables 测试用例变量step.variables = merge_variables(step.variables, self.__config.variables)# parse variables 解析变量step.variables = parse_variables_mapping(step.variables, self.__project_meta.functions)# run step 运行步骤if USE_ALLURE:with allure.step(f"step: {step.name}"):extract_mapping = self.__run_step(step)else:extract_mapping = self.__run_step(step)# save extracted variables to session variables 提取参数extracted_variables.update(extract_mapping)self.__session_variables.update(extracted_variables)self.__duration = time.time() - self.__start_atreturn selfdef run_path(self, path: Text) -> "HttpRunner":  # 文件类测试用例if not os.path.isfile(path):raise exceptions.ParamsError(f"Invalid testcase path: {path}")testcase_obj = load_testcase_file(path)   # 转成测试用例对象return self.run_testcase(testcase_obj)  # 运行用例def run(self) -> "HttpRunner":""" run current testcase 运行当前测试用例Examples:>>> TestCaseRequestWithFunctions().run()"""self.__init_tests__()testcase_obj = TestCase(config=self.__config, teststeps=self.__teststeps)return self.run_testcase(testcase_obj)def get_step_datas(self) -> List[StepData]: # 步骤数据列表return self.__step_datasdef get_export_variables(self) -> Dict:  # 导出变量字典# override testcase export vars with step exportexport_var_names = self.__export or self.__config.exportexport_vars_mapping = {}for var_name in export_var_names:if var_name not in self.__session_variables:raise ParamsError(f"failed to export variable {var_name} from session variables {self.__session_variables}")export_vars_mapping[var_name] = self.__session_variables[var_name]return export_vars_mappingdef get_summary(self) -> TestCaseSummary: # 结果集"""get testcase result summary"""start_at_timestamp = self.__start_atstart_at_iso_format = datetime.utcfromtimestamp(start_at_timestamp).isoformat()return TestCaseSummary(name=self.__config.name,success=self.success,case_id=self.__case_id,time=TestCaseTime(start_at=self.__start_at,start_at_iso_format=start_at_iso_format,duration=self.__duration,),in_out=TestCaseInOut(config_vars=self.__config.variables,export_vars=self.get_export_variables(),),log=self.__log_path,step_datas=self.__step_datas,)def test_start(self, param: Dict = None) -> "HttpRunner":"""main entrance, discovered by pytest test_start函数 由Pytest发现收集执行"""self.__init_tests__()self.__project_meta = self.__project_meta or load_project_meta(self.__config.path)self.__case_id = self.__case_id or str(uuid.uuid4())self.__log_path = self.__log_path or os.path.join(self.__project_meta.RootDir, "logs", f"{self.__case_id}.run.log")log_handler = logger.add(self.__log_path, level="DEBUG")# parse config nameconfig_variables = self.__config.variablesif param:config_variables.update(param)config_variables.update(self.__session_variables)self.__config.name = parse_data(self.__config.name, config_variables, self.__project_meta.functions)if USE_ALLURE:# update allure report metaallure.dynamic.title(self.__config.name)allure.dynamic.description(f"TestCase ID: {self.__case_id}")logger.info(f"Start to run testcase: {self.__config.name}, TestCase ID: {self.__case_id}")try:return self.run_testcase(TestCase(config=self.__config, teststeps=self.__teststeps))finally:logger.remove(log_handler) # 删除日志文件logger.info(f"generate testcase log: {self.__log_path}")