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_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
?
最重要的是,我如何模拟主路由器最终决定的内容,如文档中所述(我的重点)?除其他外,它不仅仅依赖一个路由器,它会连续调用两个路由器,然后如果我的自定义路由器 return None 会“做事”,因此通过单独调用它们来对我的路由器进行单元测试不会'真正复制主路由器的行为。
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_read
和 db_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))
我已经阅读了 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_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
?最重要的是,我如何模拟主路由器最终决定的内容,如文档中所述(我的重点)?除其他外,它不仅仅依赖一个路由器,它会连续调用两个路由器,然后如果我的自定义路由器 return None 会“做事”,因此通过单独调用它们来对我的路由器进行单元测试不会'真正复制主路由器的行为。
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_read
和 db_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))