使用pytest-xdist分布式插件如何保證scope=session 的fixture在多進(jìn)程運(yùn)行情況下仍然能只運(yùn)行一次

這篇文章給大家介紹使用pytest-xdist分布式插件如何保證scope=session 的fixture在多進(jìn)程運(yùn)行情況下仍然能只運(yùn)行一次，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

背景

• 使用 pytest-xdist 分布式插件可以加快運(yùn)行，充分利用機(jī)器多核 CPU 的優(yōu)勢(shì)

• 將常用功能放到 fixture，可以提高復(fù)用性和維護(hù)性

• 做接口自動(dòng)化測(cè)試的時(shí)候，通常我們會(huì)將登錄接口放到 fixture 里面，并且 scope 會(huì)設(shè)置為 session，讓他全局只運(yùn)行一次

• 但是當(dāng)使用 pytest-xdist 的時(shí)候，scope=session 的 fixture 無法保證只運(yùn)行一次，官方也通報(bào)了這一問題

官方描述

• pytest-xdist 的設(shè)計(jì)使每個(gè)工作進(jìn)程將執(zhí)行自己的測(cè)試集合并執(zhí)行所有測(cè)試子集，這意味著在不同的測(cè)試過程中，要求高級(jí)范圍的 fixture（如：session）將會(huì)被多次執(zhí)行，這超出了預(yù)期，在某些情況下可能是不希望的

• 盡管 pytest-xdist 沒有內(nèi)置支持來確保  scope=session 的fixture 僅執(zhí)行一次，但是可以通過使用鎖定文件進(jìn)行進(jìn)程間通信來實(shí)現(xiàn)

官方解決辦法（直接套用就行）

import jsonimport pytestfrom filelock import FileLock


@pytest.fixture(scope="session")def session_data(tmp_path_factory, worker_id):if worker_id == "master":# not executing in with multiple workers, just produce the data and let# pytest's fixture caching do its jobreturn produce_expensive_data()# get the temp directory shared by all workersroot_tmp_dir = tmp_path_factory.getbasetemp().parent

    fn = root_tmp_dir / "data.json"with FileLock(str(fn) + ".lock"):if fn.is_file():
            data = json.loads(fn.read_text())else:
            data = produce_expensive_data()
            fn.write_text(json.dumps(data))return data

• 若某個(gè) scope = session 的 fixture 需要確保只運(yùn)行一次的話，可以用上面的方法，直接套用，然后改需要改的部分即可（這個(gè)后面詳細(xì)講解）

• 官方原話：這項(xiàng)技術(shù)可能并非在每種情況下都適用，但對(duì)于許多情況下，它應(yīng)該是一個(gè)起點(diǎn)，在這種情況下，對(duì)于 scope = session 的fixture 只執(zhí)行一次很重要

后續(xù)栗子的代碼

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

xdist+fixture(文件夾)
│  tmp(存放 allure 數(shù)據(jù)文件夾)
│  conftest.py
│  test_1.py
│  test_2.py
│  test_3.py
│ __init__.py │

test_1.py 代碼

import osdef test_1(test):print("os 環(huán)境變量",os.environ['token'])print("test1 測(cè)試用例", test)

test_2.py 代碼

import osdef test_2(test):print("os 環(huán)境變量",os.environ['token'])print("test2 測(cè)試用例", test)

test_3.py 代碼

import osdef test_3(test):print("os 環(huán)境變量",os.environ['token'])print("test3 測(cè)試用例", test)

未解決情況下的栗子

conftest.py 代碼

import osimport pytestfrom random import random


@pytest.fixture(scope="session")def test():
    token = str(random())print("fixture:請(qǐng)求登錄接口，獲取token", token)
    os.environ['token'] = tokenreturn token

運(yùn)行命令

pytest -n 3 --alluredir=tmp

運(yùn)行結(jié)果

