测试Django应用并发能力时线程无法找到模型对象的问题

Django的TestCase默认给每个测试用例套了个数据库事务，测试结束自动回滚，用来保证测试之间互不干扰。主线程里setUp创建的MyModel实例只存在这个事务里，而你新开的线程会建立独立的数据库连接——PostgreSQL的事务隔离机制下，这个新连接看不到主线程事务里没提交的数据，所以调用process时查不到对应ID的记录，直接报错。

给你两个靠谱的解决办法，都能保住测试的隔离性：

1. 换用TransactionTestCase #

TransactionTestCase不用事务隔离测试，而是每个测试完成后清空所有数据库表。这样线程的数据库连接能正常看到setUp里创建的实例，改个继承类就行：

import threading
from django.test import TransactionTestCase  # 替换原来的TestCase

class test_smtg(TransactionTestCase):
    def setUp(self):
        self.my_model = MyModel()
        self.my_model.save()

    def test_single_processing(self):
        self.my_model.process()

    def test_concurrent_processing(self):
        thread = threading.Thread(target=self.my_model.process)
        thread.start()
        thread.join()  # 建议加join，等线程跑完再结束测试，不然没法验证结果

2. 线程里重新拉取实例 #

要是不想换测试类，就别直接把主线程的模型对象传给线程，而是把ID传过去，让线程自己从数据库重新查实例。不过要注意得先把主线程的事务提交，不然线程还是看不到：

import threading
from django.test import TestCase
from django.db import transaction

class test_smtg(TestCase):
    def setUp(self):
        self.my_model = MyModel()
        self.my_model.save()
        # 手动提交事务，让其他连接能读到这条数据
        transaction.commit()

    def test_concurrent_processing(self):
        def thread_task(model_id):
            my_model = MyModel.objects.get(id=model_id)
            my_model.process()

        thread = threading.Thread(target=thread_task, args=(self.my_model.id,))
        thread.start()
        thread.join()

⚠️ 注意：手动提交事务会破坏TestCase的自动回滚机制，可能导致测试之间的数据污染，所以优先选第一种方法。

测试并发一定要加thread.join()，不然测试主线程可能提前结束，根本没法确认process有没有正常执行完。

内容的提问来源于stack exchange，提问作者V. Brunelle