@@ -1,6 +1,8 @@
|
||||
from decimal import Decimal
|
||||
from unittest.mock import patch
|
||||
|
||||
from django.conf import settings
|
||||
from django.db import IntegrityError
|
||||
from django.test import TestCase
|
||||
from rest_framework.test import APIClient
|
||||
|
||||
@@ -103,6 +105,21 @@ class SingleDeviceLoginTests(TestCase):
|
||||
self.assertEqual(active.count(), 1)
|
||||
self.assertEqual(active.get().user_agent, "Device B")
|
||||
|
||||
def test_login_succeeds_when_device_audit_write_fails(self):
|
||||
"""设备审计表异常不能回滚已经签发的登录 Token。"""
|
||||
with patch(
|
||||
"apps.accounts.views.LoginSession.objects.create",
|
||||
side_effect=IntegrityError("device audit unavailable"),
|
||||
):
|
||||
response = APIClient().post(
|
||||
"/api/auth/login/",
|
||||
{"username": self.user.username, "password": "strong-password"},
|
||||
format="json",
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("token", response.data)
|
||||
|
||||
|
||||
class InvitationFlowTests(TestCase):
|
||||
def _register(self, client, username, **extra):
|
||||
|
||||
@@ -99,7 +99,10 @@ def issue_single_device_token(request, user):
|
||||
LoginSession.objects.filter(user=locked_user, revoked_at__isnull=True).update(revoked_at=timezone.now())
|
||||
Token.objects.filter(user=locked_user).delete()
|
||||
token = Token.objects.create(user=locked_user)
|
||||
record_login_session(request, locked_user)
|
||||
# 设备记录只用于设置页展示和审计,不属于 Token 签发的关键路径。此前它在
|
||||
# 同一事务里即使被 record_login_session 捕获,数据库事务仍可能已被标记为
|
||||
# rollback,离开 atomic 后登录照样变成 500。
|
||||
record_login_session(request, locked_user)
|
||||
return token
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user