Pytest multi process / multi thread execution test case

Preface ：

• The number of use cases in the actual project will be very large , Hundreds, thousands ; If the Single process serial Implementation will be very time-consuming . Assume that each use case takes 2s,1000 This article is needed 2000s $\approx$ 33min; Plus use case loading 、 Before testing / The post kit and so on take time ; As a result, the efficiency of test execution will be relatively low .
• Imagine if development changes a piece of code , We need to return , At this time, it takes more than half an hour or several hours to execute the automated use case , This is something we cannot tolerate .
• To save project testing time , Multiple test cases are required at the same time Parallel execution ; This is a kind of Distributed scenarios To shorten the execution time of test cases , Increase of efficiency .

Principles of distributed execution of use cases ：

• Between use cases is Mutually independent , No dependency , It can run independently ;
• Use case execution does not Sequence requirements , Random sequence can be executed normally ;
• Every use case can Run repeatedly , The results of the run do not affect other use cases .

Project structure

The test script

# test1/test_1.py
import time

def test1_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test1_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
	
class TestDemo1:
	def test_inner_1(self):
		time.sleep(1)
		assert 1 == 1, "1==1"


class TestDemo2:
	def test_inner_2(self):
		time.sleep(1)
		assert 1 == 1, "1==1"
# test1/inner/test_3.py
import time

def test3_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test3_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
# test2/test_2.py
import time

def test2_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test2_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"
	
# test2/inner/test_3.py
import time

def test4_test1():
	time.sleep(1)
	assert 1 == 1, "1==1"


def test4_test2():
	time.sleep(1)
	assert 1 == 1, "1==1"

Normal execution ： need 8.10s

Multi process execution use case pytest-xdist

install ：

pip install pytest-xdist

many cpu Execute use cases in parallel , Add a... Directly -n Parameters can be , Back num The parameter is the number of parallels , such as num Set to 3

pytest -v -n num

Parameters ：

1. -n auto ： Automatic detection system CPU number
2. -n num ： Specify the number of processor processes to run the test

Parallel execution of multiple processes ： Time consuming 2.66s It greatly shortens the execution time of test cases .

pytest-xdist The principle of distributed testing ：

1. xdist It is similar to the structure of one master and many slaves ,master Be responsible for issuing orders , control slave;slave according to master Command to perform specific test tasks .

2. stay xdist in , The Lord is master, From is workers;xdist Will produce one or more workers,workers All pass master To control , Every worker Equivalent to one mini edition pytest actuator .

3. master Do not perform test tasks , Only right worker All collected use cases are distributed ; Every worker Responsible for executing test cases , Then feed back the execution results to master; from master Count the final test results .

pytest-xdist The flow of distributed testing ：

First step ：master establish worker

1. master stay Test session （test session） Produce one or more before starting worker.

2. master and worker Between is through execnet and gateway To communicate .

3. Actually compile and execute the test code worker It could be a local machine or a remote machine .

The second step ：workers Collect test item cases

1. Every worker It's like a mini pytest actuator .

2. worker Will perform a complete test collection The process .【 The process of collecting all test cases 】

3. And then put the test case's ids Return to master.【ids Indicates the path of the collected test cases 】

4. master Do not execute any test cases .

Be careful ： Distributed testing （pytest-xdist） Mode will not output print Content , because master Do not execute test cases .

The third step ：master testing workers Collected test suite

1. master Receive all worker After collecting the test suite ,master There will be some integrity checks , To make sure that all worker They all collect the same set of test cases （ Including the order ）.

2. If the inspection passes , Will test case ids List into a simple index list , Each index corresponds to the location of a test case in the original test set .

3. The reason this plan works is ： All nodes hold the same set of test cases .

4. And using this method can save bandwidth , because master Just tell me workers The index corresponding to the test case to be executed , Instead of telling the complete test case information .

Step four ：master Distribute test cases

