时间戳

首页 > 解决方案 > 并行 Pytest 夹具处理

python-3.x - 并行 Pytest 夹具处理

问题描述

我正在使用 PyTest，假设我有一个名为 的测试函数test_func，它有两个名为docker_mysql和的参数docker_clickhouse，它们是 PyTest 术语中的函数夹具。在这两个函数中，我正在拉取 docker 映像并启动它。由于fixtures的机制，test_func一步步调用fixtures，所以，第二个docker容器只有在第一个启动后才会运行。

所以，问题是：如何并行化拉取 docker 容器的过程以节省时间，因为在不久的将来等待许多 docker 容器在队列中会很烦人。

就个人而言，我尝试过pytest-xdist但在 CI-CD 中我没有提到速度的任何变化。另外，我尝试制作可以使用 执行线程的夹具from threading import Thread，但我崩溃了。

标签： python-3.xdockerparallel-processingpytestfixtures

解决方案

我使用 Thread 进行了尝试，并且成功了。

from datetime import datetime
from queue import Queue
from threading import Thread
import time

import pytest


def _pull_and_start_docker_mysql():
    print(datetime.now(), "Start downloading docker_mysql")
    time.sleep(5)  # Simulate download
    print(datetime.now(), "Finished downloading docker_mysql")
    return "MySQL"

def _pull_and_start_docker_clickhouse():
    print(datetime.now(), "Start downloading docker_clickhouse")
    time.sleep(8)  # Simulate download
    print(datetime.now(), "Finished downloading docker_clickhouse")
    return "Clickhouse"


@pytest.fixture(scope="session")
def docker():
    results = Queue()  # Needed to store the results of the threads
    p1 = Thread(target=lambda: results.put(("docker_mysql", _pull_and_start_docker_mysql())))  # To identify the result, we put an identifier name "docker_mysql"
    p2 = Thread(target=lambda: results.put(("docker_clickhouse", _pull_and_start_docker_clickhouse())))  # To identify the result, we put an identifier name "docker_clickhouse"
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    return [results.get() for _ in range(results.qsize())]


def test_func(docker):
    print("Inside test_func", docker)
    assert True


def test_func2(docker):
    print("Inside test_func2", docker)
    assert True

$ pytest -rP
=========================================================================================== test session starts ===========================================================================================
collected 2 items                                                                                                                                                                                         

test_docker.py ..                                                                                                                                                                                    [100%]

================================================================================================= PASSES ==================================================================================================
________________________________________________________________________________________________ test_func ________________________________________________________________________________________________
------------------------------------------------------------------------------------------ Captured stdout setup ------------------------------------------------------------------------------------------
2021-08-17 01:26:21.113417 Start downloading docker_mysql
2021-08-17 01:26:21.113675 Start downloading docker_clickhouse
2021-08-17 01:26:26.125081 Finished downloading docker_mysql
2021-08-17 01:26:29.122935 Finished downloading docker_clickhouse
------------------------------------------------------------------------------------------ Captured stdout call -------------------------------------------------------------------------------------------
Inside test_func [('docker_mysql', 'MySQL'), ('docker_clickhouse', 'Clickhouse')]
_______________________________________________________________________________________________ test_func2 ________________________________________________________________________________________________
------------------------------------------------------------------------------------------ Captured stdout call -------------------------------------------------------------------------------------------
Inside test_func2 [('docker_mysql', 'MySQL'), ('docker_clickhouse', 'Clickhouse')]
============================================================================================ 2 passed in 8.04s ============================================================================================

一些注意事项：

  1. 从日志中可以看出，进程在 01:26:21docker_mysql同时docker_clickhouse开始，分别在 5 秒和 8 秒的睡眠后结束（模拟不同的执行时间）。
  2. 我使用了此处"session"记录的范围，因此它不会为每个测试重新执行夹具，正如在运行和后在日志中可见的那样。检查这是否也适用于您的场景。test_functest_func2
  3. 目前，夹具响应的格式是 <Identifier, Result> 的列表，例如[('docker_mysql', 'MySQL'), ...]。您可能需要重新格式化它，例如根据您的用例进行 dict。

推荐阅读