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 夹具就足够了？

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

编辑： 我的 conftest:py 有很多东西。 在 django_db_modify_db_settings_xdist_suffix 结束时，如果我记录 settings.DATABASES 它会显示正确和预期的信息。但是查询仍然会转到不同的数据库。 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 来扩展并发压力测试，并使用 gateway-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: 所示函数

然后在 worker 中你也许可以利用 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']