runner

runner.py

run_testcase()

__run_step_request()

__run_step_testcase()

可用资料

无

导包

import os
import time
import uuid  # 32个十六进制数字组成的字符串
from datetime import datetime
from typing import List, Dict, Text, NoReturn
try:  # 导成功就用allure报告
import allure

USE_ALLURE = True
except ModuleNotFoundError:
USE_ALLURE = False

from loguru import logger

from httprunner import utils, exceptions
from httprunner.client import HttpSession
from httprunner.exceptions import ValidationFailure, ParamsError
from httprunner.ext.uploader import prepare_upload_step
from httprunner.loader import load_project_meta, load_testcase_file
from httprunner.parser import build_url, parse_data, parse_variables_mapping
from httprunner.response import ResponseObject
from httprunner.testcase import Config, Step
from httprunner.utils import merge_variables
from httprunner.models import (
TConfig,
TStep,
VariablesMapping,
StepData,
TestCaseSummary,
TestCaseTime,
TestCaseInOut,
ProjectMeta,
TestCase,
Hooks,
)

源码附注释

class HttpRunner(object):
# 属性： Config 对象
config: Config
# 步骤列表： Step
teststeps: 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())

@property
def 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_meta
return self

def with_session(self, session: HttpSession) -> "HttpRunner":
self.__session = session
return self

def with_case_id(self, case_id: Text) -> "HttpRunner":
self.__case_id = case_id
return self

def with_variables(self, variables: VariablesMapping) -> "HttpRunner":
self.__session_variables = variables
return self

def with_export(self, export: List[Text]) -> "HttpRunner":
self.__export = export
return self

def __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 variables

request: parsed request dict
response: ResponseObject for current response

hook_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}")
return

for 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_eval
else:
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 request
step.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，返回对应的value
method = parsed_request_dict.pop("method")
url_path = parsed_request_dict.pop("url")
# 组装最终请求url
url = build_url(self.__config.base_url, url_path)
parsed_request_dict["verify"] = self.__config.verify
parsed_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 request
err_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 response
err_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.extract
extract_mapping = resp_obj.extract(extractors)
step_data.export_vars = extract_mapping

variables_mapping = step.variables
variables_mapping.update(extract_mapping)

# validate 验证
validators = step.validators
session_success = False
try:
resp_obj.validate(
validators, variables_mapping, self.__project_meta.functions
)
session_success = True
except ValidationFailure:
session_success = False
log_req_resp_details()
# log testcase duration before raise ValidationFailure
self.__duration = time.time() - self.__start_at
raise
finally:
self.success = session_success
step_data.success = session_success

if hasattr(self.__session, "data"):
# httprunner.client.HttpSession, not locust.clients.HttpSession
# save request & response meta data
self.__session.data.success = session_success
self.__session.data.validators = resp_obj.validation_results

# save step data
step_data.data = self.__session.data
# 返回步骤数据
return step_data

def __run_step_testcase(self, step: TStep) -> StepData:
"""run teststep: referenced testcase 步骤中引入的其他测试用例"""
step_data = StepData(name=step.name)
step_variables = step.variables
step_export = step.export

# setup hooks
if 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.testcase
case_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.testcase
else:
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 hooks
if step.teardown_hooks:
self.__call_hooks(step.teardown_hooks, step.variables, "teardown testcase")

step_data.data = case_result.get_step_datas()  # list of step data
step_data.export_vars = case_result.get_export_variables()
step_data.success = case_result.success
self.success = case_result.success

if step_data.export_vars:
logger.info(f"export variables: {step_data.export_vars}")

return step_data

def __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_vars

def __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.config
self.__teststeps = testcase.teststeps

# prepare
self.__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 teststeps
extracted_variables: VariablesMapping = {}

# run teststeps
for 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_at
return self

def 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_datas

def get_export_variables(self) -> Dict:  # 导出变量字典
# override testcase export vars with step export
export_var_names = self.__export or self.__config.export
export_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_mapping

def get_summary(self) -> TestCaseSummary: # 结果集
"""get testcase result summary"""
start_at_timestamp = self.__start_at
start_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 name
config_variables = self.__config.variables
if 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 meta
allure.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}")