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

Question

我在异步代码中使用 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

我试过交换 sync_to_async与 database_sync_to_async来自 django-channels，但这并没有改变任何东西。

我怎样才能解决这个问题？

Answer 1

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

用于测试 django-channels我建议使用 pytest与 pytest-asyncio用于测试 channel 。当然还有 pytest-django .

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

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

有关如何测试 channel 代码的一些示例，请查看 here .