Parallel Pytest Fixtures Processing

user_13293690 0
Viewed 221

I am using PyTest, and suppose I have a test function named test_func, which has two arguments named docker_mysql and docker_clickhouse, which are functions-fixtures in PyTest terminology. In this two functions I am pulling docker image and start it. Due to mechanism of fixtures, test_func calls fixtures step by step, so, the second docker container runs only after the first starts.

So, the question is: how to parallelize process of pulling docker containers to save time, because waiting for many docker containers in queue would be so annoying in the nearest future.

Personally, I've tried pytest-xdist but in CI-CD I didn't mention any changes in velocity. Also, I tried to make fixture that would do a thread using from threading import Thread, but I crashed.

docker pytest fixtures parallel-processing python-3.x
1 Answers

I tried this using Thread and it worked.

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 ============================================================================================

Some notes:

  1. As visible in the logs, both the processes for docker_mysql and docker_clickhouse started at the same time 01:26:21 and ended after the sleep of 5 and 8 seconds respectively (to simulate different execution times).
  2. I used "session" scope as documented here so that it doesn't re-execute the fixture for every test, as visible in the logs after running both test_func and test_func2. Check if this is also applicable to your scenario.
  3. Currently, the format of the fixture's response is a list of <Identifier, Result> e.g. [('docker_mysql', 'MySQL'), ...]. You may want to reformat this e.g. to dict based on your use case.
user_110438250
Related