自动化-Httprunner3源码阅读-Ongoing

打铁读书郎

发布于 2024-04-11 20:58:37

450

发布于 2024-04-11 20:58:37

文章被收录于专栏：暴走的程序质检员暴走的程序质检员

自动化-Httprunner3源码阅读-Ongoing

S背景

我现在的公司目前使用的自动化测试框架为Httprunner3 , 框架本身完备度较高, 但是在实际使用过程中发现一个bug:

一个pytest格式用例,单独运行OK, 整个包一起运行, 一个参数传递为None,导致用例运行失败,修改变量名运行OK

目前判断为框架批量化运行时参数解析代码存在问题,希望能从源码找到原因

运行方式	使用参数txxxxx_project_id	使用参数 txxxxx_project_id_gtpi
单模块运行	PASS	PASS
整个包运行	FAIL	PASS

T目标

研究hrun技术框架
研究hrun运行流程
弄清楚参数传递错误原因

A执行:hrun技术框架

框架功能

可集成pytest
集成request库
支持参数抓取,参数化,断言,钩子函数
debugtalk函数
jmethpath形式处理json数据
可集成allure生成报告
集成locust运行性能测试 –>不涉及
用例格式支持yaml, json,pytest –>不涉及
har文件自动转化用例 –>不涉及

源码框架

目录结构

# 重要结构 #
builtin	
├── comparators.py 	# 内置对比方法
├── functions.py 	# 内置通用方法
init.py
modules.py		# 设定了测试用例相关类
loader.py		# 加载不同类型用例
make.py			# 将用例转化为pytest用例,并执行
cli.py

client.py
parse.py		# 测试数据解析相关   !!!这是重点
runner.py		# httprunner基类
testcases.py	# config, step相关类,用于将测试用例转化为对象

# 不重要结构 #
app  	# fastapi框架
ext		# 第三方框架
├── har2case
├── locust
├── uploader # 用于文件上传接口
__main__.py		# 执行cli中的main() 
compat.py		# 用于兼容httprunner历史用例
exception.py		# 定义用例失败类型
scaffold.py			# 搭建脚手架相关
utils.py		# 工具类

模块顺序: 由底到高

init.py

__version__ = "3.1.11"
__description__ = "One-stop solution for HTTP(S) testing."

# import firstly for monkey patch if needed
from httprunner.parser import parse_parameters as Parameters  	# 数据驱动封装
from httprunner.runner import HttpRunner		
from httprunner.testcase import Config, Step, RunRequest, RunTestCase	# 测试用例结构类

__all__ = [
    "__version__",
    "__description__",
    "HttpRunner",
    "Config",
    "Step",
    "RunRequest",
    "RunTestCase",
    "Parameters",
]

models.py

此模块内定义不同级别用于存放测试数据的类

以下都是部分代码

class TRequest(BaseModel):
    """requests.Request model

    用例请求类: 放置请求信息 
    """

    method: MethodEnum
    url: Url
    params: Dict[Text, Text] = {}
    headers: Headers = {}
    req_json: Union[Dict, List, Text] = Field(None, alias="json")
    data: Union[Text, Dict[Text, Any]] = None
    cookies: Cookies = {}
    timeout: float = 120
    allow_redirects: bool = True
    verify: Verify = False
    upload: Dict = {}  # used for upload files

loader.py

加载器,

将文件中的数据加载为对象 : 包括 pytest测试用例, .env, csv文件,文件夹
将function记载为字典对象: python脚本

def _load_yaml_file(yaml_file: Text) -> Dict:
    """ load yaml file and check file content format
    """
    with open(yaml_file, mode="rb") as stream:
        try:
            yaml_content = yaml.load(stream, Loader=yaml.FullLoader)
        except yaml.YAMLError as ex:
            err_msg = f"YAMLError:\nfile: {yaml_file}\nerror: {ex}"
            logger.error(err_msg)
            raise exceptions.FileFormatError

        return yaml_content
        
def locate_debugtalk_py(start_path: Text) -> Text:
    """ locate debugtalk.py file

    Args:
        start_path (str): start locating path,
            maybe testcase file path or directory path

    Returns:
        str: debugtalk.py file path, None if not found

    """
    try:
        # locate debugtalk.py file.
        debugtalk_path = locate_file(start_path, "debugtalk.py")
    except exceptions.FileNotFound:
        debugtalk_path = None

    return debugtalk_path

make.py

类型转化相关方法

将json,yaml文件转化为 pytest文件
相对路径/绝对路径转化

def __ensure_absolute(path: Text) -> Text:
    # 返回绝对路径

    if path.startswith("./"):
        # Linux/Darwin, hrun ./test.yml
        path = path[len("./"):]
    elif path.startswith(".\\"):
        # Windows, hrun .\\test.yml
        path = path[len(".\\"):]

    path = ensure_path_sep(path)
    project_meta = load_project_meta(path)

    if os.path.isabs(path):
        absolute_path = path
    else:
        absolute_path = os.path.join(project_meta.RootDir, path)

    if not os.path.isfile(absolute_path):
        logger.error(f"Invalid testcase file path: {absolute_path}")
        sys.exit(1)

    return absolute_path