There are four distribution strategies ： Command line arguments --dist=mode Options ( Default load)

• each：master Distribute the complete test index list to each worker, each worker All use cases will be executed once .
  

• load：master Will be about $\frac{1}{n}$ The test cases of are distributed to each worker, The rest of the test cases will wait worker Distribute after executing the test cases ; Each use case will only be used by one worker Do it once .
  

• loadfile：master The strategy of distributing use cases is as follows ids File name in (test_xx.py or xx_test.py) distributed , That is, the test cases in the same test file will only be distributed to one of them worker; It has a certain degree of isolation .
  

• loadscope：master The strategy of distributing use cases is to distribute by scope , Test functions under the same module or in a test class will be distributed to the same worker To execute ; namely py If there is no test class in the file ( Only tests function) Distribute the module to the same worker perform , If there are test classes, the test classes in this file will only be distributed to the same worker perform , Multiple classes may be distributed to multiple worker; You cannot customize the grouping at this time , By category class Grouping takes precedence over by module module grouping .
  

Be careful ： have access to pytest_xdist_make_scheduler This hook To implement custom test distribution logic .
Such as ： Want to distribute test cases at the catalog level ：

from xdist.scheduler import LoadScopeScheduling


class CustomizeScheduler(LoadScopeScheduling):
	def _split_scope(self, nodeid):
		return nodeid.split("/", 1)[0]


def pytest_xdist_make_scheduler(config, log):
	return CustomizeScheduler(config, log)

1. Just on the outermost layer conftest In the inheritance xdist.scheduler.LoadScopeScheduling And rewrite _split_scope Method
2. Rewrite hook function pytest_xdist_make_scheduler

pytest -v -n 4 --dist=loadfile

Step five ：worker Execute test case

1. workers Rewrote pytest_runtestloop：pytest The default implementation of is loop execution of all in test_session The test cases collected in this object .
2. But in xdist in , workers It's actually waiting master Send it the test cases that need to be executed .
3. When worker Received test task , Just do it in sequence pytest_runtest_protocol.
4. One detail worth noting is ：workers You must always keep at least one test case in your task queue , To be compatible with pytest_runtest_protocol(item, nextitem)hook The parameter requirements of , In order to nextitem Pass to hook.
5. master stay worker After executing the assigned set of tests , Based on the length of test execution and each worker The rest of the test cases synthesize to decide whether to send this worker Send more test cases .
6. worker Will wait before executing the last test item master More instructions for .
7. If it receives more tests , Then it can be executed safely pytest_runtest_protocol, Because at this time nextitem The parameters can be determined .
8. If it receives one shutdown The signal , Then we will nextitem The parameter is set to None, And then execute pytest_runtest_protocol

Step six ： End of test

1. When master When there are no more test tasks to perform , It will send one shutdown Signal to all worker.
2. When worker Exit the process after executing the remaining test cases .
3. When workers At the end of the test execution , The result will be sent back master, then master Forward the results to other pytest hooks such as ：pytest_runtest_logstart、pytest_runtest_logreport Ensure the normal operation of the whole test activity .
4. master Wait for all worker Exit all and close the test session .

Be careful ：pytest-xdist It's about having everyone worker The process executes all test cases under its own test case set . It means in different processes , Different test cases may call the same scope The scope level is higher （ for example session） Of fixture, The fixture It will be executed many times , This is not true. scope=session The expected .

pytest-xdist There is no built-in support to ensure session scoped fixture Only once , But it can be achieved by using file locks for interprocess communication ; Give Way scope=session Of fixture stay test session Only once in .

Example ： Need to install filelock package , Installation command pip install filelock

1. For example, it only needs to be executed once login（ Or define configuration options 、 Initialize database connection, etc ）.
2. When I first asked for this fixture when , Will use FileLock Only once fixture data .
3. When other processes request this again fixture when , It will not be repeated fixture.

