前言：

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

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

• 用例之間是相互獨立的，沒有依賴關系，完全可以獨立運行；
• 用例執(zhí)行沒有順序要求，隨機順序都能正常執(zhí)行；
• 每個用例都能重復運行，運行結(jié)果不會影響其他用例。

項目結(jié)構

測試腳本

# 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

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

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

pytest -v -n num

參數(shù)：

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

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

pytest-xdist分布式測試的原理：

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

• 在xdist中，主是master，從是workers；xdist會產(chǎn)生一個或多個workers，workers都通過master來控制，每個worker相當于一個mini版pytest執(zhí)行器 。

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

pytest-xdist分布式測試的流程：

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

• master在測試會話（test session）開始前產(chǎn)生一個或多個worker。

• master和worker之間是通過execnet和網(wǎng)關來通信的。

• 實際編譯執(zhí)行測試代碼的worker可能是本地機器也可能是遠程機器。

第二步：workers收集測試項用例

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

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

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

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

注意：分布式測試（pytest-xdist）方式執(zhí)行測試時不會輸出測試用例中的print內(nèi)容，因為master并不執(zhí)行測試用例。

第三步：master檢測workers收集到的測試用例集

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

• 如果檢查通過，會將測試用例的ids列表轉(zhuǎn)換成簡單的索引列表，每個索引對應一個測試用例的在原來測試集中的位置。

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

• 并且使用這種方式可以節(jié)省帶寬，因為master只需要告知workers需要執(zhí)行的測試用例對應的索引，而不用告知完整的測試用例信息。

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

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

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

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

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

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

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

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并重寫_split_scope方法
• 重寫鉤子函數(shù)pytest_xdist_make_scheduler

pytest -v -n 4 --dist=loadfile

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

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

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

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

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

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

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

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

import pytest
from filelock import FileLock

 
@pytest.fixture(scope="session")
def login(tmp_path_factory, worker_id):
    # 代表是單機運行
    if worker_id == "master":
        token = str(random())
        print("fixture：請求登錄接口，獲取token", token)
        os.environ['token'] = token
        
        return token
        
    # 分布式運行
    # 獲取所有子節(jié)點共享的臨時目錄，無需修改【不可刪除、修改】
    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)有進程執(zhí)行過該fixture
            token = json.loads(fn.read_text())
        else:  # 代表該fixture第一次被執(zhí)行
            token = str(random())
            fn.write_text(json.dumps(token))
        # 最好將后續(xù)需要保留的數(shù)據(jù)存在某個地方，比如這里是os的環(huán)境變量
        os.environ['token'] = token
	return token

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

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

pip install pytest-parallel

常用參數(shù)配置

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

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

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

【注意】

• 在windows上進程數(shù)永遠為1。

• 需要使用 if name == “main” :在命令行窗口運行測試用例會報錯

示例：

• pytest test.py --workers 3 ：3個進程運行
• pytest test.py --tests-per-worker 4 ：4個線程運行
• pytest test.py --workers 2 --tests-per-worker 4 ：2個進程并行，且每個進程最多4個線程運行，即總共最多8個線程運行。

import pytest


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

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

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

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

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

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


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

pytest-parallel與pytest-xdist對比說明:

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

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

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

最新評論