Django场景下如何测试cron定时任务执行时的并发冲突问题

问题根因 #

你之前用线程测试时报数据库会话冲突的错误，是因为Django默认测试用例的数据库连接不会在子线程运行结束后自动释放，导致测试结束销毁测试库时检测到仍有活跃连接。同时默认TestCase的事务隔离特性会导致子线程无法读取主线程中未提交的测试数据。

方案1：多线程测试（最贴近真实场景） #

实现要点 #

• 测试类继承TransactionTestCase而非默认TestCase，关闭单测试事务隔离
• 子线程运行结束后主动关闭所有数据库连接，释放会话
• 启动两个线程同时执行任务，等待全部运行完成后做断言

完整代码 #

import threading
import time
from django.test import TransactionTestCase
from unittest.mock import patch
# 导入你自己的业务代码
from your_app.module import task, MyModel

class TaskConcurrencyTest(TransactionTestCase):
    @patch("your_app.module.api_call")
    def test_no_duplicate_update_when_concurrent(self, mock_api_call):
        # 构造测试数据
        test_obj = MyModel.objects.create(foo=10, task_ran=False, bar=None)

        # 模拟API延迟响应
        def api_side_effect(foo_val):
            time.sleep(2)
            return foo_val + 1
        mock_api_call.side_effect = api_side_effect

        # 封装带连接清理的任务执行函数
        def run_task():
            try:
                task()
            finally:
                from django.db import connections
                connections.close_all()

        # 同时启动两个任务
        t1 = threading.Thread(target=run_task)
        t2 = threading.Thread(target=run_task)
        t1.start()
        t2.start()
        # 等待两个任务执行完成
        t1.join()
        t2.join()

        # 验证API仅调用一次，数据无重复更新
        mock_api_call.assert_called_once_with(10)
        test_obj.refresh_from_db()
        self.assertEqual(test_obj.bar, 11)
        self.assertTrue(test_obj.task_ran)

方案2：单线程模拟并发（更轻量，无数据库连接问题） #

不需要真实启动多线程，在第一次API调用的时机手动触发第二次任务执行，模拟前一个任务未完成、后一个任务启动的场景，更适合单元测试：

from django.test import TestCase
from unittest.mock import patch
from your_app.module import task, MyModel

class TaskConcurrencyTest(TestCase):
    @patch("your_app.module.api_call")
    def test_no_duplicate_update_when_concurrent(self, mock_api_call):
        test_obj = MyModel.objects.create(foo=10, task_ran=False, bar=None)
        call_count = 0

        def api_side_effect(foo_val):
            nonlocal call_count
            call_count += 1
            # 第一次调用API时触发第二次任务执行，模拟并发场景
            if call_count == 1:
                task()
            return foo_val + 1
        mock_api_call.side_effect = api_side_effect

        task()

        mock_api_call.assert_called_once_with(10)
        test_obj.refresh_from_db()
        self.assertEqual(test_obj.bar, 11)
        self.assertTrue(test_obj.task_ran)

注意：上述两种方案验证的前提是你已经对task函数做了并发控制（比如行锁、分布式锁、select_for_update等），如果你的原task代码没有加控制，测试会不通过。

内容的提问来源于stack exchange，提问作者andreashhp