pytest 夹具和线程同步答案

【问题标题】：pytest fixtures and threads synchronizationspytest 夹具和线程同步
【发布时间】：2019-07-26 01:01:21
【问题描述】：

我正在尝试使用 pytest-xdist 以使我的测试并行运行，问题是每个线程都将转到与所有测试共享的夹具并根据线程数执行它。

这给我带来了一个问题，因为该夹具角色是为我的测试创建数据，一旦它已经创建，我就会得到错误，因为它已经创建（通过 REST）。

conftest.py：

lock = threading.Lock()

@pytest.fixture(scope=session)
def init_data(request):

    lock.acquire()
    try:
        data = []
        if checking_data_not_created():
            data.append(some_rest_command_creating_data_1())
            data.append(some_rest_command_creating_data_2())
    finally:
        lock.release()

    yield data

    lock.acquire()
    try:
        remove_all_data()
    finally:
        lock.release()

tests_class.py：

class classTests():

    def first_test(init_data):
        test body and validation....

     def second_test(init_data):
        test body and validation....

我正在使用命令： pytest -v -n2

假设第一个线程应该运行 first_test() 第二个线程应该运行 second_test() 其中一个总是会失败，因为第一个已经在fixtures部分创建了数据，另一个线程会出现异常，他应该运行的所有测试都将失败。

如您所见，我尝试使用锁来同步线程，但它也不起作用。

知道如何解决这个问题吗？

谢谢。

【问题讨论】：

第二个线程得到了什么异常？
与我的服务器相关的异常，而不是环境，这不是重点。关键是我只想用 scope=session 运行一次夹具
pytest-xdist 使用多个进程，而不是线程！我认为threading.lock 不足以阻止并发访问。
我很难理解，感谢您的回复 Samuel

标签： python multithreading pytest xdist

【解决方案1】：

这种方法不适用于 pytest-xdist，因为它使用多处理而不是多线程，但是它可以使用 --tests-per-worker 选项与pytest-parallel 一起使用，它将使用多个线程运行测试。

在使用以下夹具的多线程 pytest 执行中，数据只会设置一次并清理一次：

conftest.py：

lock = threading.Lock()
threaded_count = 0

@pytest.fixture(scope='session')
def init_data():
    global lock
    global threaded_count

    lock.acquire()
    threaded_count += 1
    try:
        if threaded_count == 1:
            # Setup Data Once
            data = setup_data()
    finally:
        lock.release()

    yield data

    lock.acquire()
    threaded_count -= 1
    try:
        if threaded_count == 0:
            # Cleanup Data Once
            data = cleaup_data()
    finally:
        lock.release()

命令：

pytest -v --tests-per-worker 2

【讨论】：

“这种方法不适用于 pytest-xdist，因为它使用多处理而不是多线程 [...] 它将使用多个线程运行测试。” 如果它使用 @ 987654323@ 而不是multithreading，那么测试应该在多个进程中运行，而不是“多线程”。

【解决方案2】：

请注意，pytest-xdist 不支持 'session' 范围内的固定装置。 pytest-xdist 中的'session' 作用域fixture 实际上是指特定于进程的会话级fixture，因此它会为每个进程单独创建和拆除，并且fixture 的状态不会在进程之间共享。有一些长期存在的建议来为 pytest-xdist 添加真正的会话范围固定装置的锁定和共享，但它们都遇到了许多经典的理由，要不惜一切代价尝试避免多线程或多处理锁和同步，因此它被 pytest-xdist 开发人员严重降低了优先级（这是可以理解的）。

【讨论】：

【解决方案3】：

Python xdist has a way to do this.

我相信你的例子，应用他们的建议应该是这样的：

def create_data():
    data = []
    data.append(some_rest_command_creating_data_1())
    data.append(some_rest_command_creating_data_2())

    return data


@pytest.fixture(scope="session")
def session_data(request, tmp_path_factory, worker_id):
    if worker_id == "master":
        # Not running multiple processes, just create the data.
        data = create_data()
    else:
        # Running multiple processes, manage lockfile.
        root_tmp_dir = tmp_path_factory.getbasetemp().parent

        fn = root_tmp_dir / "data.json"
        with FileLock(str(fn) + ".lock"):
            if fn.is_file():
                # Data has been created, read it in.
                data = json.loads(fn.read_text())
            else:
                # Data not created, create and write it out.
                data = create_data()
                fn.write_text(json.dumps(data))

    yield data

    remove_all_data()

与您的示例不同，此示例通过保存在共享临时测试目录tmp_path_factory.getbasetemp().parent（即provided with pytest）中的锁定文件提供data。

【讨论】：