Django 数据库路由器 - 你如何测试它们,尤其是迁移?

Django database routers - how do you test them, especially for migrations?

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

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

设置中的数据库:

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_migratedb_for_readdb_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):

The master router is used by Django’s database operations to allocate database usage. Whenever a query needs to know which database to use, it calls the master router, providing a model and a hint (if available). Django then tries each router in turn until a database suggestion can be found. If no suggestion can be found, it tries the current _state.db of the hint instance. If a hint instance wasn’t provided, or the instance doesn’t currently have database state, the master router will allocate the default database.

我的 db_for_readdb_for_write 主要是为了表现,但我正在努力让迁移正常工作:大多数模型最终都是 userdb 而不是 default.

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

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

你可以在 source code [GitHub], below is a list of the methods that are likely useful for your testing (in fact they mostly are the same as those on the database routers 中看到这个 class 的各种方法):

  • db_for_read :与记录的相同。 Returns 用于读取模型的数据库
  • db_for_write :与记录的相同。 Returns 用于编写模型的数据库
  • allow_relation :与记录的相同。 Returns 是否允许两个实例之间的关系
  • allow_migrate :与记录的相同。 Returns 是否可以将某个应用程序中的模型/模型迁移到某个数据库。
  • 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))