前言：

• 實(shí)際項(xiàng)目中的用例數(shù)量會(huì)非常多，幾百上千；如果采用單進(jìn)程串行執(zhí)行的話會(huì)非常耗費(fèi)時(shí)間。假設(shè)每條用例耗時(shí)2s，1000條就需要2000s $\approx$ 33min；還要加上用例加載、測(cè)試前/后置套件等耗時(shí)；導(dǎo)致測(cè)試執(zhí)行效率會(huì)相對(duì)低。
• 想象一下如果開(kāi)發(fā)改動(dòng)一塊代碼，我們需要回歸一下，這時(shí)候執(zhí)行一下自動(dòng)化用例需要花費(fèi)大半個(gè)小時(shí)或者好幾個(gè)小時(shí)的時(shí)間，這是我們無(wú)法容忍的。
• 為了節(jié)省項(xiàng)目測(cè)試時(shí)間，需要多個(gè)測(cè)試用例同時(shí)并行執(zhí)行；這就是一種分布式場(chǎng)景來(lái)縮短測(cè)試用例的執(zhí)行時(shí)間，提高效率。

分布式執(zhí)行用例的原則：

• 用例之間是相互獨(dú)立的，沒(méi)有依賴關(guān)系，完全可以獨(dú)立運(yùn)行；
• 用例執(zhí)行沒(méi)有順序要求，隨機(jī)順序都能正常執(zhí)行；
• 每個(gè)用例都能重復(fù)運(yùn)行，運(yùn)行結(jié)果不會(huì)影響其他用例。

項(xiàng)目結(jié)構(gòu)

測(cè)試腳本

# 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"

正常執(zhí)行：需要8.10s

多進(jìn)程執(zhí)行用例之pytest-xdist

多cpu并行執(zhí)行用例，直接加個(gè)-n參數(shù)即可，后面num參數(shù)就是并行數(shù)量，比如num設(shè)置為3

pytest -v -n num

參數(shù)：

• -n auto ： 自動(dòng)偵測(cè)系統(tǒng)里的CPU數(shù)目
• -n num ： 指定運(yùn)行測(cè)試的處理器進(jìn)程數(shù)

多進(jìn)程并行執(zhí)行：耗時(shí)2.66s大大的縮短了測(cè)試用例的執(zhí)行時(shí)間。

pytest-xdist分布式測(cè)試的原理：

• xdist的分布式類似于一主多從的結(jié)構(gòu)，master負(fù)責(zé)下發(fā)命令，控制slave；slave根據(jù)master的命令執(zhí)行特定測(cè)試任務(wù)。

• 在xdist中，主是master，從是workers；xdist會(huì)產(chǎn)生一個(gè)或多個(gè)workers，workers都通過(guò)master來(lái)控制，每個(gè)worker相當(dāng)于一個(gè)mini版pytest執(zhí)行器 。

• master不執(zhí)行測(cè)試任務(wù)，只對(duì)worker收集到的所有用例進(jìn)行分發(fā)；每個(gè)worker負(fù)責(zé)執(zhí)行測(cè)試用例，然后將執(zhí)行結(jié)果反饋給master；由master統(tǒng)計(jì)最終測(cè)試結(jié)果。

pytest-xdist分布式測(cè)試的流程：

第一步：master創(chuàng)建worker

• master在測(cè)試會(huì)話（test session）開(kāi)始前產(chǎn)生一個(gè)或多個(gè)worker。

• master和worker之間是通過(guò)execnet和網(wǎng)關(guān)來(lái)通信的。

• 實(shí)際編譯執(zhí)行測(cè)試代碼的worker可能是本地機(jī)器也可能是遠(yuǎn)程機(jī)器。

第二步：workers收集測(cè)試項(xiàng)用例

• 每個(gè)worker類似一個(gè)迷你型的pytest執(zhí)行器。

• worker會(huì)執(zhí)行一個(gè)完整的test collection過(guò)程?！臼占袦y(cè)試用例的過(guò)程】

• 然后把測(cè)試用例的ids返回給master?！緄ds表示收集到的測(cè)試用例路徑】

• master不執(zhí)行任何測(cè)試用例。

注意：分布式測(cè)試（pytest-xdist）方式執(zhí)行測(cè)試時(shí)不會(huì)輸出測(cè)試用例中的print內(nèi)容，因?yàn)閙aster并不執(zhí)行測(cè)試用例。

第三步：master檢測(cè)workers收集到的測(cè)試用例集

• master接收到所有worker收集的測(cè)試用例集之后，master會(huì)進(jìn)行一些完整性檢查，以確保所有worker都收集到一樣的測(cè)試用例集（包括順序）。

• 如果檢查通過(guò)，會(huì)將測(cè)試用例的ids列表轉(zhuǎn)換成簡(jiǎn)單的索引列表，每個(gè)索引對(duì)應(yīng)一個(gè)測(cè)試用例的在原來(lái)測(cè)試集中的位置。

• 這個(gè)方案可行的原因是：所有的節(jié)點(diǎn)都保存著相同的測(cè)試用例集。

