pytest-xdist--分散式執行用例
pytest-xdist基本的介紹
宣告:在介紹pytest-xdist時,本人不講任何原理,需要看原理的請移至官方:https://pypi.org/project/pytest-xdist/
當我們自動化測試用例非常多的時候, 一條條按順序執行會非常慢,pytest-xdist的出現就是為了讓自
動化測試用例可以分散式執行,從而節省自動化測試時間,pytest-xdist是屬於程序級別的併發。
pytest-xdist外掛在測試過程中可以使我們的測試用例一起並行測試,執行情況是根據你執行環境存在多個CPU,執行過程中可以進行組合測試執行,從而縮短我們的測試時間。
pytest-xdist外掛安裝
只需要在終端中執行如下命令:pip install pytest-xdist
pytest-xdist執行用例的條件
需滿足以下條件:
- 每一條用例必須是獨立的。用例之間沒有依賴關係,用例可以完全獨立執行【獨立執行】
- 每一條用例沒有特定的執行順序,就是每條用例都要遵循隨機執行【隨機執行】
- 每條用例的測試結果不能影響到其他的測試用例。【不影響其他用例】
pytest-xdist使用方法
pytest -n x
n:表示`使用並行引數
x:表示需要啟動多少個分散式,也即使用CPU的個數
- n auto:可以自動檢測到系統的CPU核數;從測試結果來看,檢測到的是邏輯處理器的
數量- 使用auto等於利用了所有CPU來跑用例,此時CPU佔用率會特別高
說明:建議最多使用1/2的CPU個數來進行執行,消耗資源太多,導致電腦太卡
接下來看例項:
在終端中分別輸入並執行:
pytest -vs
、pytest -vs -n 2
從結果可以看出,不是用分散式執行測試用例,總共用時10.71秒
那麼接下來我們使用分散式進行執行用例,看看他執行的時間:
可以看出使用分散式執行用例,同時有2個執行緒進行執行用例,時間為6.08秒,縮短了很多的時間。
pytest-xdist自定義執行模式
1.按照同一個作用域方法來分組,然後將每個測試組發給可以執行的worker,確保同一個組的測試
用例在同一個程序中執行:
--dist=loadscope #每個worker按類執行
示例:pytest -v -n 3 --dist=loadscope test_demo.py
2.按照同一個檔名來分組,然後將每個測試組發給可以執行的worker,確保同一個組的測試用例在
同一個程序中執行:
--dist=loadfile #每個worker按檔案執行
示例:pytest -v -n 3 --dist=loadfile test_xdist.py test_xdist_02.py test_xdist_03.py
3.是將每個用例,分別發給所有的執行器worker,相當於開了幾個執行器worker,同一個用例就執行幾遍:
--dist=each
示例:pytest -v -n 3 --dist=each test_xdist.py
4.將待執行的用例隨機發給可用的執行器worker,用例執行順序隨機的,目前預設採用這種方式
:
--dist=load 和 --dist==no
示例:pytest -v -n 3 --dist=load test_xdist.py
如何讓scope=session的fixture在test session中僅僅執行一次
pytest-xdist是讓每個worker程序執行屬於自己的測試用例集下的所有測試用例,這意味著在不同
程序中,不同的測試用例可能會呼叫同一個scope範圍級別較高(例如session)的fixture,該
fixture則會被執行多次,這不符scope=session的預期。
雖然pytest-xdist沒有內建的支援來確保會話範圍的夾具僅執行一次,但是可以通過使用鎖定檔案
進行程序間通訊來實現。
import pytest
from filelock import FileLock
@pytest.fixture(scope="session",autouse=True)
def login(tmp_path_factory, worker_id):
# 如果是單機執行 則執行這裡的程式碼塊【不可刪除、修改】
if worker_id == "master":
"""
【自定義程式碼塊】
這裡就寫你要本身應該要做的操作,比如:登入請求、新增資料、清空資料庫歷史資料等等
"""
uuid_value = uuid.uuid1()
token = uuid_value.hex
print("fixture:請求登入介面,獲取token", token)
os.environ['token'] = 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():
# 快取檔案中讀取資料,像登入操作的話就是 token 【不可刪除、修改】
token = json.loads(fn.read_text())
print(f"讀取快取檔案,token 是{token} ")
else:
"""
【自定義程式碼塊】
跟上面 if 的程式碼塊一樣就行
"""
uuid_value = uuid.uuid1()
token = uuid_value.hex
print("fixture:請求登入介面,獲取token", token)
# 【不可刪除、修改】
fn.write_text(json.dumps(token))
print(f"首次執行,token 是{token} ")
# 最好將後續需要保留的資料存在某個地方,比如這裡是 os 的環境變數
os.environ['token'] = token
return token
- 示例只需要執行一次login(因為它是隻需要執行一次來定義配置選項,等等)
- 當第一次請求這個fixture時,則會利用FileLock僅產生一次fixture資料
- 當其他程序再次請求這個fixture時,則會從檔案中讀取資料