使用异步查询时无法销毁Django测试数据库的问题求助
问题描述
我有一个配置了PostgreSQL数据库的Django项目,requirements.txt中的相关依赖如下:
Django==5.0.1 psycopg2-binary==2.9.9
定义的Item模型代码:
class Item(models.Model): name = models.CharField(max_length=100)
测试用例代码:
class ItemTestCase(TestCase): def test_create(self): asyncio.run(Item.objects.acreate(name="Test")) self.assertEqual(Item.objects.count(), 1)
运行python manage.py test时测试本身能通过,但测试结束后出现异常,提示无法销毁测试数据库:
2024-01-18 19:05:03 django.db.utils.OperationalError: database "test_axpress" is being accessed by other users
2024-01-18 19:05:03 DETAIL: There is 1 other session using the database.
需要解决的问题:如何确保销毁数据库前关闭所有活跃连接?
解决方案
问题根源是异步ORM操作创建的数据库连接未被正确回收,导致测试结束后PostgreSQL仍有活跃会话占用测试库。以下是几种可行的解决方法:
方法1:使用Django官方异步测试类
Django 5.0原生支持异步测试,无需手动调用asyncio.run()。改用AsyncTestCase并定义异步测试方法,Django会自动管理连接的创建与关闭:
from django.test import AsyncTestCase class ItemTestCase(AsyncTestCase): async def test_create(self): await Item.objects.acreate(name="Test") self.assertEqual(Item.objects.count(), 1)
方法2:测试结束后手动关闭所有连接
如果必须保留asyncio.run()的写法,可以在测试用例的tearDown方法中主动关闭所有数据库连接:
class ItemTestCase(TestCase): def test_create(self): asyncio.run(Item.objects.acreate(name="Test")) self.assertEqual(Item.objects.count(), 1) def tearDown(self): from django.db import connections # 遍历并关闭所有数据库连接 for connection in connections.all(): connection.close()
方法3:强制断开PostgreSQL连接(临时方案)
如果前两种方法无效,可以通过SQL强制终止测试库的所有活跃会话,再销毁数据库。可以在测试的tearDown中执行该SQL:
def tearDown(self): from django.db import connection with connection.cursor() as cursor: cursor.execute(""" SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = 'test_axpress' AND pid != pg_backend_pid(); """) connection.close()
注意:这种方法属于强制清理,可能影响其他无关会话,仅作为临时应急方案使用。
内容的提问来源于stack exchange,提问作者Artiom
相关产品推荐
相关产品推荐