import pytest
import uuid
from filelock import FileLock

 
@pytest.fixture(scope="session")
def login(tmp_path_factory, worker_id):
    #  Stands for stand-alone operation 
    if worker_id == "master":
        token = uuid.uuid4()
        print("fixture： Request login interface , obtain token", token)
        os.environ['token'] = token
        
        return token
        
    #  Distributed operation 
    #  Get the temporary directory shared by all child nodes , There is no need to modify 【 Not delete 、 modify 】
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + ".lock"):
        if fn.is_file():  #  The representative has already had a process to execute this fixture
            token = json.loads(fn.read_text())
        else:  #  On behalf of fixture For the first time 
            token = uuid.uuid4()
            fn.write_text(json.dumps(token))
        #  It's better to store the data that needs to be preserved in the future somewhere , For example, here is os Environment variables of 
        os.environ['token'] = token
	return token

---

Multithreaded execution of use cases pytest-parallel

be used for parallel and Concurrent The test of pytest plug-in unit

pip install pytest-parallel

Common parameter configuration

1. --workers=n ： This parameter is required for multi process operation , n It's the number of processes . The default is 1

2. --tests-per-worker=n ： Multiple threads need to add this parameter ,n It's the number of threads

If both parameters are configured , Process parallelism ; Each process has a maximum of n Threads , Bus number ： Number of processes * Number of threads

【 Be careful 】

1. stay windows The upper process number is always 1.

2. Need to use if name == “main” : Running the test example in the command line window will report an error

Example ：

• pytest test.py --workers 3 ：3 A process runs
• pytest test.py --tests-per-worker 4 ：4 Threads running
• pytest test.py --workers 2 --tests-per-worker 4 ：2 Processes in parallel , And each process has a maximum of 4 Threads running , That is, up to 8 Threads running .

  import pytest
  
  
  def test_01():
      print(' The test case 1 operation ')
  
  def test_02():
      print(' The test case 2 operation ')
  
  def test_03():
      print(' The test case 3 operation ')
  
  def test_04():
      print(' The test case 4 operation ')
      
  def test_05():
      print(' The test case 5 operation ')
  
  def test_06():
      print(' The test case 6 operation ')
      
  def test_07():
      print(' The test case 7 operation ')
  
  def test_08():
      print(' The test case 8 operation ')
  
  
  if __name__ == "__main__":
      pytest.main(["-s", "test_b.py", '--workers=2', '--tests-per-worker=4'])
  

---

pytest-parallel And pytest-xdist Contrast notes :

• pytest-parallel Than pytst-xdist Relatively easy to use , Multi function support ;
• pytst-xdist Multithreading not supported ;
• pytest-parallel Support python3.6 And above , So if you want to do multi process concurrency in linux perhaps mac Do on , stay Windows Up doesn't work （Workers=1）, If you do multithreading linux/mac/windows All platforms support , The process for workers Value .
• pytest-xdist The applicable scenario is ：
  • Not thread safe
  • Poor performance tests when multithreading
  • State isolation is required
• pytest-parallel For some use cases （ Such as Selenium） Better ：
  • Can be thread safe
  • It can be done to http Request to use non blocking IO To improve performance

In short ,pytest-xdist Parallelism pytest-parallel Parallelism and concurrency .

Finally, thank everyone who reads my article carefully , Watching the rise and attention of fans all the way , Reciprocity is always necessary , Although it's not very valuable , If you can use it, you can take it

These materials , For doing 【 software test 】 For our friends, it should be the most comprehensive and complete war preparation warehouse , This warehouse also accompanied me through the most difficult journey , I hope it can help you ！ Everything should be done as soon as possible , Especially in the technology industry , We must improve our technical skills . I hope that's helpful …….

If you don't want to experience it again, you can't find information when you study on your own , No one answers the question , If you insist on giving up after a few days , You can add mine below qq Group discussion and Exchange , There are also various software testing materials and technical exchanges .