|
-loader.py
核心代码：
将载入的yaml/json文件转化为字典，字典转化为testcase对象
def load_testcase_file(testcase_file: Text) -> TestCase:
"""load testcase file and validate with pydantic model"""
testcase_content = load_test_file(testcase_file)
testcase_obj = load_testcase(testcase_content)
testcase_obj.config.path = testcase_file
return testcase_obj
?将载入的.env中的环境变量输出为字典
def load_dot_env_file(dot_env_path: Text) -> Dict:
将载入的csv文件中定义数据输出为字典列表 用于参数化测试数据
def load_csv_file(csv_file: Text) -> List[Dict]:
输出文件夹下所有yml/.yaml/.json/_test.py列表
def load_folder_files(folder_path: Text, recursive: bool = True) -> List:
载入模块的所有方法，返回方法字典
def load_module_functions(module) -> Dict[Text, Callable]:
""" load python module functions.
Args:
module: python module
Returns:
dict: functions mapping for specified python module
{
"func1_name": func1,
"func2_name": func2
}
"""
?搜索某文件 返回绝对路径 递归不断向上查找直到根目录
def locate_file(start_path: Text, file_name: Text) -> Text:
定位debugtalk.py的位置，该文件作为httprunner项目的锚点，相对路径依据于此?
def locate_debugtalk_py(start_path: Text) -> Text:
载入debugtalk中的function ，载入模型，载入模型中的function
def load_debugtalk_functions() -> Dict[Text, Callable]:
从testpath中载入项目相关的test case/.env/debugtalk.py
def load_project_meta(test_path: Text, reload: bool = False) -> ProjectMeta:
?将绝对路径改为相对路径
def convert_relative_project_root_dir(abs_path: Text) -> Text:
-make.py 主要用来讲yaml/json文件转化为pytest
核心代码
make_testcase(testcase: Dict, dir_path: Text = None) -> Text:
"""将testcase字典转化为pytest文件"""
# ensure compatibility with testcase format v2
testcase = ensure_testcase_v3(testcase)
# validate testcase format
load_testcase(testcase)
testcase_abs_path = __ensure_absolute(testcase["config"]["path"])
logger.info(f"start to make testcase: {testcase_abs_path}")
testcase_python_abs_path, testcase_cls_name = convert_testcase_path(
testcase_abs_path
)
if dir_path:
testcase_python_abs_path = os.path.join(
dir_path, os.path.basename(testcase_python_abs_path)
)
global pytest_files_made_cache_mapping
if testcase_python_abs_path in pytest_files_made_cache_mapping:
return testcase_python_abs_path
config = testcase["config"]
config["path"] = convert_relative_project_root_dir(testcase_python_abs_path)
config["variables"] = convert_variables(
config.get("variables", {}), testcase_abs_path
)
# prepare reference testcase
imports_list = []
teststeps = testcase["teststeps"]
for teststep in teststeps:
if not teststep.get("testcase"):
continue
# 处理每个teststep引用的testcase
ref_testcase_path = __ensure_absolute(teststep["testcase"])
test_content = load_test_file(ref_testcase_path)
if not isinstance(test_content, Dict):
raise exceptions.TestCaseFormatError(f"Invalid teststep: {teststep}")
# api in v2 format, convert to v3 testcase
if "request" in test_content and "name" in test_content:
test_content = ensure_testcase_v3_api(test_content)
test_content.setdefault("config", {})["path"] = ref_testcase_path
ref_testcase_python_abs_path = make_testcase(test_content)
#将引用testcase的输出加入本step的输出
ref_testcase_export: List = test_content["config"].get("export", [])
if ref_testcase_export:
step_export: List = teststep.setdefault("export", [])
step_export.extend(ref_testcase_export)
teststep["export"] = list(set(step_export))
# prepare ref testcase class name
ref_testcase_cls_name = pytest_files_made_cache_mapping[
ref_testcase_python_abs_path
]
teststep["testcase"] = ref_testcase_cls_name
# 准备import testcase
ref_testcase_python_relative_path = convert_relative_project_root_dir(
ref_testcase_python_abs_path
)
ref_module_name, _ = os.path.splitext(ref_testcase_python_relative_path)
ref_module_name = ref_module_name.replace(os.sep, ".")
import_expr = f"from {ref_module_name} import TestCase{ref_testcase_cls_name} as {ref_testcase_cls_name}"
if import_expr not in imports_list:
imports_list.append(import_expr)
testcase_path = convert_relative_project_root_dir(testcase_abs_path)
# current file compared to ProjectRootDir
diff_levels = len(testcase_path.split(os.sep))
data = {
"version": __version__,
"testcase_path": testcase_path,
"diff_levels": diff_levels,
"class_name": f"TestCase{testcase_cls_name}",
"imports_list": imports_list,
"config_chain_style": make_config_chain_style(config),
"parameters": config.get("parameters"),
"teststeps_chain_style": [
make_teststep_chain_style(step) for step in teststeps
],
}
content = __TEMPLATE__.render(data)
#确保新的文件目录存在
dir_path = os.path.dirname(testcase_python_abs_path)
if not os.path.exists(dir_path):
os.makedirs(dir_path)
with open(testcase_python_abs_path, "w", encoding="utf-8") as f:
f.write(content)
pytest_files_made_cache_mapping[testcase_python_abs_path] = testcase_cls_name
__ensure_testcase_module(testcase_python_abs_path)
logger.info(f"generated testcase: {testcase_python_abs_path}")
return testcase_python_abs_path
?主要将路径和文件下的testcase转化为pytest并加入到pytest_files_run_set中
def __make(tests_path: Text) -> NoReturn:
""" make testcase(s) with testcase/testsuite/folder absolute path
generated pytest file path will be cached in pytest_files_made_cache_mapping
从绝对路径中制作testcase 生成的pytest换存在pytest_files_made_cache_mapping
Args:
tests_path: should be in absolute path
"""
#所有的test_file被加入到队列中
logger.info(f"make path: {tests_path}")
test_files = []
if os.path.isdir(tests_path):
files_list = load_folder_files(tests_path)
test_files.extend(files_list)
elif os.path.isfile(tests_path):
test_files.append(tests_path)
else:
raise exceptions.TestcaseNotFound(f"Invalid tests path: {tests_path}")
for test_file in test_files:
#如果已经是pytest文件直接加入到待跑序列
if test_file.lower().endswith("_test.py"):
pytest_files_run_set.add(test_file)
continue
#将yaml/json转化为字典
try:
test_content = load_test_file(test_file)
except (exceptions.FileNotFound, exceptions.FileFormatError) as ex:
logger.warning(f"Invalid test file: {test_file}\n{type(ex).__name__}: {ex}")
continue
#一些检查确保格式正确
if not isinstance(test_content, Dict):
logger.warning(
f"Invalid test file: {test_file}\n"
f"reason: test content not in dict format."
)
continue
# api in v2 format, convert to v3 testcase
if "request" in test_content and "name" in test_content:
test_content = ensure_testcase_v3_api(test_content)
if "config" not in test_content:
logger.warning(
f"Invalid testcase/testsuite file: {test_file}\n"
f"reason: missing config part."
)
continue
elif not isinstance(test_content["config"], Dict):
logger.warning(
f"Invalid testcase/testsuite file: {test_file}\n"
f"reason: config should be dict type, got {test_content['config']}"
)
continue
# ensure path absolute
test_content.setdefault("config", {})["path"] = test_file
#dict类型的testcase和testsuite转化为pytest
# testcase
if "teststeps" in test_content:
try:
testcase_pytest_path = make_testcase(test_content)
pytest_files_run_set.add(testcase_pytest_path)
except exceptions.TestCaseFormatError as ex:
logger.warning(
f"Invalid testcase file: {test_file}\n{type(ex).__name__}: {ex}"
)
continue
# testsuite
elif "testcases" in test_content:
try:
make_testsuite(test_content)
except exceptions.TestSuiteFormatError as ex:
logger.warning(
f"Invalid testsuite file: {test_file}\n{type(ex).__name__}: {ex}"
)
continue
# invalid format
else:
logger.warning(
f"Invalid test file: {test_file}\n"
f"reason: file content is neither testcase nor testsuite"
)
-parse.py 解析testcase中的各种变量/function/
from sentry_sdk import capture_exception #错误日志监控
def parse_string_value(str_value: Text) -> Any:
""" parse string to number if possible
e.g. "123" => 123
"12.2" => 12.3
"abc" => "abc"
"$var" => "$var"
"""
try:
return ast.literal_eval(str_value)#将数据还原成它可以转化的类型
将原始带变量和函数的字符串转化为计算后的字符串
def parse_string(
raw_string: Text,
variables_mapping: VariablesMapping,
functions_mapping: FunctionsMapping,
) -> Any:
-runner.py
调用钩子函数 参数为勾函数(${func()}或是{"var": "${func()}"})，step_variables（请求/响应）,hook_msg
解析 调用parse_data函数 将其返回函数运行值
def __call_hooks(
self, hooks: Hooks, step_variables: VariablesMapping, hook_msg: Text,
) -> NoReturn:
运行teststep中的请求
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:]}",
)
step.variables["request"] = parsed_request_dict
# setup hooks
if step.setup_hooks:
self.__call_hooks(step.setup_hooks, step.variables, "setup request")
# prepare arguments
method = parsed_request_dict.pop("method")
url_path = parsed_request_dict.pop("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
未完。。。
|