跳至内容
Pytest & 实用工具

Pytest & 实用工具

前提

Pytest 是一个功能强大且灵活的 Python 测试框架,广泛用于单元测试、功能测试以及自动化测试。它以简洁的语法和丰富的插件生态系统著称,能够满足从简单到复杂的测试需求。

pytest 的设计思想,围绕测试用例(测试函数)的发现、执行、管理和报告展开,其灵活性和可扩展性也主要服务于测试用例(测试函数)的高效编写和维护。

务必阅读pytest 文档,了解其使用方法。

超时标记

第三方插件 @pytest.mark.timeout 是一个比较好用的工具,可以捕捉过长的测试时间,在测试用例耗时太长时终止它并标记为 FAILED。我们可以用它测试用例的执行时间以及防止长时间被挂起。读者可以阅读pytest-timeout 2.3.1了解使用方法。

该插件在 Windows 平台上失效,我针对 Windows 平台,编写了简单的超时装饰器。

common/timeout.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
# -*- coding: utf-8 -*-
"""
超时装饰器模块

模块功能:
    提供一个可复用的超时装饰器,为目标函数添加执行时间限制,超时后抛出AssertionError并记录错误日志。
适用场景:
    1. 测试用例执行超时控制,防止单个用例阻塞整体测试流程;
    2. 外部接口调用、耗时操作的执行时间限制;
    3. 需要严格控制执行时长的函数调用场景。
"""

import concurrent.futures
import functools
from typing import Any, Callable

from common.log_config import setup_logger

logger = setup_logger()


def timeout(seconds: int = 10) -> Callable:
    """
    超时装饰器:为函数添加执行时间限制,超时则抛出AssertionError并记录日志。

    装饰器内部通过ThreadPoolExecutor实现超时控制,executor实例作为函数属性单例存在,
    避免重复创建线程池资源。

    Args:
        seconds (int): 超时时间(单位:秒),非负整数,默认为10秒。
    
    Returns:
        Callable: 包装后的目标函数,保留原函数的元信息(名称、文档等)。
    
    Raises:
        AssertionError: 当被装饰函数执行时间超过指定seconds时抛出,包含超时提示信息。
    
    注意:
        1. 装饰器内部的executor为单例ThreadPoolExecutor,max_workers=1,高并发场景下可能存在执行排队;
        2. 超时异常会屏蔽原TimeoutError,通过`from None`切断异常链,仅暴露AssertionError;
        3. 未校验seconds参数的合法性(如负数),传入负数会导致future.result()抛出ValueError。
    """
    if not hasattr(timeout, 'executor'):
        setattr(timeout, 'executor', concurrent.futures.ThreadPoolExecutor(max_workers=1))
    
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            """
            装饰器内部包装函数,执行目标函数并监控超时。

            Args:
                *args (Any): 目标函数的位置参数。
                **kwargs (Any): 目标函数的关键字参数。
            
            Returns:
                Any: 目标函数的执行结果。
            
            Raises:
                AssertionError: 函数执行超时时抛出。
            """
            executor = getattr(timeout, 'executor')
            future = executor.submit(func, *args, **kwargs)
            try:
                return future.result(timeout=seconds)
            except concurrent.futures.TimeoutError:
                error_msg = f"测试用例 {func.__name__} 已执行超过 {seconds} 秒。"
                logger.error(error_msg)
                raise AssertionError(error_msg) from None

        return wrapper

    return decorator
    

使用方法:

test_mark_decorator.py
1
2
3
4
5
from common.timeout import timeout

@timeout(3)
def test_case1():
    assert True

除此之外,还可以查阅pytest插件列表,有很多有用的插件。

文件数据参数化

测试数据硬编码无疑是一个不聪明的行为,测试数据需要在文件中管理。我们来实现一下。

无论使用什么格式文件管理测试数据,只需要根据对应格式文件进行数据预处理,最终再传递给 pytest 的参数化标记器。如:@pytest.mark.parametrize("n,n1,n2", [(1, 2, 3), (3, 4, 5), (5, 6, 7)])

规范文件数据结构

n1,n2,n3,mark
1,2,3.01,user-smoke
2,3,4.01,smoke
2,3,5.01,

预期结果:参数化数据为 ("n1,n2,n3", [(1, 2, 3.01), (2,3,4.01), (2,3,5.01)])(1,2,3.01) 被标记为 usersmoke(2,3,4.01) 被标记为 smoke

实现装饰器

common/parametrize.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
# -*- coding: utf-8 -*-
"""
pytest参数化数据加载模块

模块功能:提供基于文件的pytest参数化装饰器工具,支持从外部文件读取测试数据,
          自动转换为pytest.mark.parametrize所需格式,并支持为测试用例添加自定义pytest标记。
适用场景:自动化测试场景中,需从Excel/CSV等文件批量加载测试数据,
          并为不同测试用例配置自定义pytest标记(如skip、xfail等)的场景。
"""
from typing import Callable, Tuple, Dict, List, cast