scope=session 的 fixture 很明顯執(zhí)行了三次，三個(gè)進(jìn)程下的三個(gè)測(cè)試用例得到的數(shù)據(jù)不一樣，明顯不會(huì)是我們想要的結(jié)果

使用官方解決方法的栗子rt 

#!/usr/bin/env python# -*- coding: utf-8 -*-"""__title__  = 
__Time__   = 2021/4/27 11:28
__Author__ = 小菠蘿測(cè)試筆記
__Blog__   = https://www.cnblogs.com/poloyy/"""import jsonimport osimport pytestfrom random import randomfrom filelock import FileLock

@pytest.fixture(scope="session")def test(tmp_path_factory, worker_id):# 如果是單機(jī)運(yùn)行 則運(yùn)行這里的代碼塊【不可刪除、修改】if worker_id == "master":"""【自定義代碼塊】
        這里就寫你要本身應(yīng)該要做的操作，比如：登錄請(qǐng)求、新增數(shù)據(jù)、清空數(shù)據(jù)庫歷史數(shù)據(jù)等等"""token = str(random())print("fixture:請(qǐng)求登錄接口，獲取token", token)
        os.environ['token'] = token# 如果測(cè)試用例有需要，可以返回對(duì)應(yīng)的數(shù)據(jù)，比如 tokenreturn token# 如果是分布式運(yùn)行# 獲取所有子節(jié)點(diǎn)共享的臨時(shí)目錄，無需修改【不可刪除、修改】root_tmp_dir = tmp_path_factory.getbasetemp().parent# 【不可刪除、修改】fn = root_tmp_dir / "data.json"# 【不可刪除、修改】with FileLock(str(fn) + ".lock"):# 【不可刪除、修改】if fn.is_file():# 緩存文件中讀取數(shù)據(jù)，像登錄操作的話就是 token 【不可刪除、修改】token = json.loads(fn.read_text())print(f"讀取緩存文件，token 是{token} ")else:"""【自定義代碼塊】
            跟上面 if 的代碼塊一樣就行"""token = str(random())print("fixture:請(qǐng)求登錄接口，獲取token", token)# 【不可刪除、修改】            fn.write_text(json.dumps(token))print(f"首次執(zhí)行，token 是{token} ")# 最好將后續(xù)需要保留的數(shù)據(jù)存在某個(gè)地方，比如這里是 os 的環(huán)境變量os.environ['token'] = tokenreturn token

運(yùn)行命令

pytest -n 3 --alluredir=tmp

運(yùn)行結(jié)果

可以看到 fixture 只執(zhí)行了一次，不同進(jìn)程下的測(cè)試用例共享一個(gè)數(shù)據(jù) token

重點(diǎn)

• 讀取緩存文件并不是每個(gè)測(cè)試用例都會(huì)讀，它是按照進(jìn)程來讀取的

• 比如 -n 3 指定三個(gè)進(jìn)程運(yùn)行，那么有一個(gè)進(jìn)程會(huì)執(zhí)行一次 fixture（隨機(jī)），另外兩個(gè)進(jìn)程會(huì)各讀一次緩存

• 假設(shè)每個(gè)進(jìn)程有很多個(gè)用例，那也只是讀一次緩存文件，而不會(huì)讀多次緩存文件

• 所以最好要將從緩存文件讀出來的數(shù)據(jù)保存在特定的地方，比如上面代碼的 os.environ 可以將數(shù)據(jù)保存在環(huán)境變量中

兩個(gè)進(jìn)程跑三個(gè)測(cè)試用例文件

還是上面栗子的代碼

運(yùn)行命令

pytest -n 2 --alluredir=tmp

運(yùn)行結(jié)果

可以看到 test_3 的測(cè)試用例就沒有讀緩存文件了，每個(gè)進(jìn)程只會(huì)讀一次緩存文件，記住哦！

關(guān)于使用pytest-xdist分布式插件如何保證scope=session 的fixture在多進(jìn)程運(yùn)行情況下仍然能只運(yùn)行一次就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。