1、安裝

2、示例

3、原理和流程

4、解決：多行程運行次數

當測驗用例非常多的時候，一條條按順序執行測驗用例，是很浪費測驗時間的，這時候就可以用到 pytest-xdist，讓自動化測驗用例可以分布式執行，從而大大節省測驗時間，

pytest-xdist 是屬于行程級別的并發，

分布式測驗用例的設計原則：

（1）獨立運行：用例之間是獨立的，并且沒有依賴關系，還可以完全獨立運行，

（2）隨機執行：用例執行不強制按順序執行，支持順序執行或隨機執行，

（3）不影響其他用例：每個用例都能重復運行，運行結果不會影響其他用例，

pytest-xdist 通過一些獨特的測驗執行模式擴展了 pytest：

（1）測驗運行并行化：如果有多個CPU或主機，則可以將它們用于組合的測驗運行，這樣可以加快開發速度或使用遠程計算機的特殊資源，

（2）--looponfail：在子行程中重復運行測驗，每次運行之后，pytest 都會等到專案中的檔案更改后再運行之前失敗的測驗，重復此程序，直到所有測驗通過，然后再次執行完整運行，

（3）跨平臺覆寫：可以指定不同的 Python 解釋器或不同的平臺，并在所有這些平臺上并行運行測驗，

1、安裝

在命令列中運行以下命令進行安裝：

pip install pytest-xdist

或者（使用國內的豆瓣源，資料會定期同步國外官網，速度快，）

pip install pytest-xdist -i http://pypi.douban.com/simple/ --trusted-host pypi.douban.com

2、示例

創建My_pytest_Demo3專案，并創建如下檔案，

如圖所示：專案目錄結構

根目錄下conftest.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest

@pytest.fixture(scope="session")
def login():
    print("===登錄，回傳：name，token===")
    name = "AllTests"
    token = "123456qwe"
    yield name, token
    print("===退出===")

根目錄下test_case.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest
from time import sleep

@pytest.mark.parametrize("n", list(range(5)))
def test_get_info(login, n):
    sleep(1)
    name, token = login
    print("===獲取用戶個人資訊===", n)
    print(f"用戶名:{name}, token:{token}")

test_baidu包下conftest.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest

@pytest.fixture(scope="module")
def open_baidu(login):
    name, token = login
    print(f"===用戶 {name} 打開baidu===")

test_baidu包下test_case1.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest
from time import sleep

@pytest.mark.parametrize("n", list(range(5)))
def test_case1_1(open_baidu, n):
    sleep(1)
    print("===baidu 執行測驗用例test_case1_1===", n)

@pytest.mark.parametrize("n", list(range(5)))
def test_case1_2(open_baidu, n):
    sleep(1)
    print("===baidu 執行測驗用例test_case1_2===", n)

test_weibo包下test_case2.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest
from time import sleep

@pytest.mark.parametrize("n", list(range(5)))
def test_case2_no_fixture(login, n):
    sleep(1)
    print("===weibo 沒有__init__測驗用例，執行測驗用例test_case2_no_fixture===", login)

test_douyin包下conftest.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest

@pytest.fixture(scope="function")
def open_douyin(login):
    name, token = login
    print(f"===用戶 {name} 打開douyin===")

test_douyin包下test_case3.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest
from time import sleep

@pytest.mark.parametrize("n", list(range(5)))
class TestDouyin:
    def test_case3_1(self, open_douyin, n):
        sleep(1)
        print("===douyin 執行測驗用例test_case3_1===", n)

    def test_case3_2(self, open_douyin, n):
        sleep(1)
        print("===douyin 執行測驗用例test_case3_2===", n)

1、不使用分布式測驗執行測驗用例

打開命令列，在該專案根目錄下，輸入執行命令

pytest -s

執行一條用例大概1s，因為每個用例都加了sleep(1)，一共30條用例，總共運行30.16s，

2、使用分布式測驗執行測驗用例

引數 -n auto：可以自動檢測到系統的CPU核數，

使用 auto 等于利用了所有CPU來跑用例，此時CPU占用率會特別高，

打開命令列，在該專案根目錄下，輸入執行命令

pytest -s -n auto

執行30條用例，只用了4.81s，

3、使用分布式測驗執行測驗用例（指定多少行程）

打開命令列，在該專案根目錄下，輸入執行命令

pytest -s -n 5

指定5個行程同時執行30條用例，用時6.99s，

4、pytest-xdist 和 pytest-html 聯合使用

打開命令列，在該專案根目錄下，輸入執行命令

pytest -s -n auto --html=report.html --self-contained-html

執行完成后自動生成的報告

