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

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

xdist+fixture(文件夹)
│  tmp(存放 allure 数据文件夹)
│  conftest.py
│  test_1.py
│  test_2.py
│  test_3.py
│ __init__.py │ 

import os
def test_1(test):
    print("os 环境变量",os.environ['token'])
    print("test1 测试用例", test)

import os
def test_2(test):
    print("os 环境变量",os.environ['token'])
    print("test2 测试用例", test)

import os
def test_3(test):
    print("os 环境变量",os.environ['token'])
    print("test3 测试用例", test)

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

#!/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