python 接口自动化测试常用的装饰器函数
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
感谢每一个认真阅读我文章的人，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走：
这些资料，对于【软件测试】的朋友来说应该是最全面最完整的备战仓库，这个仓库也陪伴上万个测试工程师们走过最艰难的路程，希望也能帮助到你!有需要的小伙伴可以点击下方小卡片领取
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献14条内容
所有评论(0)