英文:
Pytest-xdist: tearDown after all workers finish
问题
当我正常运行pytest时，它运行得非常好（在第一个测试之前升级迁移，在最后一个测试之后降级迁移）。不使用xdist时，我只需将fixture范围设置为'session'，它就可以正常工作。但是当我运行pytest-xdist（并行测试）时，它在每个测试之前升级迁移，并在每个测试之后降级迁移（我希望在第一个测试之前进行setUp - 第一个worker，并在最后一个测试之后进行tearDown - 当最后一个worker完成时）。
我认为在第一个worker之前进行setUp，我可以使用文件锁来处理（而且我不太需要它，因为在迁移升级中，我只是在需要时创建一些表，因此如果已经升级，它就不会执行任何操作）。
我特别关注最后一个worker完成后的tearDown。
import pytest
import os
import subprocess
@pytest.fixture(autouse=True, scope='session')
def handle_database():
"""在数据库中创建必要的表（使用迁移），并在测试后删除它们"""
# 切换到 test/api_simulator 目录
os.chdir('test/api_simulator')
# 执行迁移升级
subprocess.run(['flask', 'db', 'upgrade'])
print('\n\n升级迁移\n\n')
# 切换回原始目录
os.chdir('../..')
yield
# 切换到 test/api_simulator 目录
os.chdir('test/api_simulator')
# 执行迁移降级
subprocess.run(['flask', 'db', 'downgrade'])
print('\n\n降级迁移\n\n')
# 切换回原始目录
os.chdir('../..')
我在网上搜索了这个问题，但在使用xdist的情况下找不到什么有关tearDown的好解决方案。
英文:
When I'm running pytest normally it works perfectly fine (before first test it's upgrading migration and after last test it's downgrading migration). Without using xdist I can just set fixture scope='session' and its work that way. But when I run pytest-xdist (tests in parallel) it's upgrading migration before each test and downgrading migration after each test (I want to setUp before first test - first worker, and tearDown after last test - when last worker finishes).
I think that setUp before first worker I can handle with File Lock (and also I don't need it that much because in migration upgrade I'm just creating some tables if not exist so if it was already upgraded it just won't do anything).
I'm interested especially in tearDown after last worker finshes.
import pytest
import os
import subprocess
@pytest.fixture(autouse=True, scope='session')
def handle_database():
"""Create necesarry tables in db (using migration) and drop them after tests"""
# Change directory to test/api_simulator
os.chdir('test/api_simulator')
# Perform migration upgrade
subprocess.run(['flask', 'db', 'upgrade'])
print('\n\nUpgrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
yield
# Change directory to test/api_simulator
os.chdir('test/api_simulator')
# Perform migration downgrade
subprocess.run(['flask', 'db', 'downgrade'])
print('\n\nDowngrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
I was searching about this issue in web but I found nothing good about tearDown with xdist.
答案1
得分: 1
fixture 是作为测试的一部分运行的，即使它只在会话范围内运行一次。这意味着它在 xdist 打开的进程中运行。
要仅运行一次，请将代码移到 conftest.py 中的 pytest_sessionstart 和 pytest_sessionfinish。这些方法将在 "常规" 范围内在 xdist 启动之前/之后执行一次。请注意，pytest_session 函数将再次作为测试周期的一部分运行，因此您需要检查它。
def __handle_database(config, command):
if not hasattr(config, 'workerinput'):
os.chdir('test/api_simulator')
# 执行迁移升级
subprocess.run(['flask', 'db', command])
print('\n\n升级迁移\n\n')
# 切换回原始目录
os.chdir('../..')
def pytest_sessionstart(session):
__handle_database(session.config, 'upgrade')
def pytest_sessionfinish(session, exitstatus):
__handle_database(session.config, 'downgrade')
英文:
The fixture is running as part of a test, even if it's only once in session scope. This means it's running in the process xdist opened.
To run this only once you move the code to pytest_sessionstart and pytest_sessionfinish in conftest.py. Those methods will be executed once in
"regular" scope before/after xdist kicks in. Note that the pytest_session functions will run again as part of the tests cycle so you need to check it
def __handle_database(config, command):
if not hasattr(config, 'workerinput'):
os.chdir('test/api_simulator')
# Perform migration upgrade
subprocess.run(['flask', 'db', command])
print('\n\nUpgrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
def pytest_sessionstart(session):
__handle_database(session.config, 'upgrade')
def pytest_sessionfinish(session, exitstatus):
__handle_database(session.config, 'downgrade')
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论