1

我正在尝试使用 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"]
4

1 回答 1

0

我在这里做的事情与你所做的类似，但它可能看起来更简单或更优雅。在我看来，上下文似乎更倾向于“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 所示的函数：

然后在工作人员中，您也许可以利用 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']
于 2022-01-06T20:13:43.357 回答