使用pytest-xdist 并行运行用例时,报错different test were collect between gw0 and gw3

@pytest.fixture(scope="session")
def test_token():
    res = None
    # 获取 token
    while FileLock("session.lock"):
        corpid = "wwa26dee94d70aa6e7"
        corpsecret = "TM-PNqujpmiAo46PQGBiikCWEYFXWzD8RYjmi0d--EU"
        res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={corpid}&"
                           f"corpsecret={corpsecret}")
    return res.json()["access_token"]


def test_get(userid, test_token):
    # 根据 user-id查询成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/get?access_token={test_token}&userid={userid}")
    return res.json()


def test_create(userid, name, mobile, test_token):
    # 添加成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
        "department": [1],
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/create?access_token={test_token}",
                        json=data
                        )
    return res.json()


def test_update(userid, name, mobile, test_token):
    # 更新成员
    data = {
        "userid": userid,
        "name": name,
        "mobile": mobile,
    }
    res = requests.post(f"https://qyapi.weixin.qq.com/cgi-bin/user/update?access_token={test_token}",
                        json=data)
    return res.json()


def test_delete(userid, test_token):
    # 删除成员
    res = requests.get(f"https://qyapi.weixin.qq.com/cgi-bin/user/delete?access_token={test_token}&userid={userid}")
    return res.json()


def test_create_data():
    data = [("wu123fff" + str(x), "zhangsan", "138%08d" % x) for x in range(20)]
    return data


@pytest.mark.parametrize("userid, name, mobile", test_create_data())
def test_all(userid, name, mobile, test_token):
    # 可能发生创建失败
    try:
        assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    except AssertionError as e:
        if "mobile existed" in e.__str__():
            # 如果手机号被使用了,找出使用手机号的 userid ,进行删除
            re_userid = re.findall(":(.*)", e.__str__())[0]
            if re_userid.endswith("'") or re_userid.endswith('"'):
                re_userid = re_userid[:-1]
            assert "deleted" == test_delete(re_userid, test_token)['errmsg']
            assert 60111 == test_get(re_userid, test_token)['errcode']
            assert "created" == test_create(userid, name, mobile, test_token)["errmsg"]
    # 可能发生userid不存在异常
    assert name == test_get(userid, test_token)['name']
    assert "updated" == test_update(userid, "xxxxxxx", mobile, test_token)['errmsg']
    assert "xxxxxxx" == test_get(userid, test_token)['name']
    assert "deleted" == test_delete(userid, test_token)['errmsg']
    assert 60111 == test_get(userid, test_token)['errcode']`预格式化文本`

代码如上:
问题:
使用pytest-xdist插件并行跑用例时,报如下错误,不使用插件并行跑用例是成功的
image
请教老师和同学看有人遇到这个问题了么

不用插件看看报错信息

不用插件,单独跑用例是好的

Different tests were collected between gw0 and gw1. The difference is: --- gw0_成都 - 阿木木的博客-CSDN博客
参照这个, data写成固定的吧,随机好像不行。

参考:https://github.com/pytest-dev/pytest-xdist/issues/432

“generating tests twice must produce identical tests” check

参数化的用例,使用固定数据。

xdist要求每个地方收集用例的时候必须一致,的确限制了并发的灵活性,当时我们为了解决这个问题自己实现了一套并发逻辑。如果不是特别有必要可以先遵从他的约束,不然就得自己改造了。

可以按文件级别并发,不用每个用例单独并发