Pytest系列（30）- 使用 pytest-xdist 分布式插件，如何保證 scope=session 的 fixture 在多進程運行情況下仍然能只運行一次

本文轉載自查看原文 2021-04-27 14:28 1290 Pytest/ 測試高級進階技能系列 - Pytest

如果你還想從頭學起Pytest，可以看看這個系列的文章哦！

https://www.cnblogs.com/poloyy/category/1690628.html

背景

使用 pytest-xdist 分布式插件可以加快運行，充分利用機器多核 CPU 的優勢
將常用功能放到 fixture，可以提高復用性和維護性
做接口自動化測試的時候，通常我們會將登錄接口放到 fixture 里面，並且 scope 會設置為 session，讓他全局只運行一次
但是當使用 pytest-xdist 的時候，scope=session 的 fixture 無法保證只運行一次，官方也通報了這一問題

官方描述

pytest-xdist 的設計使每個工作進程將執行自己的測試集合並執行所有測試子集，這意味着在不同的測試過程中，要求高級范圍的 fixture（如：session）將會被多次執行，這超出了預期，在某些情況下可能是不希望的
盡管 pytest-xdist 沒有內置支持來確保 scope=session 的fixture 僅執行一次，但是可以通過使用鎖定文件進行進程間通信來實現

前置知識

pytest-xdist 分布式插件使用詳細教程

https://www.cnblogs.com/poloyy/p/12694861.html

pytest-xdist 分布式插件原理

https://www.cnblogs.com/poloyy/p/12703290.html

fixture 的使用詳細教程

https://www.cnblogs.com/poloyy/p/12642602.html

官方文檔

https://pypi.org/project/pytest-xdist/

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

import json

import pytest
from 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 job
        return produce_expensive_data()

    # get the temp directory shared by all workers
    root_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

若某個 scope = session 的 fixture 需要確保只運行一次的話，可以用上面的方法，直接套用，然后改需要改的部分即可（這個后面詳細講解）
官方原話：這項技術可能並非在每種情況下都適用，但對於許多情況下，它應該是一個起點，在這種情況下，對於 scope = session 的fixture 只執行一次很重要

后續栗子的代碼

項目結構

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

test_1.py 代碼

import os


def test_1(test):
    print("os 環境變量",os.environ['token'])
    print("test1 測試用例", test)

test_2.py 代碼

import os


def test_2(test):
    print("os 環境變量",os.environ['token'])
    print("test2 測試用例", test)

test_3.py 代碼

import os


def test_3(test):
    print("os 環境變量",os.environ['token'])
    print("test3 測試用例", test)

未解決情況下的栗子

conftest.py 代碼

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

運行命令

pytest -n 3 --alluredir=tmp

運行結果

scope=session 的 fixture 很明顯執行了三次，三個進程下的三個測試用例得到的數據不一樣，明顯不會是我們想要的結果

使用官方解決方法的栗子

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
__title__  = 
__Time__   = 2021/4/27 11:28
__Author__ = 小菠蘿測試筆記
__Blog__   = https://www.cnblogs.com/poloyy/
"""
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