Python 装饰器在测试中的 5 大妙用:让自动化更智能、更健壮

别再只把装饰器当“语法糖”！在测试领域，它是提升效率、增强稳定性的秘密武器。

import functoolsfrom faker import Fakerfrom threading import Lock_faker = Faker("zh_CN")_used_values = set()_lock = Lock()def _ensure_unique(value: str) -> str:    with _lock:        if value not in _used_values:            _used_values.add(value)            return value        # 自动加后缀保证唯一        for i in range(1, 100):            candidate = f"{value}_{i}"            if candidate not in _used_values:                _used_values.add(candidate)                return candidate    raise RuntimeError("无法生成唯一值")def generate_test_data(**fields):    """    自动生成测试数据并注入 test_data 参数    示例: @generate_test_data(username=True, email=True)    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            test_data = {}            for field, enabled in fields.items():                if not enabled:                    continue                if field == "username":                    raw = _faker.user_name()                    test_data[field] = _ensure_unique(f"auto_{raw}")                elif field == "email":                    raw = _faker.free_email()                    test_data[field] = _ensure_unique(f"auto.{raw}")                elif field == "phone":                    test_data[field] = _faker.phone_number()            return func(*args, test_data=test_data, **kwargs)        return wrapper    return decorator

@generate_test_data(username=True, email=True, phone=True)def test_user_register(test_data):    resp = api.register(        username=test_data["username"],        email=test_data["email"],        phone=test_data["phone"]    )    assert resp.status_code == 201

import timeimport pytestdef performance_check(max_time: float = 1.0):    """    监控接口响应时间，超过阈值则失败    :param max_time: 最大允许耗时（秒）    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            start = time.time()            result = func(*args, **kwargs)            duration = time.time() - start            if duration > max_time:                pytest.fail(f"性能超标！耗时 {duration:.2f}s > {max_time}s")            print(f"⏱️ {func.__name__} 耗时: {duration:.2f}s")            return result        return wrapper    return decorator

@performance_check(max_time=0.5)def test_search_api():    resp = api.search(keyword="手机")    assert resp.status_code == 200

import timeimport loggingdef retry_on_failure(max_retries: int = 3, delay: float = 1.0, exceptions=(AssertionError,)):    """    失败时自动重试    :param max_retries: 最大重试次数    :param delay: 重试间隔（秒）    :param exceptions: 触发重试的异常类型    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            last_exception = None            for attempt in range(max_retries + 1):                try:                    return func(*args, **kwargs)                except exceptions as e:                    last_exception = e                    if attempt < max_retries:                        logging.warning(f"⚠️ {func.__name__} 第 {attempt+1} 次失败，{delay}s 后重试...")                        time.sleep(delay)                    else:                        logging.error(f"❌ {func.__name__} 重试 {max_retries} 次后仍失败")            raise last_exception        return wrapper    return decorator

@retry_on_failure(max_retries=2, delay=2.0)def test_external_payment():    resp = requests.post("https://payment-gateway.com/pay", ...)    assert resp.json()["status"] == "success"

def require_env(*services):    """    确保依赖服务可用，否则跳过用例    示例: @require_env("mysql", "redis")    """    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            for service in services:                if not _is_service_healthy(service):                    pytest.skip(f"依赖服务 {service} 不可用，跳过 {func.__name__}")            return func(*args, **kwargs)        return wrapper    return decoratordef _is_service_healthy(service_name: str) -> bool:    """检查服务健康状态（示例）"""    if service_name == "mysql":        try:            # 尝试连接数据库            return True        except:            return False    elif service_name == "redis":        # 检查 Redis 连接        return True    return True  # 默认认为健康

@require_env("mysql", "payment_gateway")def test_order_create():    # 仅当 MySQL 和支付网关正常时执行    ...

# 全局记录用例结果_test_results = {}def depends_on(*test_names):    """标记当前用例依赖其他用例成功"""    def decorator(func):        func._dependencies = test_names        @functools.wraps(func)        def wrapper(*args, **kwargs):            for dep in test_names:                if _test_results.get(dep) != "passed":                    pytest.skip(f"依赖用例 {dep} 未通过，跳过 {func.__name__}")            result = func(*args, **kwargs)            _test_results[func.__name__] = "passed"            return result        return wrapper    return decorator# 在 conftest.py 中记录结果def pytest_runtest_logreport(report):    if report.when == "call":        _test_results[report.nodeid.split("::")[-1]] = "passed" if report.passed else "failed"

def test_login():    # 登录逻辑    assert login_success    # 自动记录为 passed@depends_on("test_login")def test_create_order():    # 仅当 test_login 成功时执行    ...