Python自动化测试必备：retry库的5种实战用法与避坑指南
Python自动化测试进阶：告别Flaky Test，用retry库构建稳定测试体系
最近在重构一个大型Web应用的自动化测试套件时，我又遇到了那个老对手——Flaky Test（不稳定的测试）。一个在本地运行十次能成功九次的测试用例，到了持续集成环境中，可能因为网络波动、页面加载延迟或者第三方接口的瞬时不可用，就随机失败。这种“时好时坏”的测试不仅消耗团队大量时间去排查，更严重的是，它会逐渐侵蚀我们对自动化测试结果的信任。过去，我们习惯在代码里到处写try...except加上while循环，代码变得冗长且难以维护。直到系统性地应用了retry库，才真正把我们从这种重复劳动中解放出来，让测试脚本的健壮性上了一个台阶。这篇文章，我就结合这几年在电商、金融等多个领域的测试实战经验，和你深入聊聊retry库那些超越基础用法的实战技巧，以及如何避开常见的“坑”。
1. 理解核心：为什么自动化测试需要智能重试？
在理想世界里，我们的测试环境应该是稳定、隔离且可预测的。但现实是，尤其是在集成测试和端到端测试中，我们不得不面对诸多不确定性。元素定位失败，可能是因为页面尚未完全渲染；API调用超时，可能是网关的瞬时抖动；数据库查询无结果，可能是数据同步的延迟。这些瞬态故障是Flaky Test的主要根源。
传统的异常处理加循环重试，虽然直接，但带来了几个显著问题：
- 代码重复：每个可能需要重试的函数都要包裹几乎相同的
try-except和循环逻辑，违反了DRY原则。 - 策略单一：通常只是简单的“失败后立即重试N次”，缺乏等待间隔、退避策略等，可能加剧服务器压力或在错误发生时疯狂重试。
- 可观测性差：重试行为隐藏在业务逻辑中，难以统一监控和记录，出了问题不好排查。
retry库的价值，就在于它以一种声明式的方式，将重试这一横切关注点从业务逻辑中剥离。你只需要用一个装饰器告诉函数：“当遇到某种异常时，请按照这样的策略重试”，剩下的就交给库来处理。这不仅让代码更简洁，更重要的是，它让重试策略变得可配置、可复用、可观测。
提示：将重试视为一种弹性设计模式，而不是简单的错误补救。它的目标是在面对预期内的瞬时故障时，保持系统的可用性和测试的稳定性。
2. 从入门到精通：retry库的5种核心参数组合实战
安装retry库非常简单，pip install retry即可。它的核心是@retry装饰器，其威力全在于参数的灵活组合。下面我们跳出简单的示例，直接看五种在自动化测试中最具代表性的参数组合方案。
2.1 方案一：基础容错 - 应对元素定位的随机失败
在UI自动化中，ElementNotVisibleException或NoSuchElementException是最常见的。一个粗暴的固定等待time.sleep(10)效率低下，而WebDriverWait配合expected_conditions是首选，但有时对于复杂的、非标准的交互，我们仍需要在操作层面加入重试。
from retry import retry
from selenium.common.exceptions import NoSuchElementException, StaleElementReferenceException
@retry(exceptions=(NoSuchElementException, StaleElementReferenceException),
tries=4, # 最大重试3次（首次执行+3次重试）
delay=1, # 每次重试前固定等1秒
logger=my_test_logger) # 使用项目自定义的日志器
def click_dynamic_button(driver, button_id):
"""
尝试点击一个可能延迟加载或动态渲染的按钮。
重试策略：遇到元素找不到或元素过期异常时，最多重试3次，每次间隔1秒。
"""
element = driver.find_element_by_id(button_id)
element.click()
return element
关键点解析：
exceptions：这里用元组指定了两种Selenium常见异常。StaleElementReferenceException（元素过期）在单页面应用频繁更新DOM时尤其重要。tries=4：注意，tries指的是总执行次数（首次+重试）。设置4意味着最多尝试4次，即重试3次。logger：强烈建议传入自定义的日志器。retry库在每次重试时都会记录一条WARNING级别的日志，这对于后续分析Flaky Test至关重要。默认的日志器可能不符合你的项目格式。
2.2 方案二：指数退避 - 缓解接口或数据库压力
当测试需要调用外部接口或频繁查询数据库时，瞬时的高负载或网络拥堵可能导致失败。立即重试可能会“雪上加霜”。指数退避是一种优雅的策略，它让每次重试的等待时间按指数增长，给系统恢复留出时间。
import requests
from retry import retry
from requests.exceptions import Timeout, ConnectionError
@retry(exceptions=(Timeout, ConnectionError),
tries=5, # 总尝试5次
delay=2, # 首次重试等待2秒
backoff=2, # 退避因子为2
max_delay=30, # 最长等待不超过30秒
logger=None) # 本例中我们可能在函数内更精细地记录
def call_flaky_external_api(url, payload):
"""
调用一个可能不稳定的外部API。
重试策略：超时或连接错误时，重试间隔按2的指数增长（2, 4, 8, 16...秒），但不超过30秒。
"""
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status() # 如果HTTP状态码不是200，会抛出HTTPError，此异常未被@retry捕获，将直接抛出。
return response.json()
等待时间序列计算（假设第一次调用失败）：
- 第1次重试等待：
delay = 2秒 - 第2次重试等待：
delay = 2 * backoff = 4秒 - 第3次重试等待：
delay = 4 * backoff = 8秒 - 第4次重试等待：
delay = 8 * backoff = 16秒 (因为max_delay=30，所以16秒有效) - 如果还需要第5次，等待将是32秒，但被
max_delay限制为30秒。
这种策略非常适合与下游服务的交互测试，能有效避免因测试脚本的密集重试而引发的“惊群效应”。
2.3 方案三：随机抖动 - 避免多个进程同时重试
在并行测试或分布式执行环境中，如果大量测试用例在同一时刻因同一原因（如服务重启）失败，并使用相同的重试策略，它们可能会在完全相同的时刻发起重试，形成一波新的请求洪峰。加入jitter（抖动）可以打散这个时间点。
from retry import retry
from some_db_library import DatabaseError, OperationalError
@retry(exceptions=(DatabaseError, OperationalError),
tries=-1, # 无限重试，直到成功（通常需要结合超时）
delay=1,
jitter=(0, 3), # 在每次计算出的间隔上，增加一个0到3秒之间的随机等待
logger=logger)
def wait_for_database_ready(connection):
"""
等待数据库连接就绪。用于测试环境初始化或恢复。
重试策略：无限重试，基础间隔1秒，并增加随机抖动。
"""
return connection.execute("SELECT 1").fetchone()
jitter参数可以是一个数字（固定加数），也可以是一个(min, max)元组。使用元组时，每次重试会增加一个在该范围内的随机秒数。这能显著降低多个并发进程重试动作的同步性。
2.4 方案四：精准捕获 - 只对特定异常进行重试
不是所有异常都值得重试。像AssertionError（断言失败）通常意味着测试逻辑或预期结果错误，重试没有意义。ValueError可能意味着输入参数永远无效，重试也不会成功。我们必须精确指定需要重试的异常类型。
from retry import retry
import pymysql
from pymysql.err import MySQLError
# 假设我们只关心“锁等待超时”和“死锁”这两种可以重试的数据库错误
RETRIABLE_DB_ERRORS = (1205, 1213) # MySQL error codes for LOCK WAIT TIMEOUT, DEADLOCK
def is_retriable_db_error(exc):
"""自定义判断函数，检查是否为可重试的数据库错误"""
if isinstance(exc, MySQLError):
return exc.args[0] in RETRIABLE_DB_ERRORS
return False
# retry库本身不支持直接传入判断函数，但我们可以通过捕获基类，在函数内再判断并重新抛出不可重试的异常。
@retry(exceptions=MySQLError, tries=3, delay=0.5)
def execute_transaction(sql):
"""
执行数据库事务，仅对特定的可重试错误进行重试。
"""
try:
return db_cursor.execute(sql)
except MySQLError as e:
if not is_retriable_db_error(e):
raise # 如果是不可重试的错误，直接原样抛出，终止重试
logger.warning(f"遇到可重试数据库错误 {e.args[0]}，将进行重试")
raise # 重新抛出，触发retry装饰器的重试逻辑
这个例子展示了更精细的控制：在函数内部捕获异常后，先判断是否属于我们定义的可重试类型，如果不是，则直接raise，这会让retry装饰器停止重试，并将该异常向上传递。这需要你对可能抛出的异常有深入了解。
2.5 方案五：组合日志与监控 - 让重试行为可见
在CI/CD流水线（如Jenkins）中运行测试时，清晰的日志是诊断问题的生命线。retry库的日志需要妥善集成到你的测试框架日志系统中。
import logging
from retry import retry
# 创建一个专门用于记录重试的日志器，可以输出到单独的文件或带有特定格式
retry_logger = logging.getLogger('test.retry')
retry_handler = logging.FileHandler('retry_events.log')
retry_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
retry_logger.addHandler(retry_handler)
retry_logger.setLevel(logging.WARNING) # retry库默认记录WARNING级别
@retry(exceptions=Exception, tries=3, delay=1, logger=retry_logger)
def a_flaky_operation():
# ... 操作逻辑 ...
pass
# 在Jenkins的Post-build Actions中，你可以配置收集`retry_events.log`，
# 并基于“WARNING”日志的数量来设置一个“重试健康度”指标，直观反映测试的稳定性。
Jenkins集成技巧：你可以在Jenkins的构建后步骤中，添加一个脚本，分析本次构建产生的重试日志文件，统计重试事件的总数、分布在哪几个测试用例。甚至可以设置一个质量关卡：如果单个构建中重试事件超过某个阈值，则标记本次构建为“不稳定”，即使所有测试最终都通过了。这能帮助我们提前发现测试环境或应用本身的潜在稳定性问题。
3. 超越retry库：实现自定义超时重试策略
retry库的核心是基于次数的重试。但在自动化测试中，我们有时更需要基于时间的约束。例如：“在30秒内，不断尝试这个操作，直到成功为止，不管中间重试了多少次”。原生retry库无法直接实现这个逻辑，这就需要我们进行自定义扩展。
下面是一个结合了decorator库（让装饰器写法更优雅）和time模块实现的超时重试装饰器：
import time
import functools
from typing import Callable, Any
def retry_on_timeout(timeout_seconds: float, delay_seconds: float = 1.0):
"""
创建一个基于总超时时间的重试装饰器。
:param timeout_seconds: 总超时时间（秒）
:param delay_seconds: 每次重试之间的固定间隔（秒）
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs) -> Any:
deadline = time.time() + timeout_seconds
last_exception = None
attempt = 0
while time.time() < deadline:
attempt += 1
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
remaining = deadline - time.time()
if remaining <= 0:
break
wait_time = min(delay_seconds, remaining) # 确保不会等待超过截止时间
print(f"Attempt {attempt} failed with {type(e).__name__}. "
f"Retrying in {wait_time:.2f}s. Timeout in {remaining:.2f}s.")
time.sleep(wait_time)
# 循环结束（超时），抛出最后一次捕获的异常
raise TimeoutError(f"Function '{func.__name__}' did not succeed within {timeout_seconds} seconds.") from last_exception
return wrapper
return decorator
# 使用示例
@retry_on_timeout(timeout_seconds=30, delay_seconds=2)
def wait_for_condition():
"""模拟一个在特定条件满足后才成功的操作"""
current_time = int(time.time())
if current_time % 10 == 0: # 假设只有当前秒数是10的倍数时才“成功”
return "Success!"
raise RuntimeError("Condition not met yet")
# 这个函数会最多执行30秒，每2秒重试一次，直到`current_time % 10 == 0`为真。
这个自定义装饰器提供了与retry库不同的维度控制。你可以将它和@retry结合使用，实现更复杂的策略，例如“最多重试5次，但总时间不超过30秒”。这只需要在函数上叠加两个装饰器即可，但要注意执行顺序和异常处理的逻辑。
4. 避坑指南：使用retry时必须警惕的陷阱
重试不是银弹，滥用反而会掩盖真正的问题，甚至制造新的问题。以下是一些关键的注意事项：
陷阱一：对非幂等操作进行重试
- 问题：如果被装饰的函数操作不是幂等的（即多次执行会产生不同的副作用），例如“发送邮件”、“创建订单”，重试可能导致重复操作。
- 对策：重试机制应主要用于查询、获取状态、验证等幂等操作。对于非幂等操作，重试逻辑必须与业务层的防重机制（如唯一ID、令牌）紧密结合，或者根本不应该使用自动重试。
陷阱二：无限重试阻塞测试进程
@retry(tries=-1)会无限重试。如果异常条件永远无法恢复，测试用例将永远挂起，占用资源。- 对策：始终设置一个合理的
tries上限，或结合我们上面自定义的timeout机制。在测试框架中，确保有全局的用例超时设置。
陷阱三：异常掩盖与错误报告
- 经过多次重试后最终抛出的异常，是最初的异常还是最后一次的异常？堆栈信息是否清晰？
- 对策：
retry库在达到重试次数后，抛出的是最后一次捕获的异常。这通常是合理的。确保你的日志记录了每次重试的异常信息和时间点，方便回溯。使用logger参数并设置恰当的日志级别。
陷阱四：性能损耗与测试时长
- 不加思考地给所有函数加上重试，尤其是长延迟的重试，会显著增加测试套件的总执行时间。
- 对策：按需使用。为不同的操作定义不同的重试策略。对于快速失败的操作，可以设置较少的重试次数和较短的延迟；对于依赖外部慢速系统的操作，可以适当放宽。在CI流水线中监控测试执行时间的趋势。
陷阱五：与测试框架的断言冲突
- 像
pytest的assert语句失败会抛出AssertionError。如果你用@retry(exceptions=Exception)装饰了一个包含断言测试函数，那么断言失败也会被重试，这完全扭曲了测试的本意——测试失败应该立刻报告。 - 对策：永远不要捕获
AssertionError进行重试。确保你的exceptions参数明确排除了测试框架的断言异常。重试应只用于处理环境或外部依赖的不稳定，而不是测试逻辑本身的错误。
5. 架构视角：将重试策略融入测试框架
对于大型项目，不应该在每个测试函数上零散地添加@retry装饰器。更好的做法是在框架层面进行集成，提供统一、可配置的重试能力。
思路一：创建公共工具模块 建立一个retry_policies.py模块，预定义几种常用的重试策略装饰器：
# retry_policies.py
from retry import retry
from requests.exceptions import Timeout, ConnectionError
from selenium.common.exceptions import WebDriverException
ui_retry = retry(exceptions=WebDriverException, tries=3, delay=1, logger=ui_logger)
api_retry = retry(exceptions=(Timeout, ConnectionError), tries=4, delay=2, backoff=2, logger=api_logger)
db_retry = retry(exceptions=(DatabaseError,), tries=5, delay=0.5, jitter=1, logger=db_logger)
# 在测试用例中直接使用预定义的策略
@ui_retry
def test_login_ui():
# ... UI测试逻辑 ...
@api_retry
def test_checkout_api():
# ... API测试逻辑 ...
思路二：与pytest插件结合 如果你使用pytest，可以创建一个插件，通过pytest_runtest_setup钩子或自定义标记（mark）来为标记的测试用例自动添加重试逻辑。
# conftest.py
import pytest
from retry import retry
def pytest_configure(config):
config.addinivalue_line("markers", "flaky: Mark test as flaky and apply retry logic.")
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
"""在测试用例执行时，如果标记了@flaky，则用重试逻辑包裹执行"""
if item.get_closest_marker("flaky"):
original_func = item.obj
retry_decorator = retry(exceptions=Exception, tries=3, delay=1)
item.obj = retry_decorator(original_func)
yield
# 在测试用例文件中
import pytest
@pytest.mark.flaky
def test_unstable_external_service():
# 这个测试会自动获得重试能力
assert call_service() == "expected"
这种方式将重试策略的配置与测试用例的实现完全解耦，策略的调整只需要在框架配置中修改一处即可，维护性大大增强。
最后，记住重试的目的是提高测试的稳定性和可靠性，而不是让一个本身有问题的测试勉强通过。每次看到重试日志被触发，都应该是一个信号，促使我们去思考：是测试环境需要优化？是应用程序的稳定性有待提升？还是测试用例本身过于脆弱需要重构？善用retry，让它成为你构建坚如磐石的自动化测试体系的得力工具，而不是掩盖问题的“创可贴”。在实际项目中，我从一个到处写try...except循环的新手，到开始滥用@retry，再到如今谨慎、有针对性地使用它，这个过程中最大的收获就是：最优雅的代码，往往来自于对问题本质的深刻理解，而非对工具的生搬硬套。 希望这些实战经验和避坑思路，能帮助你在自己的测试工程中更有效地驾驭重试机制。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
6
0- 0
已为社区贡献2条内容
所有评论(0)