1. 程式人生 > 其它 >pytest-xdist--分散式執行用例

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 -vspytest -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
  1. 示例只需要執行一次login(因為它是隻需要執行一次來定義配置選項,等等)
  2. 當第一次請求這個fixture時,則會利用FileLock僅產生一次fixture資料
  3. 當其他程序再次請求這個fixture時,則會從檔案中讀取資料