• 并且使用這種方式可以節(jié)省帶寬，因?yàn)閙aster只需要告知workers需要執(zhí)行的測(cè)試用例對(duì)應(yīng)的索引，而不用告知完整的測(cè)試用例信息。

第四步：master分發(fā)測(cè)試用例

有以下四種分發(fā)策略：命令行參數(shù) --dist=mode選項(xiàng)(默認(rèn)load)

each：master將完整的測(cè)試索引列表分發(fā)到每個(gè)worker，即每個(gè)worker都會(huì)執(zhí)行一遍所有的用例。

load：master將大約$\frac{1}{n}$的測(cè)試用例以輪詢的方式分發(fā)到各個(gè)worker，剩余的測(cè)試用例則會(huì)等待worker執(zhí)行完測(cè)試用例以后再分發(fā)；每個(gè)用例只會(huì)被其中一個(gè)worker執(zhí)行一次。

loadfile：master分發(fā)用例的策略為按ids中的文件名(test_xx.py/xx_test.py)進(jìn)行分發(fā)，即同一個(gè)測(cè)試文件中的測(cè)試用例只會(huì)分發(fā)給其中一個(gè)worker；具有一定的隔離性。

loadscope：master分發(fā)用例對(duì)策略為按作用域進(jìn)行分發(fā)，同一個(gè)模塊下的測(cè)試函數(shù)或某個(gè)測(cè)試類中的測(cè)試函數(shù)會(huì)分發(fā)給同一個(gè)worker來(lái)執(zhí)行；即py文件中無(wú)測(cè)試類的話(只有測(cè)試function)將該模塊分發(fā)給同一個(gè)worker執(zhí)行，如果有測(cè)試類則會(huì)將該文件中的測(cè)試類只會(huì)分發(fā)給同一個(gè)worker執(zhí)行，多個(gè)類可能分發(fā)給多個(gè)worker；目前無(wú)法自定義分組，按類 class 分組優(yōu)先于按模塊 module 分組。

注意：可以使用pytest_xdist_make_scheduler這個(gè)hook來(lái)實(shí)現(xiàn)自定義測(cè)試分發(fā)邏輯。
如：想按目錄級(jí)別來(lái)分發(fā)測(cè)試用例：

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)

• 只需在最外層conftest中繼承xdist.scheduler.LoadScopeScheduling并重寫(xiě)_split_scope方法
• 重寫(xiě)鉤子函數(shù)pytest_xdist_make_scheduler

pytest -v -n 4 --dist=loadfile

第五步：worker執(zhí)行測(cè)試用例

• workers 重寫(xiě)了pytest_runtestloop：pytest的默認(rèn)實(shí)現(xiàn)是循環(huán)執(zhí)行所有在test_session這個(gè)對(duì)象里面收集到的測(cè)試用例。
• 但是在xdist里, workers實(shí)際上是等待master為其發(fā)送需要執(zhí)行的測(cè)試用例。
• 當(dāng)worker收到測(cè)試任務(wù), 就順序執(zhí)行pytest_runtest_protocol。
• 值得注意的一個(gè)細(xì)節(jié)是：workers 必須始終保持至少一個(gè)測(cè)試用例在的任務(wù)隊(duì)列里, 以兼容pytest_runtest_protocol(item, nextitem)hook的參數(shù)要求，為了將nextitem傳給hook。
• master在worker執(zhí)行完分配的一組測(cè)試后，基于測(cè)試執(zhí)行時(shí)長(zhǎng)以及每個(gè)worker剩余測(cè)試用例綜合決定是否向這個(gè)worker發(fā)送更多的測(cè)試用例。
• worker會(huì)在執(zhí)行最后一個(gè)測(cè)試項(xiàng)前等待master的更多指令。
• 如果它收到了更多測(cè)試項(xiàng), 那么就可以安全的執(zhí)行 pytest_runtest_protocol，因?yàn)檫@時(shí)nextitem參數(shù)已經(jīng)可以確定。
• 如果它收到一個(gè) shutdown信號(hào), 那么就將nextitem參數(shù)設(shè)為None, 然后執(zhí)行 pytest_runtest_protocol

第六步：測(cè)試結(jié)束

• 當(dāng)master沒(méi)有更多執(zhí)行測(cè)試任務(wù)時(shí)，它會(huì)發(fā)送一個(gè)shutdown信號(hào)給所有worker。
• 當(dāng)worker將剩余測(cè)試用例執(zhí)行完后退出進(jìn)程。
• 當(dāng)workers在測(cè)試執(zhí)行結(jié)束時(shí)，會(huì)將結(jié)果被發(fā)送回master，然后master將結(jié)果轉(zhuǎn)發(fā)到其他pytest hooks比如：pytest_runtest_logstart、pytest_runtest_logreport 確保整個(gè)測(cè)試活動(dòng)進(jìn)行正常運(yùn)作。
• master等待所有worker全部退出并關(guān)閉測(cè)試會(huì)話。

注意：pytest-xdist 是讓每個(gè) worker 進(jìn)程執(zhí)行屬于自己的測(cè)試用例集下的所有測(cè)試用例。這意味著在不同進(jìn)程中，不同的測(cè)試用例可能會(huì)調(diào)用同一個(gè) scope 范圍級(jí)別較高（例如session）的 fixture，該 fixture 則會(huì)被執(zhí)行多次，這不符合 scope=session 的預(yù)期。

