Django 测试用例中的线程未关闭数据库连接

Question

我正在尝试测试我为防止竞争条件而设置的代码是否正常工作。

在我的测试用例中，我启动了两个线程并让它们都调用有问题的代码。线程 return 正常，但 postgres 数据库连接似乎保持打开状态。

我已将整个测试用例缩减为一个非常简单的示例，其中出现了问题：

    def test_threading(self):
        obj = mommy.make(self.model_class)

        def test_concurrency():
            self.model_class.objects.get(id=obj.id)

        t1 = Thread(target=test_concurrency)
        t2 = Thread(target=test_concurrency)

        t1.start()
        t2.start()

        t1.join()
        t2.join()

测试运行后，测试数据库被销毁，出现以下错误：

psycopg2.errors.ObjectInUse: database "test_db" is being accessed by other users
DETAIL:  There are 2 other sessions using the database.

我尝试在测试用例末尾手动关闭连接，方法是添加：

for conn in db.connections.all():
    conn.close()

但这似乎没有任何效果。

我正在使用 django.test.TransactionTestCase 作为我的基本测试用例 class、Django 2.2 和 PostgreSQL 10.6。

Answer 1

通过更多的谷歌搜索解决了这个问题。 Django中的线程在线程执行完后需要手动关闭DB连接

子类线程：

from django.db import connection
from threading import Thread

class TestThread(Thread):

    def run(self):
        super().run()

        connection.close()

Django 测试用例中的线程未关闭数据库连接

Threads in Django Test Case are not closing DB Connections

django

django-testing