Threads in Django Test Case are not closing DB Connections

Question

I am trying to test out that the code I put in place to prevent race conditions is working.

In my test case I've started two threads and have them both call the code in question. The threads return normally but it seems that the postgres database connections remain open.

I've reduced the entire test case to a very simple example where the problem shows up:

    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()

After the test runs and the test database is being destroyed I get the following error:

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

I've tried manually closing the connections at the end of the test case by adding:

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

But this doesn't seem to have any effect.

I am using django.test.TransactionTestCase as my base TestCase class, Django 2.2 and PostgreSQL 10.6.

Answer 1

Figured this out with a bit more googling. Threads in Django need to manually close DB connections after the thread is executed.

Subclass thread:

from django.db import connection
from threading import Thread

class TestThread(Thread):

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

        connection.close()