5、按照一定順序執行

pytest-xdist 默認是無序執行的，可以通過 --dist 引數來控制執行順序，

--dist=loadscope：將按照同一個模塊 module 下的函式和同一個測驗類 class 下的方法來分組，然后將每個測驗組發給可以執行的 worker，確保同一個組的測驗用例在同一個行程中執行，目前無法自定義分組，按類 class 分組優先于按模塊 module 分組，

--dist=loadfile：按照同一個檔案名來分組，然后將每個測驗組發給可以執行的 worker，確保同一個組的測驗用例在同一個行程中執行，

6、使 scope=session 的 fixture 在 test session 中僅執行一次

pytest-xdist 是讓每個 worker 行程執行屬于自己的測驗用例集下的所有測驗用例，

這意味著在不同行程中，不同的測驗用例可能會呼叫同一個 scope 范圍級別較高（例如session）的 fixture，該 fixture 則會被執行多次，這不符合 scope=session 的預期，

盡管 pytest-xdist 沒有內置的支持來確保會話范圍的 fixture 僅執行一次，但是可以通過使用鎖定檔案進行行程間通信來實作，

示例：

（1）該示例只需要執行一次login（如只需要執行一次來定義配置選項等），

（2）當第一次請求這個fixture時，則會利用FileLock僅產生一次fixture資料，需要安裝filelock包，安裝命令pip install filelock

（3）當其他行程再次請求這個fixture時，則會從檔案中讀取資料，

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import pytest
from filelock import FileLock

@pytest.fixture(scope="session")
def login():
    print("===登錄，回傳：name，token===")
    with FileLock("session.lock"):
        name = "AllTests"
        token = "123456qwe"

        # Web App UI自動化，宣告一個driver，再回傳
        # 介面自動化，發起一個登錄請求，將token回傳

    yield name, token
    print("===退出===")

3、原理和流程

xdist 的分布式類似于一主多從的結構，master 機負責下發命令，控制 slave 機；slave 機根據 master 機的命令執行特定測驗任務，

在 xdist 中，主是 master，從是 workers，

分布式測驗的原理：

（1）xdist 會產生一個或多個 workers，workers 都通過 master 來控制；

（2）每個 worker 負責執行完整的測驗用例集，然后按照 master 的要求運行測驗，而 master 機不執行測驗任務，

分布式測驗的流程：

1、創建 worker

（1）master 會在總測驗會話（test session）開始前產生一個或多個 worker；

（2）master 和 worker 之間是通過 execnet 和網關來通信的；

（3）實際編譯執行測驗代碼的 worker 可能是本地機器也可能是遠程機器，

2、收集測驗用例

（1）每個 worker 類似一個迷你型的 pytest 執行器；

（2）worker 會執行一個完整的 test collection 程序（收集所有測驗用例的程序）；

（3）然后把測驗用例的 ids 回傳給 master；

（4）master 是不會執行任何測驗用例集的，

注：所以為什么腳本代碼里有列印陳述句（print）通過分布式測驗時結果沒有輸出用例的列印內容，因為主機并不執行測驗用例，PyCharm 相當于一個 master，

3、master 檢測 workers 收集到的測驗用例集

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

（2）如果檢查通過，會將測驗用例的 ids 串列轉換成簡單的索引串列，每個索引對應一個測驗用例的在原來測驗集中的位置；

（3）所有的節點都保存著相同的測驗用例集，并且使用這種方式可以節省帶寬，因為 master 只需要告知 workers 需要執行的測驗用例對應的索引，而不用告知完整的測驗用例資訊，

4、測驗用例分發

--dist-mode 選項

each：master 將完整的測驗索引串列分發到每個 worker，

load：master 將大約25%的測驗用例以輪詢的方式分發到各個 worker，剩余的測驗用例則會等待 workers 執行完測驗用例以后再分發，

注：可以使用 pytest_xdist_make_scheduler 這個 hook 來實作自定義測驗分發邏輯，

5、測驗用例的執行

（1）workers 重寫了 pytest_runtestloop（pytest 的默認實作是回圈執行所有在 test session 這個物件里面收集到的測驗用例）；

（2）但是在 xdist 里, workers 實際上是等待 master 為其發送需要執行的測驗用例；

（3）當 worker 收到測驗任務, 就順序執行 pytest_runtest_protocol；

（4）值得注意的一個細節是：workers 必須始終保持至少一個測驗用例在任務佇列里, 以兼容 pytest_runtest_protocol(item, nextitem)hook 的引數要求，為了將 nextitem 傳給 hook；

（5）worker 會在執行最后一個測驗項前等待 master 的更多指令；

