Django 3.0 — 异步测试后数据库连接未关闭

Posted 2023-03-30

【中文标题】Django 3.0 — 异步测试后数据库连接未关闭

【英文标题】：Django 3.0 — database connections are not closed after async tests

【发布时间】：2020-02-15 19:33:51

【问题描述】：

我在异步代码中使用 Django ORM。一切正常，所有测试都通过。但是，数据库连接在测试后没有正确关闭。这是一个例子：

from asgiref.sync import sync_to_async, async_to_sync


@sync_to_async
def count_books():
    return Book.objects.count()


class FooTest(TestCase):
    def setUp(self):
        Book.objects.create(title='Haha')

    def test1(self):
        import asyncio
        c = asyncio.run(count_books())
        self.assertEqual(1, c)

    def test2(self):
        c = async_to_sync(count_books)()
        self.assertEqual(1, c)

Postgres 错误：

django.db.utils.OperationalError: database "test_mydbname" is being accessed by other users

Sqlite 错误：

sqlite3.OperationalError: database table is locked: test_mydbname

我尝试将 django-channels 中的 sync_to_async 与 database_sync_to_async 交换，但这并没有改变任何东西。

我该如何解决这个问题？

【问题讨论】：

你在尝试 django 3.0 使用电机吗？

【参考方案1】：

问题在于您的异步运行循环如何与主线程交互，自己处理会变得相当复杂。

对于测试django-channels，我建议使用pytest 和pytest-asyncio 来测试通道。当然还有pytest-django。

这将为测试异步代码提供一些有用的工具。

@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test1():
    count = await database_sync_to_async(Book.objects.count)
    ....

有关如何测试通道代码的一些示例，请查看here。

【讨论】：

谢谢！如果我根本不使用 django 频道怎么办？

我发现你需要测试 async python pytest-asyncio 应该是你的首选，Pytest 非常灵活，也可以调用你的常规 python UnitTest 类，因此你可以保留现有的测试代码放置非异步代码的地方，pytest 会很高兴地调用它。

关于 django3（orm 访问）django 还没有提供 database_sync_to_async 的等价物