如何在Django单元测试中模拟两用户同时预订同一车辆

问题背景 #

你当前的单元测试以串行方式执行两次预订请求，无法模拟真实场景下两个用户同时发起同一车辆预订的并发情况，需要改用多线程机制来模拟并发请求，验证预订逻辑的并发约束是否生效。

现有代码回顾 #

现有测试代码 #

def test_if_two_users_can_reserve_the_same_car_simultaneously(self):
    with patch.object(
        timezone,
        "now",
        return_value=make_aware(
            datetime.datetime.combine(
                datetime.date.today() + datetime.timedelta(days=1), datetime.time(10, 30, 0)
            ),
            timezone=pytz.timezone("UTC"),
        ),
    ):
        self.client.login(username=self.user.username, password=self.PASSWORD)
        url = reverse("booking-list")
        data = {
            "book_for": datetime.datetime.combine(
                datetime.date.today() + datetime.timedelta(days=1), datetime.time(11, 30, 0)
            ),
        }
        response = self.client.post(url, data=data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

    with patch.object(
        timezone,
        "now",
        return_value=make_aware(
            datetime.datetime.combine(
                datetime.date.today() + datetime.timedelta(days=1), datetime.time(10, 30, 0)
            ),
            timezone=pytz.timezone("UTC"),
        ),
    ):
        self.client.login(
            username=self.another_user.username, password=self.PASSWORD
        )
        url = reverse("booking-list")
        data = {
            "book_for": datetime.datetime.combine(
                datetime.date.today() + datetime.timedelta(days=1), datetime.time(11, 30, 0)
            ),
        }
        response = self.client.post(url, data=data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

Booking模型 #

class Booking(models.Model):
    id = models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True)
    user = models.ForeignKey(User, on_delete=models.SET_NULL, null=True)
    user_location = models.PointField(help_text="用户当前位置")
    user_address = models.TextField(help_text="用户当前地址")
    start_location = models.PointField(
        blank=True,
        help_text="用户取车地点",
    )
    start_address = models.TextField(
        blank=True,
        default="",
        help_text="用户取车地址",
    )
    end_location = models.PointField(
        null=True,
        blank=True,
        help_text="用户目的地位置",
    )
    end_address = models.TextField(
        blank=True,
        default="",
        help_text="用户目的地地址",
    )
    vehicle = models.ForeignKey(
        Vehicle,
        on_delete=models.PROTECT,
    )
    book_for = models.DateTimeField()

    drop_off_datetime = models.DateTimeField(
        null=True,
        blank=True,
        help_text="还车时间",
    )
    status = models.CharField(
        max_length=25,
        choices=BookingStatus.choices,
        default=BookingStatus.OPEN,
    )
    created_on = models.DateTimeField(auto_now_add=True)
    updated_on = models.DateTimeField(auto_now=True)

    class Meta:
        constraints = [
            UniqueConstraint(
                fields=["user"],
                condition=Q(status=BookingStatus.ACTIVE),
                name="unique_active_booking",
            )
        ]
        ordering = (
            "-book_for",
            "start_address",
        )

视图perform_create方法 #

def perform_create(self, serializer):
    serializer.save(user=self.request.user)

并发测试实现方案 #

使用Python的concurrent.futures.ThreadPoolExecutor创建多线程模拟并发请求，需注意以下关键点：

• 每个线程使用独立的Django测试客户端（Client），避免共享客户端导致状态混乱
• 确保所有线程的时间上下文一致，避免时序偏差
• 处理数据库连接的线程隔离，Django测试框架会自动为每个线程分配独立连接
• 补充模型中缺失的车辆并发预订约束（当前仅限制单用户活跃预订唯一）

修改后的并发测试代码 #

from concurrent.futures import ThreadPoolExecutor
from django.test import Client

def test_concurrent_car_booking(self):
    # 提前准备测试车辆（需在setUp中创建或确保数据库存在测试数据）
    target_vehicle = Vehicle.objects.first()
    # 统一预订时间和当前时间
    book_datetime = make_aware(
        datetime.datetime.combine(
            datetime.date.today() + datetime.timedelta(days=1), datetime.time(11, 30, 0)
        ),
        timezone=pytz.timezone("UTC"),
    )
    mock_now = make_aware(
        datetime.datetime.combine(
            datetime.date.today() + datetime.timedelta(days=1), datetime.time(10, 30, 0)
        ),
        timezone=pytz.timezone("UTC"),
    )

    def send_booking_request(user, password):
        # 每个线程创建独立客户端
        client = Client()
        # 登录当前用户
        client.login(username=user.username, password=password)
        # 统一补丁时间
        with patch.object(timezone, "now", return_value=mock_now):
            url = reverse("booking-list")
            data = {
                "book_for": book_datetime,
                "vehicle": target_vehicle.id,
                "user_location": "POINT(116.403874 39.914885)",
                "user_address": "测试地址",
                # 补充其他必填字段
            }
            response = client.post(url, data=data)
            return response.status_code

    # 启动两个并发线程
    with ThreadPoolExecutor(max_workers=2) as executor:
        future_user1 = executor.submit(send_booking_request, self.user, self.PASSWORD)
        future_user2 = executor.submit(send_booking_request, self.another_user, self.PASSWORD)
        
        # 获取请求结果
        status_user1 = future_user1.result()
        status_user2 = future_user2.result()

    # 验证结果：同一车辆只能有一个用户预订成功
    success_count = sum(1 for status in [status_user1, status_user2] if status == status.HTTP_201_CREATED)
    self.assertEqual(success_count, 1, "同一车辆不允许同时被两个用户预订")

关键补充说明 #

1. 模型约束完善：当前模型仅限制单用户的活跃预订唯一，需添加车辆的并发约束，示例如下：

# 在Booking模型的Meta.constraints中添加
UniqueConstraint(
    fields=["vehicle"],
    condition=Q(status__in=[BookingStatus.OPEN, BookingStatus.ACTIVE]) & 
              Q(book_for__lte=F('drop_off_datetime')),
    name="unique_vehicle_booking",
)

2. 必填字段检查：测试数据需包含模型所有必填字段，否则请求会因字段缺失返回400错误
3. 线程安全验证：务必为每个线程创建独立的Client实例，禁止共享测试类的self.client

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