cli.py

hrun 终端指令相关

def main_run(extra_args) -> enum.IntEnum:       # 看起来很重要
    ga_client.track_event("RunAPITests", "hrun")        # 访问GA链接
    # keep compatibility with v2
    extra_args = ensure_cli_args(extra_args)        # 兼容hunv2版本的用例

    tests_path_list = []
    extra_args_new = []
    for item in extra_args:                 # 对extra_args中的链接遍历, 路径存在的放到tests_path_list 中
        if not os.path.exists(item):
            # item is not file/folder path
            extra_args_new.append(item)
        else:
            # item is file/folder path
            tests_path_list.append(item)

    if len(tests_path_list) == 0:           # 未收集到测试用例判断
        # has not specified any testcase path
        logger.error(f"No valid testcase path in cli arguments: {extra_args}")
        sys.exit(1)

    testcase_path_list = main_make(tests_path_list)     # 对路径及文件进行格式化
    if not testcase_path_list:
        logger.error("No valid testcases found, exit 1.")
        sys.exit(1)

    if "--tb=short" not in extra_args_new:
        extra_args_new.append("--tb=short")

    extra_args_new.extend(testcase_path_list)
    logger.info(f"start to run tests with pytest. HttpRunner version: {__version__}")
    return pytest.main(extra_args_new)              # 开始使用pytest进行测试了

client.py

get_req_resp_record():通过request和response对象解析请求响应信息, 并做日志输出

HttpSession: 对requests库中的Session进行二次封装, 并对request配置默认参数

def request(self, method, url, name=None, **kwargs):
"""        Constructs and sends a :py:class:`requests.Request`.
      Returns :py:class:`requests.Response` object.
      """
      self.data = SessionData()

      # timeout default to 120 seconds  请求时间设置
      kwargs.setdefault("timeout", 120)

      # set stream to True, in order to get client/server IP/Port         记录ip端口信息
      kwargs["stream"] = True

      start_timestamp = time.time()
      response = self._send_request_safe_mode(method, url, **kwargs)      # 通过request发送请求
      response_time_ms = round((time.time() - start_timestamp) * 1000, 2)

      try:
          client_ip, client_port = response.raw._connection.sock.getsockname()
          self.data.address.client_ip = client_ip
          self.data.address.client_port = client_port
          logger.debug(f"client IP: {client_ip}, Port: {client_port}")
      except Exception:
          pass

      try:
          server_ip, server_port = response.raw._connection.sock.getpeername()
          self.data.address.server_ip = server_ip
          self.data.address.server_port = server_port
          logger.debug(f"server IP: {server_ip}, Port: {server_port}")
      except Exception:
          pass

      # get length of the response content        获取响应长度
      content_size = int(dict(response.headers).get("content-length") or 0)

      # record the consumed time          记录消耗时间
      self.data.stat.response_time_ms = response_time_ms
      self.data.stat.elapsed_ms = response.elapsed.microseconds / 1000.0
      self.data.stat.content_size = content_size

      # record request and response histories, include 30X redirection        记录请求响应及重定向信息
      response_list = response.history + [response]
      self.data.req_resps = [
          get_req_resp_record(resp_obj) for resp_obj in response_list
      ]

      try:
          response.raise_for_status()                 # 通过status_code 判断是否 raise Exception
      except RequestException as ex:
          logger.error(f"{str(ex)}")
      else:
          logger.info(
              f"status_code: {response.status_code}, "
              f"response_time(ms): {response_time_ms} ms, "
              f"response_length: {content_size} bytes"
          )

      return response

parse.py

主要依靠re库正则表达式, 解析数据

数据格式转换如str2int
相对路径/绝对路径转换
解析各种结构数据, 将变量和函数进行参数替换

def parse_data(
    raw_data: Any,
    variables_mapping: VariablesMapping = None,
    functions_mapping: FunctionsMapping = None,
) -> Any:
    """ parse raw data with evaluated variables mapping.                    # 解析字符串,集合 中的 string 交给 pasrse_string,将变量解析为传递参数
        Notice: variables_mapping should not contain any variable or function.
    """
    if isinstance(raw_data, str):
        # content in string format may contains variables and functions       # 解析string中的变量和函数
        variables_mapping = variables_mapping or {}
        functions_mapping = functions_mapping or {}
        # only strip whitespaces and tabs, \n\r is left because they maybe used in changeset
        raw_data = raw_data.strip(" \t")
        return parse_string(raw_data, variables_mapping, functions_mapping)

    elif isinstance(raw_data, (list, set, tuple)):
        return [
            parse_data(item, variables_mapping, functions_mapping) for item in raw_data         # 如果为集合那就递归, 逐个解析
        ]

    elif isinstance(raw_data, dict):                       # 如果data是字典, 分别解析key和value, 最终返回一个 解析完后的字典
        parsed_data = {}
        for key, value in raw_data.items():
            parsed_key = parse_data(key, variables_mapping, functions_mapping)
            parsed_value = parse_data(value, variables_mapping, functions_mapping)
            parsed_data[parsed_key] = parsed_value

        return parsed_data

    else:
        # other types, e.g. None, int, float, bool
        return raw_data