httprunner3.x源码阅读

-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

未完。。。