import pytest
from pandas import DataFrame

from common.file_data_reader import FileDataReader
from common.log_config import setup_logger

logger = setup_logger()


def _dataframe_to_parametrize_data(df: DataFrame) -> Tuple[str, List[tuple], Dict[int, List[str]]]:
    """
    将包含测试数据的DataFrame转换为pytest.mark.parametrize可直接使用的数据格式

    处理逻辑:提取DataFrame中的参数列名、参数数据,解析'mark'列生成标记映射(标记间用短横线分隔)。

    Args:
        df (DataFrame): 包含测试数据的DataFrame对象,可选包含'mark'列(用于定义pytest标记)。

    Returns:
        Tuple[str, List[tuple], Dict[int, List[str]]]: 元组包含三个核心元素:
            - str: 以逗号分隔的参数名称字符串(如"username,password");
            - List[tuple]: 参数化数据的元组列表,每个元组对应一行测试数据;
            - Dict[int, List[str]]: 行索引到标记列表的映射字典(无标记则为空字典)。

    注意:
        1. DataFrame中的空值会被替换为空字符串,避免参数化时出现None值;
        2. 若DataFrame为空,返回的参数化数据列表也为空,需调用方自行校验。
    """
    mark_col = df.get("mark")
    mark_data = {}
    if mark_col is not None:
        mark_series = mark_col.fillna('').apply(lambda x: [s for s in str(x).split('-') if s])
        mark_data = mark_series.to_dict()

    non_mark_df = df.drop(columns=["mark"], errors="ignore").fillna('')
    parameterized_variables = ",".join(non_mark_df.columns)
    parameterized_data = [tuple(row) for row in non_mark_df.values.tolist()]

    return parameterized_variables, parameterized_data, mark_data


def parametrize(file_path: str, **kwargs) -> Callable[[Callable], Callable]:
    """
    从指定文件加载测试数据,生成pytest参数化装饰器(支持自定义标记)

    核心流程:读取文件数据→转换为DataFrame→解析参数和标记→生成带标记的parametrize装饰器。

    Args:
        file_path (str): 测试数据文件路径(支持FileDataReader可读取的类型:Excel/CSV等);
        **kwargs: 传递给FileDataReader.read()的额外参数(如sheet_name、encoding、sep等)。

    Returns:
        Callable[[Callable], Callable]: pytest的参数化装饰器函数,可直接装饰测试函数。

    Raises:
        ValueError: 当文件中无有效测试数据时抛出;
        FileNotFoundError: 当指定的file_path文件不存在时(由FileDataReader触发);
        IOError: 当文件读取失败时(由FileDataReader触发)。

    注意:
        1. 若mark列中的标记名称不存在于pytest.mark中(如自定义标记未注册),
           该标记会被忽略并记录警告日志;
        2. 数据文件中的空值会被统一替换为空字符串,避免测试函数接收None值导致异常。
    """
    reader = FileDataReader(file_path)
    _, data_frame = reader.read(**kwargs)

    variables, data, marks_map = _dataframe_to_parametrize_data(cast(DataFrame, data_frame))

    if not data:
        error_msg = f"{file_path} 文件中,无测试数据"
        logger.error(error_msg)
        raise ValueError(error_msg)

    def _apply_marks_to_data(
        data_list: List[tuple],
        marks_mapping: Dict[int, List[str]]
    ) -> List:
        """
        为参数化数据列表中的测试用例项绑定对应的pytest标记

        Args:
            data_list (List[tuple]): 原始参数化数据元组列表;
            marks_mapping (Dict[int, List[str]]): 行索引到标记列表的映射字典。

        Returns:
            List[pytest.param]: 包含pytest标记的参数化数据列表,每个元素为pytest.param对象。

        注意:
            若marks_mapping为空,会为所有测试用例生成无标记的pytest.param对象。
        """
        if not marks_mapping:
            logger.debug("没有找到 'mark' 列或该列为空,跳过标记处理。")
            return [pytest.param(*item) for item in data_list]

        marked_params = []
        for index, item in enumerate(data_list):
            current_marks = []
            if index in marks_mapping:
                for mark_name in marks_mapping[index]:
                    if hasattr(pytest.mark, mark_name):
                        current_marks.append(getattr(pytest.mark, mark_name))
                    else:
                        logger.warning(f"标记 '{mark_name}' 不存在,将被跳过。")
            marked_params.append(pytest.param(*item, marks=current_marks))
        
        return marked_params

    final_data = _apply_marks_to_data(data, marks_map)
    return pytest.mark.parametrize(variables, final_data)
    
if __name__ == "__main__":
    pass

使用装饰器

依赖 pandas 的能力,可以自动识别数字的类型,还可以指定列的类型。

cases/test_par.py
1
2
3
4
5
6
7
from common.parametrize import parametrize

@parametrize("data/test.csv", dtype={"n1": str})
def test_pardd(n1,n2,n3):
    assert type(n3) is float
    assert type(n2) is int
    assert type(n1) is str
最后更新于