Pytest-xdist不尊重settings.DATABASES (或者我做错了什么)

Question

我正在尝试使用pytest-xdist来运行并行测试。这是有效的，我看到了更好的性能。

为了进一步改进，我希望使用django_db_modify_db_settings_xdist_suffix提供多个数据库。

我在我的conftest.py中重写了这个函数。

因为我有四个员工，所以我手工创建了四个数据库。然后，我使用conftest修改settings.DATABASES，以针对DBNAME_测试DBNAME

我验证了我的settings.DATABASES已经改变并且是正确的。

但是查询仍将转到旧的db DBNAME (不再在我的settings.DATABASES中)。

会有什么问题吗？

我还需要做别的改变吗？还是更换conftest.py夹具就足够了？

提前感谢您的帮助或指导。

编辑:我的自白:py有很多东西。在django_db_modify_db_settings_xdist_suffix的末尾，如果我记录了settings.DATABASES，就会显示正确和预期的信息。但是查询仍然转到不同的db。conftest.py在两次运行中都是相同的(pytest、-n4或pytest)。因为它依赖于xdist_suffix，所以它只修改"-n 4“中的settings.DATABASE值。在此，我认为有两项有关的职能是重要的：

@pytest.fixture(scope="session")
def django_db_setup(
    request,
    django_test_environment,
    django_db_blocker,
    django_db_use_migrations,
    django_db_keepdb,
    django_db_createdb,
    django_db_modify_db_settings,
):
    pass

和

@pytest.fixture(scope="session")
def django_db_modify_db_settings_xdist_suffix(request):
    
    from django.conf import settings

    default = settings.DATABASES["default"].copy()
    settings.DATABASES.clear()
    settings.DATABASES["default"] = default
    xdist_suffix = None
    xdist_worker_id = get_xdist_worker_id(request)
    if xdist_worker_id != 'master':
        xdist_suffix = xdist_worker_id
    if xdist_suffix:

        for db_settings in settings.DATABASES.values():
            test_name = db_settings.get("TEST", {}).get("NAME")

            if not test_name:
                test_name = "test_{}".format(db_settings["NAME"])

            db_settings.setdefault("TEST", {})
            db_settings["TEST"]["NAME"] = "{}_{}".format(test_name, xdist_suffix)
            db_settings["NAME"] = db_settings["TEST"]["NAME"]

Answer 1

我在这里做的事情与你所做的类似，但它可能看起来更简单或更优雅。在我看来，上下文似乎更倾向于'Django‘而不是pytest-xdist。

我使用pytest-xdist来扩展并发压力测试，并且使用网关id向远程工作人员发送一个允许区分节点的挂钩，这似乎与您的问题最相关。

def pytest_configure_node(self, node: WorkerController) -> None:
    """set something peculiar for each node."""
    node.workerinput['SPECIAL'] = get_db_for_node(node.gateway.id)

请尝试实现get_db_for_node(gateway_id: str) -> str所示的函数：

然后，在工作人员中，您可以利用config.workerinput来访问上面提到的特殊内容：

@pytest.fixture(scope="session")
def special(pytestconfig: Config) -> str:
    if not hasattr(pytestconfig, 'workerinput'):
        log.exception('fixture requires xdist')
        return ''
    return pytestconfig.workerinput['SPECIAL']