Created
Jul 28, 2024 05:22 AM
Favorite
Favorite
Priority
备注
推荐
🌟🌟🌟🌟
类型
测试技术
1测试前置与后置操作 (setup 和 teardown)
虽然这不是直接的装饰器应用，但在pytest等框架中，使用@pytest.fixture定义的固定件（fixtures）可以作为测试函数的前置和后置操作。
import pytest
@pytest.fixture
def setup_teardown():
# 前置操作：如登录、初始化环境
print("测试开始前的准备工作")
yield
# 后置操作：如登出、清理环境
print("测试结束后的清理工作")
def test_example(setup_teardown):
print("执行测试用例")
条件跳过测试 (skip 和 skipif)
@pytest.mark.skip(reason)：无条件跳过测试。
@pytest.mark.skipif(condition, reason)：根据条件跳过测试
import pytest
@pytest.mark.skip(reason="暂时不执行此测试")
def test_skip_example():
pass
@pytest.mark.skipif(sys.version_info < (3, 7), reason="仅适用于Python 3.7及以上版本")
def test_skipif_example():
pass
测试失败重试 (retry)
对于不稳定的服务，可能会希望在测试失败时自动重试。
import pytest
from pytest import Retry
@pytest.mark.flaky(reruns=2, reruns_delay=2)
def test_retry_example():
# 假设这是一个有时会失败的测试
assert False
注意：pytest-rerunfailures插件需要安装并使用才能支持reruns。
参数化测试 (parametrize)
允许一个测试用例以不同的参数多次运行。
import pytest
@pytest.mark.parametrize("input_value, expected", [
(3, 6),
(4, 8),
(5, 10)
])
def test_parametrize_example(input_value, expected):
assert input_value * 2 == expected
计时测试 (timer)
监控测试用例的执行时间。
import time
import pytest
def timer(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"{func.__name__} 用时: {end_time - start_time}秒")
return result
return wrapper
@pytest.mark.usefixtures("timer")
def test_timer_example():
time.sleep(1) # 模拟耗时操作
错误处理装饰器
自动捕获并处理测试中的异常，可能用于发送错误报告或进行特定的错误处理逻辑。
def error_handler(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
print(f"测试中发生错误: {e}")
# 这里可以添加错误上报逻辑
return False
return wrapper
@error_handler
def test_with_error_handling():
raise ValueError("模拟错误")
日志增强装饰器
增强测试函数的日志记录能力，记录更多的上下文信息。
import logging
def log_decorator(logger=logging.getLogger(__name__)):
def decorator(func):
def wrapper(*args, **kwargs):
logger.info(f"开始执行: {func.__name__}")
result = func(*args, **kwargs)
logger.info(f"{func.__name__} 执行完成")
return result
return wrapper
return decorator
@log_decorator()
def test_with_logging():
print("这是一个测试函数")
事务管理装饰器
确保数据库操作的原子性，用于自动回滚事务。
from contextlib import contextmanager
@contextmanager
def db_transaction(session):
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
def transactional_test(session):
@db_transaction(session)
def inner_test(session):
# 执行数据库操作
pass
inner_test(session)
API认证装饰器
自动处理API请求的认证信息，如Token注入。
def auth_required(token):
def decorator(func):
def wrapper(*args, **kwargs):
headers = kwargs.get('headers', {})
headers['Authorization'] = f"Bearer {token}"
kwargs['headers'] = headers
return func(*args, **kwargs)
return wrapper
return decorator
@auth_required("your_token_here")
def test_api_endpoint(headers=None):
# 使用headers发起API请求
pass
响应验证装饰器
自动验证API响应的状态码或内容。
def response_validator(status_code=200, content_checker=None):
def decorator(func):
def wrapper(*args, **kwargs):
response = func(*args, **kwargs)
assert response.status_code == status_code, f"状态码不匹配，期望：{status_code} 实际：{response.status_code}"
if content_checker:
assert content_checker(response.json()), "响应内容验证失败"
return response
return wrapper
return decorator
@response_validator(status_code=200, content_checker=lambda json: json.get('success'))
def test_api_response():
# 发起API请求并返回响应
pass
并发测试装饰器
用于模拟多线程或多进程环境下接口的并发测试。
import threading
from functools import wraps
def concurrent_test(num_threads=5):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
threads = []
for _ in range(num_threads):
thread = threading.Thread(target=func, args=args, kwargs=kwargs)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
return wrapper
return decorator
@concurrent_test(10)
def test_concurrent_api_call():
# 发起API请求
pass
数据清理装饰器
确保每次测试后数据状态的一致性，用于清理测试产生的数据。
def cleanup_after_test(cleanup_func):
def decorator(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
finally:
cleanup_func()
return wrapper
return decorator
def clean_database():
# 清理数据库操作
pass
@cleanup_after_test(clean_database)
def test_that_makes_changes():
# 执行可能产生需要清理的数据的操作
pass
性能监控装饰器
记录函数执行的时间，并在达到阈值时报警或记录日志。
def performance_monitor(threshold_seconds=1):
def decorator(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
elapsed_time = time.time() - start_time
if elapsed_time > threshold_seconds:
print(f"警告：{func.__name__} 执行超过{threshold_seconds}秒，实际耗时{elapsed_time}秒")
return result
return wrapper
return decorator
@performance_monitor(2)
def slow_function():
time.sleep(3)
状态验证装饰器
在执行测试前检查前置条件是否满足。
def require_state(check_state_func):
def decorator(func):
def wrapper(*args, **kwargs):
if not check_state_func():
print(f"前置条件不满足，跳过测试: {func.__name__}")
return
return func(*args, **kwargs)
return wrapper
return decorator
def is_service_ready():
# 检查服务是否就绪
pass
@require_state(is_service_ready)
def test_dependent_on_service():
# 执行依赖于服务状态的测试
pass
测试标记装饰器
为测试用例添加特定的标签，便于分类和过滤。
import pytest
@pytest.mark.smoke
def test_smoke_example():
# 用于快速验证核心功能的烟雾测试
pass
@pytest.mark.performance
def test_performance_example():
# 性能测试
pass
python高级265
python高级 · 目录
上一篇python 高级装饰器模版