如何确保Behave测试套件中止时仍执行指定清理代码?
问题
我有一个Behave测试套件,启动时会在Docker容器中运行我的应用。我使用atexit添加了代码,以确保套件“正常”退出时停止并移除容器:
def handle_exit(): context.container.stop() context.docker_client.close() atexit.register(handle_exit)
只要测试套件正常退出,即使测试失败,这段代码都能正常工作。但有时测试会被中止:
HOOK-ERROR in before_all: FileNotFoundError: [Errno 2] No such file or directory: '/home/peter/Nextcloud/PycharmProjects/boardgamelibrary/boardgamelibrary' HOOK-ERROR in after_all: AttributeError: 'Context' object has no attribute 'driver' ABORTED: By user.
这种情况下容器不会被清理,后续运行套件的尝试会失败。
是否存在比atexit更“可靠”的工具,能让我的handle_exit例程在所有退出/中止场景下都运行?
需要说明的是,这不是异常处理的问题,或者说相关异常已由Behave内部处理,我没有位置插入try-catch块。
解决方案
1. 信号处理(覆盖用户中断/进程终止场景)
Python的signal模块可以捕获Ctrl+C、进程终止等信号,覆盖atexit处理不到的异常退出场景。同时保留atexit处理正常退出,双重保障:
import signal import atexit def cleanup_resources(): # 先做属性存在性检查,避免清理时触发新报错 if hasattr(context, 'container'): context.container.stop() if hasattr(context, 'docker_client'): context.docker_client.close() # 注册正常退出的清理 atexit.register(cleanup_resources) # 定义信号处理逻辑 def handle_signal(signum, frame): cleanup_resources() exit(1) # 注册常见中断信号 signal.signal(signal.SIGINT, handle_signal) # 捕获Ctrl+C signal.signal(signal.SIGTERM, handle_signal) # 捕获进程终止指令 signal.signal(signal.SIGABRT, handle_signal) # 捕获中止信号
2. Behave钩子兜底清理
在environment.py的钩子中添加防御性清理逻辑,即使前置钩子报错,也尝试执行资源回收:
def before_all(context): # 初始化容器与客户端的代码 context.docker_client = ... context.container = ... def cleanup(): try: if hasattr(context, 'container'): context.container.stop() if hasattr(context, 'docker_client'): context.docker_client.close() except Exception: # 忽略清理时的异常,避免阻塞退出流程 pass # 同时注册atexit和信号处理 import atexit import signal atexit.register(cleanup) def handle_signal(signum, frame): cleanup() exit(1) signal.signal(signal.SIGINT, handle_signal) signal.signal(signal.SIGTERM, handle_signal) def after_all(context): # 兜底清理,即使before_all报错也执行 try: if hasattr(context, 'container'): context.container.stop() if hasattr(context, 'docker_client'): context.docker_client.close() except Exception: pass
3. 外部前置清理脚本
如果上述方法仍有遗漏,可以写一个前置清理脚本,每次运行测试前先清除残留容器:
编写pre_run_cleanup.py:
import docker def cleanup_stale_containers(): client = docker.from_env() # 根据容器名称/标签筛选测试专用容器 for container in client.containers.list(filters={"name": "your-test-container"}): try: container.stop() container.remove() except Exception: pass if __name__ == "__main__": cleanup_stale_containers()
运行测试时先执行脚本:
python pre_run_cleanup.py && behave
内容的提问来源于stack exchange,提问作者spierepf
相关产品推荐
相关产品推荐

