pytest多進程或多線程執行測試實例

前言：

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

分佈式執行用例的原則：

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

項目結構

測試腳本

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

正常執行：需要8.10s

多進程執行用例之pytest-xdist

多cpu並行執行用例，直接加個-n參數即可，後面num參數就是並行數量，比如num設置為3

pytest -v -n num

參數：

• -n auto ： 自動偵測系統裡的CPU數目
• -n num ： 指定運行測試的處理器進程數

多進程並行執行：耗時2.66s大大的縮短瞭測試用例的執行時間。

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

• xdist的分佈式類似於一主多從的結構，master負責下發命令，控制slave；slave根據master的命令執行特定測試任務。

• 在xdist中，主是master，從是workers；xdist會產生一個或多個workers，workers都通過master來控制，每個worker相當於一個mini版pytest執行器 。

• master不執行測試任務，隻對worker收集到的所有用例進行分發；每個worker負責執行測試用例，然後將執行結果反饋給master；由master統計最終測試結果。

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

第一步：master創建worker

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

• master和worker之間是通過execnet和網關來通信的。

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

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

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

• worker會執行一個完整的test collection過程。【收集所有測試用例的過程】

• 然後把測試用例的ids返回給master。【ids表示收集到的測試用例路徑】

• master不執行任何測試用例。

註意：分佈式測試（pytest-xdist）方式執行測試時不會輸出測試用例中的print內容，因為master並不執行測試用例。

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

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

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

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

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

第四步：master分發測試用例

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

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

load：master將大約$\frac{1}{n}$的測試用例以輪詢的方式分發到各個worker，剩餘的測試用例則會等待worker執行完測試用例以後再分發；每個用例隻會被其中一個worker執行一次。

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

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

註意：可以使用pytest_xdist_make_scheduler這個hook來實現自定義測試分發邏輯。
如：想按目錄級別來分發測試用例：

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方法
• 重寫鉤子函數pytest_xdist_make_scheduler

pytest -v -n 4 --dist=loadfile

第五步：worker執行測試用例

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

第六步：測試結束

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

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

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

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

• 比如隻需要執行一次login（或定義配置選項、初始化數據庫連接等）。
• 當第一次請求這個fixture時，則會利用FileLock僅產生一次fixture數據。
• 當其他進程再次請求這個fixture時，則不會重復執行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
        
    # 分佈式運行
    # 獲取所有子節點共享的臨時目錄，無需修改【不可刪除、修改】
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + ".lock"):
        if fn.is_file():  # 代表已經有進程執行過該fixture
            token = json.loads(fn.read_text())
        else:  # 代表該fixture第一次被執行
            token = str(random())
            fn.write_text(json.dumps(token))
        # 最好將後續需要保留的數據存在某個地方，比如這裡是os的環境變量
        os.environ['token'] = token
	return token

多線程執行用例之pytest-parallel

用於並行和並發測試的 pytest 插件

pip install pytest-parallel

常用參數配置

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

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

如果兩個參數都配置瞭，就是進程並行；每個進程最多n個線程，總線程數：進程數*線程數

【註意】

• 在windows上進程數永遠為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及以上版本，所以如果想做多進程並發在linux或者mac上做，在Windows上不起作用（Workers=1），如果做多線程linux/mac/windows平臺都支持，進程數為workers的值。
• pytest-xdist適用場景為：
  • 不是線程安全的
  • 多線程時性能不佳的測試
  • 需要狀態隔離
• pytest-parallel對於某些用例（如 Selenium）更好：
  • 可以是線程安全的
  • 可以對 http 請求使用非阻塞 IO 來提高性能

簡而言之，pytest-xdist並行性pytest-parallel是並行性和並發性。

到此這篇關於pytest多進程或多線程執行測試的文章就介紹到這瞭。希望對大傢的學習有所幫助，也希望大傢多多支持WalkonNet。

推薦閱讀：

• pytest實現多進程與多線程運行超好用的插件
• 詳解pytest分佈式執行插件 pytest-xdist 的高級用法
• Pytest框架 conftest.py文件的使用詳解
• python單測框架之pytest常見用法
• 詳解如何使用Pytest進行自動化測試