如何在Django中高效测试事务中的data anomalies？

针对你手动测试事务数据异常（脏读、不可重复读、幻读、丢失更新等）效率低的问题，推荐以下几种自动化方案：

1. 用多线程+Django测试框架模拟并发事务 #

利用Django的TestCase结合Python的threading模块，在测试代码中同时启动两个并发事务，一个作为测试事务，另一个模拟外部修改操作，完全替代手动操作。

示例代码（测试不可重复读）：

# tests/test_transaction_anomalies.py
from django.test import TestCase
from django.db import transaction
from store.models import Person
import threading
import time

class TransactionAnomalyTest(TestCase):
    fixtures = ['person_data.json']  # 预先加载测试数据（id=2的name为David）

    def test_non_repeatable_read(self):
        # 定义外部修改线程的逻辑
        def update_person():
            time.sleep(1)  # 等待测试事务执行第一次查询
            with transaction.atomic():
                person = Person.objects.get(pk=2)
                person.name = "Tom"
                person.save()

        # 启动修改线程
        update_thread = threading.Thread(target=update_person)
        update_thread.start()

        # 执行测试事务
        with transaction.atomic():
            first_read = Person.objects.get(pk=2).name
            time.sleep(2)  # 等待修改线程完成更新提交
            second_read = Person.objects.get(pk=2).name

        # 根据当前数据库隔离级别验证结果
        # 比如READ COMMITTED级别下，second_read会是"Tom"，即出现不可重复读
        self.assertEqual(first_read, "David")
        self.assertEqual(second_read, "Tom")

        update_thread.join()

运行测试命令：python manage.py test tests.test_transaction_anomalies

2. 封装为自定义Django管理命令 #

把测试逻辑写成自定义管理命令，直接通过命令行触发，无需请求视图或手动敲更新命令。

示例代码：

# store/management/commands/test_transaction_anomalies.py
from django.core.management.base import BaseCommand
from django.db import transaction
from store.models import Person
import threading
import time

class Command(BaseCommand):
    help = '测试事务中的数据异常'

    def handle(self, *args, **options):
        def update_person():
            time.sleep(1)
            with transaction.atomic():
                person = Person.objects.get(pk=2)
                person.name = "Tom"
                person.save()
                self.stdout.write(self.style.SUCCESS('已完成外部事务更新并提交'))

        update_thread = threading.Thread(target=update_person)
        update_thread.start()

        with transaction.atomic():
            self.stdout.write(f'第一次读取：{Person.objects.get(pk=2).name}')
            time.sleep(2)
            self.stdout.write(f'第二次读取：{Person.objects.get(pk=2).name}')

        update_thread.join()

运行命令：python manage.py test_transaction_anomalies

3. 切换数据库隔离级别批量验证不同异常 #

通过修改Django数据库配置的隔离级别，快速验证不同级别下是否会触发特定异常：

在settings.py中调整数据库选项：

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',  # 或其他数据库引擎
        'NAME': 'your_db',
        'USER': 'user',
        'PASSWORD': 'pass',
        'OPTIONS': {
            'isolation_level': 'read committed',  # 可选：repeatable read、serializable等
        },
    }
}

不同隔离级别对应异常情况：

• read committed：可能出现脏读（取决于数据库）、不可重复读、幻读
• repeatable read：避免脏读、不可重复读，可能出现幻读（如MySQL InnoDB）
• serializable：避免所有上述异常

结合前面的多线程测试代码，只需修改隔离级别，就能批量验证不同场景下的异常是否出现。

内容的提问来源于stack exchange，提问作者Super Kai - Kazuya Ito