pytest-xdist 沒(méi)有內(nèi)置的支持來(lái)確保會(huì)話范圍的 fixture 僅執(zhí)行一次，但是可以通過(guò)使用鎖定文件進(jìn)行進(jìn)程間通信來(lái)實(shí)現(xiàn)；讓scope=session 的 fixture 在 test session 中僅執(zhí)行一次。

示例：需要安裝 filelock 包，安裝命令pip install filelock

• 比如只需要執(zhí)行一次login（或定義配置選項(xiàng)、初始化數(shù)據(jù)庫(kù)連接等）。
• 當(dāng)?shù)谝淮握?qǐng)求這個(gè)fixture時(shí)，則會(huì)利用FileLock僅產(chǎn)生一次fixture數(shù)據(jù)。
• 當(dāng)其他進(jìn)程再次請(qǐng)求這個(gè)fixture時(shí)，則不會(huì)重復(fù)執(zhí)行fixture。

import pytest
from filelock import FileLock

 
@pytest.fixture(scope="session")
def login(tmp_path_factory, worker_id):
    # 代表是單機(jī)運(yùn)行
    if worker_id == "master":
        token = str(random())
        print("fixture：請(qǐng)求登錄接口，獲取token", token)
        os.environ['token'] = token
        
        return token
        
    # 分布式運(yùn)行
    # 獲取所有子節(jié)點(diǎn)共享的臨時(shí)目錄，無(wú)需修改【不可刪除、修改】
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + ".lock"):
        if fn.is_file():  # 代表已經(jīng)有進(jìn)程執(zhí)行過(guò)該fixture
            token = json.loads(fn.read_text())
        else:  # 代表該fixture第一次被執(zhí)行
            token = str(random())
            fn.write_text(json.dumps(token))
        # 最好將后續(xù)需要保留的數(shù)據(jù)存在某個(gè)地方，比如這里是os的環(huán)境變量
        os.environ['token'] = token
	return token

多線程執(zhí)行用例之pytest-parallel

用于并行和并發(fā)測(cè)試的 pytest 插件

pip install pytest-parallel

常用參數(shù)配置

• --workers=n ：多進(jìn)程運(yùn)行需要加此參數(shù)， n是進(jìn)程數(shù)。默認(rèn)為1

• --tests-per-worker=n ：多線程需要添加此參數(shù)，n是線程數(shù)

如果兩個(gè)參數(shù)都配置了，就是進(jìn)程并行；每個(gè)進(jìn)程最多n個(gè)線程，總線程數(shù)：進(jìn)程數(shù)*線程數(shù)

【注意】

• 在windows上進(jìn)程數(shù)永遠(yuǎn)為1。

• 需要使用 if name == “main” :在命令行窗口運(yùn)行測(cè)試用例會(huì)報(bào)錯(cuò)

示例：

• pytest test.py --workers 3 ：3個(gè)進(jìn)程運(yùn)行
• pytest test.py --tests-per-worker 4 ：4個(gè)線程運(yùn)行
• pytest test.py --workers 2 --tests-per-worker 4 ：2個(gè)進(jìn)程并行，且每個(gè)進(jìn)程最多4個(gè)線程運(yùn)行，即總共最多8個(gè)線程運(yùn)行。

import pytest


def test_01():
    print('測(cè)試用例1操作')

def test_02():
    print('測(cè)試用例2操作')

def test_03():
    print('測(cè)試用例3操作')

def test_04():
    print('測(cè)試用例4操作')
    
def test_05():
    print('測(cè)試用例5操作')

def test_06():
    print('測(cè)試用例6操作')
    
def test_07():
    print('測(cè)試用例7操作')

def test_08():
    print('測(cè)試用例8操作')


if __name__ == "__main__":
    pytest.main(["-s", "test_b.py", '--workers=2', '--tests-per-worker=4'])

pytest-parallel與pytest-xdist對(duì)比說(shuō)明:

• pytest-parallel 比 pytst-xdist 相對(duì)好用，功能支持多；
• pytst-xdist 不支持多線程；
• pytest-parallel 支持python3.6及以上版本，所以如果想做多進(jìn)程并發(fā)在linux或者mac上做，在Windows上不起作用（Workers=1），如果做多線程linux/mac/windows平臺(tái)都支持，進(jìn)程數(shù)為workers的值。
• pytest-xdist適用場(chǎng)景為：
  • 不是線程安全的
  • 多線程時(shí)性能不佳的測(cè)試
  • 需要狀態(tài)隔離
• pytest-parallel對(duì)于某些用例（如 Selenium）更好：
  • 可以是線程安全的
  • 可以對(duì) http 請(qǐng)求使用非阻塞 IO 來(lái)提高性能

簡(jiǎn)而言之，pytest-xdist并行性pytest-parallel是并行性和并發(fā)性。

到此這篇關(guān)于pytest多進(jìn)程或多線程執(zhí)行測(cè)試的文章就介紹到這了。希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論