1

我已经阅读了Django 2.2 中关于数据库路由器的文档。我或多或少地理解这个概念 - 包括将某些表格放在一起 - 除了在实践中似乎有些棘手。

当事情变得复杂和相互依赖时，我的直觉是使用单元测试来逐渐让我的代码返回预期的结果。除了在这种情况下我不知道如何编写测试。

设置中的数据库：

DATABASES = {
    "default": {
        "ENGINE": constants.POSTGRES_ENGINE,
        "NAME": constants.SYSDBNAME,
        ...
    },
    "userdb": {
        "ENGINE": constants.POSTGRES_ENGINE,
        "NAME": constants.USERDBNAME,

}

设置中的路由器：


DATABASE_ROUTERS = [
    #the user database - userdb - which gets the auths and user-related stuff
    "bme.websec.database_routers.UserdbRouter",

    #the default database - sysdb - gets most of the other models
    "bme.websec.database_routers.SysdbMigrateRouter",
]

理想情况下，我会使用单元测试将我的所有模型一一提交给路由器的allow_migrate, db_for_read,db_for_write方法，并且对于每次调用，验证我是否得到了预期的结果。

但是有没有办法做到这一点？我想我可以使用

models = django.apps.apps.get_models(
        include_auto_created=True, include_swapped=True
    )

然后从单元测试中驱动这些方法调用。

但要采取只是

def allow_migrate(self, db, app_label, model_name=None, **hints):

  • 我如何知道何时拥有**hints以及是否model_name始终提供？

  • 最重要的是，我如何模拟主路由器最终决定的内容，如文档中所述（我的重点）？除其他外，它不仅仅依赖于一个路由器，它会连续调用两个路由器，然后如果我的自定义路由器返回 None 则“做事”，因此通过单独调用它们来对我的路由器进行单元测试并不能真正复制主路由器的行为.

Django 的数据库操作使用主路由器来分配数据库使用量。每当查询需要知道要使用哪个数据库时，它都会调用主路由器，提供模型和提示（如果可用）。然后 Django 依次尝试每个路由器，直到找到数据库建议。如果找不到建议，它会尝试提示实例的当前 _state.db。如果没有提供提示实例，或者该实例当前没有数据库状态，主路由器将分配默认数据库。

我已经得到db_for_read并且db_for_write主要表现得很好，但我正在努力让迁移正常工作：大多数模型最终都是在userdb而不是default.

到目前为止，我正在做的是针对 2 个空数据库运行迁移并使用 postgresql 检查表的创建位置。删除数据库，调整路由器，重新运行。有没有更好的方法来单元测试哪个数据库获得什么模型表进行迁移的实际决策（写入和读取很好，不是这个问题的主要原因）？

4

1 回答 1

3

主路由器只是 的一个实例django.db.utils.ConnectionRouter，因此您可以实例化它并调用它的方法来执行您的测试。

您可以在源代码 [GitHub]中看到该类具有的各种方法，以下是可能对您的测试有用的方法列表（实际上它们大多与数据库路由器上的方法相同）：

  • db_for_read: 与记录相同。返回用于读取模型的数据库
  • db_for_write: 与记录相同。返回用于编写模型的数据库
  • allow_relation: 与记录相同。返回是否允许两个实例之间的关系
  • allow_migrate: 与记录相同。返回某个应用程序的模型/模型是否可以迁移到某个数据库。
  • allow_migrate_model: 这只是一个 allow_migrate易于调用的小实用函数。

这是一个小片段，可以帮助您入门：

from django.db.utils import ConnectionRouter
from django.test import TestCase


class SimpleTest(TestCase):
    def test_router(self):
        router = ConnectionRouter()
        
        # Test the db used for read of some model
        self.assertEqual(router.db_for_read(SomeModel), "userdb")
        
        # Test the db used for write of some model
        self.assertEqual(router.db_for_write(SomeModel), "userdb")
        
        # Test if relation is allowed between two instances
        self.assertTrue(router.allow_relation(SomeModel(), SomeModel()))
        
        # Test if a model from some app is allowed to be migrated to some database
        self.assertTrue(router.allow_migrate_model("userdb", SomeModel._meta.app_label))
        
        # Test if some model is allowed to be migrated to some database
        self.assertTrue(router.allow_migrate_model("userdb", SomeModel))
        self.assertFalse(router.allow_migrate_model("userdb", SomeOtherModel))
于 2021-07-11T08:37:16.430 回答