（6）如果它收到了更多測驗項, 那么就可以安全的執行 pytest_runtest_protocol，因為這時 nextitem 引數已經可以確定；

（7）如果它收到一個 "shutdown" 信號, 那么就將 nextitem 引數設為 None, 然后執行 pytest_runtest_protocol，

6、測驗用例再分發

--dist-mode=load

（1）當 workers 開始/結束執行時，會把測驗結果回傳給 master，這樣其他 pytest hook 比如（pytest_runtest_protocol 和 pytest_runtest_protocol 就可以正常執行）；

（2）master 在 worker 執行完一個測驗后，基于測驗執行時長以及每個 work 剩余測驗用例綜合決定是否向這個 worker 發送更多的測驗用例，

7、測驗結束

（1）當 master 沒有更多執行測驗任務時，它會發送一個 "shutdown" 信號給所有 worker；

（2）當 worker 將剩余測驗用例執行完后退出行程；

（3）master 等待所有 worker 全部退出；

（4）此時仍需要處理諸如 pytest_runtest_logreport 等事件，

4、解決：多行程運行次數

如何保證 scope=session 的 fixture 在多行程運行情況下仍然只運行一次，

1、創建My_pytest_Demo3_2專案，并創建如下檔案，

如圖所示：專案目錄結構，allure檔案夾存放allure測驗報告

根目錄下conftest.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import os
import pytest
from random import random

@pytest.fixture(scope="session")
def test():
    token = str(random())
    print("fixture：請求登錄介面，獲取token", token)
    os.environ['token'] = token
    return token

根目錄下test_case1.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import os

def test_one(test):
    print("os 環境變數：", os.environ['token'])
    print("test_one 測驗用例", test)

根目錄下test_case2.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import os

def test_two(test):
    print("os 環境變數：", os.environ['token'])
    print("test_two 測驗用例", test)

根目錄下test_case3.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""

import os

def test_three(test):
    print("os 環境變數：", os.environ['token'])
    print("test_three 測驗用例", test)

2、打開命令列，在該專案根目錄下，輸入執行命令

pytest -n 3 --alluredir=./allure
allure serve allure

3、運行結果：

scope=session的fixture執行了三次，三個行程下的三個測驗用例得到的資料不一樣，

一、解決 scope=session 的 fixture 在多行程運行情況下仍然只運行一次

1、修改根目錄下conftest.py檔案

腳本代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
微信公眾號：AllTests軟體測驗
"""
import json
import os
import pytest
from random import random
from filelock import FileLock

@pytest.fixture(scope="session")
def test(tmp_path_factory, worker_id):
    # 如果是單機運行，則運行這里的代碼塊【不可洗掉、修改】
    if worker_id == "master":
        """
        【自定義代碼塊】
        這里就寫你要本身要做的操作，比如：登錄請求、新增資料、清空資料庫歷史資料等
        """
        token = str(random())
        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的代碼塊一樣就行
            """
            token = str(random())
            print("fixture：請求登錄介面，獲取token", token)
            # 【不可洗掉、修改】
            fn.write_text(json.dumps(token))
            print(f"首次執行，token是：{token}")

        # 最好將后續需要保留的資料存在某個地方，比如這里是os的環境變數
        os.environ['token'] = token
    return token

2、打開命令列，在該專案根目錄下，輸入執行命令

pytest -n 3 --alluredir=./allure
allure serve allure

3、運行結果：

可以看到fixture只執行了一次，不同行程下的測驗用例共享一個資料token，

（1）讀取快取檔案并不是每個測驗用例都會讀，它是按照行程來讀取的，比如 -n 3 指定三個行程運行，那么有一個行程會執行一次 fixture（隨機），另外兩個行程會各讀一次快取，

（2）假設每個行程有很多個用例，那也只是讀一次快取檔案，而不會讀多次快取檔案，所以最好將從快取檔案讀出來的資料保存在指定的地方，比如 os.environ 將資料保存在環境變數中，

二、行程少測驗用例多的情況下執行

例如：兩個行程跑三個測驗用例

1、打開命令列，在該專案根目錄下，輸入執行命令

pytest -n 2 --alluredir=./allure
allure serve allure

2、運行結果：

可以看到test_three的測驗用例就沒有讀快取檔案，每個行程只會讀一次快取檔案，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/344185.html

標籤：其他

上一篇：如何通過HBuilderX開發除錯微信小程式

下一篇：Python測驗框架pytest（23）插件 - pytest-picked、pytest-lazy-fixture

Python測驗框架pytest（22）插件 - pytest-xdist（分布式執行）

1、安裝

2、示例

3、原理和流程

4、解決：多行程運行次數