zyc
416be408fd
All checks were successful
Build and Deploy Backend / build-and-deploy (push) Successful in 2m0s
951 lines
37 KiB
Python
951 lines
37 KiB
Python
"""
|
||
RTC_DEMO API 完整测试用例
|
||
覆盖所有API接口
|
||
运行方式: python manage.py test tests --verbosity=2 --keepdb
|
||
"""
|
||
import json
|
||
from datetime import date
|
||
from django.test import TestCase
|
||
from django.urls import reverse
|
||
from rest_framework.test import APITestCase, APIClient
|
||
from rest_framework import status
|
||
from apps.users.models import User
|
||
from apps.admins.models import AdminUser
|
||
from apps.spirits.models import Spirit
|
||
from apps.devices.models import DeviceType, DeviceBatch, Device, UserDevice
|
||
|
||
|
||
# ==================== App端测试 ====================
|
||
|
||
class UserAuthTests(APITestCase):
|
||
"""App端用户认证测试"""
|
||
|
||
def test_phone_login_new_user(self):
|
||
"""测试手机号一键登录 - 新用户"""
|
||
url = '/api/v1/auth/phone-login/'
|
||
data = {'phone': '13800138001'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertIn('token', response.data['data'])
|
||
self.assertTrue(response.data['data']['is_new_user'])
|
||
self.assertEqual(response.data['data']['user']['phone'], '13800138001')
|
||
|
||
def test_phone_login_existing_user(self):
|
||
"""测试手机号一键登录 - 已有用户"""
|
||
User.objects.create_user(phone='13800138002', nickname='测试用户')
|
||
|
||
url = '/api/v1/auth/phone-login/'
|
||
data = {'phone': '13800138002'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertFalse(response.data['data']['is_new_user'])
|
||
|
||
def test_phone_login_invalid_phone(self):
|
||
"""测试手机号格式验证"""
|
||
url = '/api/v1/auth/phone-login/'
|
||
data = {'phone': '1234567'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_phone_login_disabled_user(self):
|
||
"""测试禁用用户登录"""
|
||
User.objects.create_user(phone='13800138003', is_active=False)
|
||
|
||
url = '/api/v1/auth/phone-login/'
|
||
data = {'phone': '13800138003'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_token_refresh(self):
|
||
"""测试Token刷新"""
|
||
login_url = '/api/v1/auth/phone-login/'
|
||
login_response = self.client.post(login_url, {'phone': '13800138004'}, format='json')
|
||
refresh_token = login_response.data['data']['token']['refresh']
|
||
|
||
refresh_url = '/api/v1/auth/refresh/'
|
||
response = self.client.post(refresh_url, {'refresh': refresh_token}, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertIn('access', response.data['data'])
|
||
|
||
def test_token_refresh_invalid(self):
|
||
"""测试无效Token刷新"""
|
||
url = '/api/v1/auth/refresh/'
|
||
response = self.client.post(url, {'refresh': 'invalid_token'}, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_token_refresh_empty(self):
|
||
"""测试空Token刷新"""
|
||
url = '/api/v1/auth/refresh/'
|
||
response = self.client.post(url, {}, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
|
||
class UserProfileTests(APITestCase):
|
||
"""App端用户信息测试"""
|
||
|
||
def setUp(self):
|
||
self.user = User.objects.create_user(phone='13800138010', nickname='测试用户')
|
||
login_url = '/api/v1/auth/phone-login/'
|
||
response = self.client.post(login_url, {'phone': '13800138010'}, format='json')
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
def test_get_user_info(self):
|
||
"""测试获取用户信息"""
|
||
url = '/api/v1/users/me/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(response.data['data']['phone'], '13800138010')
|
||
|
||
def test_get_user_info_unauthorized(self):
|
||
"""测试未登录获取用户信息"""
|
||
self.client.credentials()
|
||
url = '/api/v1/users/me/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||
|
||
def test_update_user_info(self):
|
||
"""测试更新用户信息"""
|
||
url = '/api/v1/users/update_me/'
|
||
data = {'nickname': '新昵称'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(response.data['data']['nickname'], '新昵称')
|
||
|
||
def test_update_user_avatar(self):
|
||
"""测试更新用户头像"""
|
||
url = '/api/v1/users/update_me/'
|
||
data = {'avatar': 'https://example.com/avatar.jpg'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
|
||
class SpiritTests(APITestCase):
|
||
"""App端智能体测试"""
|
||
|
||
def setUp(self):
|
||
self.user = User.objects.create_user(phone='13800138020', nickname='测试用户')
|
||
login_url = '/api/v1/auth/phone-login/'
|
||
response = self.client.post(login_url, {'phone': '13800138020'}, format='json')
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
def test_create_spirit(self):
|
||
"""测试创建智能体"""
|
||
url = '/api/v1/spirits/'
|
||
data = {
|
||
'name': '测试心灵',
|
||
'prompt': '你是一个友好的助手',
|
||
'voice_id': 'voice_001'
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(response.data['data']['name'], '测试心灵')
|
||
|
||
def test_create_spirit_minimal(self):
|
||
"""测试创建智能体 - 最少字段"""
|
||
url = '/api/v1/spirits/'
|
||
data = {'name': '简单心灵'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
def test_list_spirits(self):
|
||
"""测试获取智能体列表"""
|
||
Spirit.objects.create(user=self.user, name='心灵1')
|
||
Spirit.objects.create(user=self.user, name='心灵2')
|
||
|
||
url = '/api/v1/spirits/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(len(response.data['data']), 2)
|
||
|
||
def test_list_spirits_empty(self):
|
||
"""测试获取空智能体列表"""
|
||
url = '/api/v1/spirits/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data['data']), 0)
|
||
|
||
def test_get_spirit_detail(self):
|
||
"""测试获取智能体详情"""
|
||
spirit = Spirit.objects.create(user=self.user, name='详情测试')
|
||
|
||
url = f'/api/v1/spirits/{spirit.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['name'], '详情测试')
|
||
|
||
def test_update_spirit(self):
|
||
"""测试更新智能体"""
|
||
spirit = Spirit.objects.create(user=self.user, name='原始名称')
|
||
|
||
url = f'/api/v1/spirits/{spirit.id}/'
|
||
data = {'name': '新名称', 'prompt': '新的提示词'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['name'], '新名称')
|
||
|
||
def test_delete_spirit(self):
|
||
"""测试删除智能体"""
|
||
spirit = Spirit.objects.create(user=self.user, name='待删除')
|
||
|
||
url = f'/api/v1/spirits/{spirit.id}/'
|
||
response = self.client.delete(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertFalse(Spirit.objects.filter(id=spirit.id).exists())
|
||
|
||
def test_spirit_isolation(self):
|
||
"""测试智能体用户隔离 - 不能访问其他用户的智能体"""
|
||
other_user = User.objects.create_user(phone='13800138021')
|
||
other_spirit = Spirit.objects.create(user=other_user, name='其他用户的心灵')
|
||
|
||
url = f'/api/v1/spirits/{other_spirit.id}/'
|
||
response = self.client.get(url)
|
||
|
||
# 应该返回404或权限错误
|
||
self.assertNotEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
|
||
class DeviceTests(APITestCase):
|
||
"""App端设备测试"""
|
||
|
||
def setUp(self):
|
||
self.user = User.objects.create_user(phone='13800138050', nickname='测试用户')
|
||
login_url = '/api/v1/auth/phone-login/'
|
||
response = self.client.post(login_url, {'phone': '13800138050'}, format='json')
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
self.device_type = DeviceType.objects.create(
|
||
brand='AL',
|
||
product_code='DZBJ-ON',
|
||
name='电子吧唧-联网版'
|
||
)
|
||
self.device = Device.objects.create(
|
||
sn='AL-DZBJ-ON-25W45-A01-00001',
|
||
device_type=self.device_type,
|
||
mac_address='AA:BB:CC:DD:EE:FF',
|
||
status='in_stock'
|
||
)
|
||
|
||
def test_query_by_mac(self):
|
||
"""测试通过MAC地址查询SN码（无需登录）"""
|
||
self.client.credentials() # 清除认证
|
||
url = '/api/v1/devices/query-by-mac/?mac=AA:BB:CC:DD:EE:FF'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(response.data['data']['sn'], 'AL-DZBJ-ON-25W45-A01-00001')
|
||
|
||
def test_query_by_mac_lowercase(self):
|
||
"""测试MAC地址查询 - 小写格式"""
|
||
self.client.credentials()
|
||
url = '/api/v1/devices/query-by-mac/?mac=aa:bb:cc:dd:ee:ff'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
def test_query_by_mac_dash_format(self):
|
||
"""测试MAC地址查询 - 横杠格式"""
|
||
self.client.credentials()
|
||
url = '/api/v1/devices/query-by-mac/?mac=AA-BB-CC-DD-EE-FF'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
def test_query_by_mac_not_found(self):
|
||
"""测试MAC地址查询 - 设备不存在"""
|
||
self.client.credentials()
|
||
url = '/api/v1/devices/query-by-mac/?mac=11:22:33:44:55:66'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
|
||
|
||
def test_query_by_mac_empty(self):
|
||
"""测试MAC地址查询 - 空参数"""
|
||
self.client.credentials()
|
||
url = '/api/v1/devices/query-by-mac/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_verify_device(self):
|
||
"""测试验证设备SN"""
|
||
url = '/api/v1/devices/verify/'
|
||
data = {'sn': 'AL-DZBJ-ON-25W45-A01-00001'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertTrue(response.data['data']['is_bindable'])
|
||
|
||
def test_verify_device_not_found(self):
|
||
"""测试验证设备SN - 不存在"""
|
||
url = '/api/v1/devices/verify/'
|
||
data = {'sn': 'NOT-EXIST-SN'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_bind_device(self):
|
||
"""测试绑定设备"""
|
||
spirit = Spirit.objects.create(user=self.user, name='测试心灵')
|
||
|
||
url = '/api/v1/devices/bind/'
|
||
data = {
|
||
'sn': 'AL-DZBJ-ON-25W45-A01-00001',
|
||
'spirit_id': spirit.id
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
# 验证设备状态
|
||
self.device.refresh_from_db()
|
||
self.assertEqual(self.device.status, 'bound')
|
||
|
||
def test_bind_device_without_spirit(self):
|
||
"""测试绑定设备 - 不绑定智能体"""
|
||
device2 = Device.objects.create(
|
||
sn='AL-DZBJ-ON-25W45-A01-00002',
|
||
device_type=self.device_type,
|
||
status='in_stock'
|
||
)
|
||
|
||
url = '/api/v1/devices/bind/'
|
||
data = {'sn': 'AL-DZBJ-ON-25W45-A01-00002'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
def test_my_devices(self):
|
||
"""测试我的设备列表"""
|
||
UserDevice.objects.create(user=self.user, device=self.device, is_active=True)
|
||
|
||
url = '/api/v1/devices/my_devices/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(len(response.data['data']), 1)
|
||
|
||
def test_my_devices_empty(self):
|
||
"""测试我的设备列表 - 空列表"""
|
||
url = '/api/v1/devices/my_devices/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data['data']), 0)
|
||
|
||
def test_unbind_device(self):
|
||
"""测试解绑设备"""
|
||
user_device = UserDevice.objects.create(user=self.user, device=self.device, is_active=True)
|
||
self.device.status = 'bound'
|
||
self.device.save()
|
||
|
||
url = f'/api/v1/devices/{user_device.id}/unbind/'
|
||
response = self.client.delete(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
user_device.refresh_from_db()
|
||
self.assertFalse(user_device.is_active)
|
||
|
||
def test_update_spirit_on_device(self):
|
||
"""测试更新设备绑定的智能体"""
|
||
spirit1 = Spirit.objects.create(user=self.user, name='心灵1')
|
||
spirit2 = Spirit.objects.create(user=self.user, name='心灵2')
|
||
user_device = UserDevice.objects.create(
|
||
user=self.user, device=self.device, spirit=spirit1, is_active=True
|
||
)
|
||
|
||
url = f'/api/v1/devices/{user_device.id}/update-spirit/'
|
||
data = {'spirit_id': spirit2.id}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
user_device.refresh_from_db()
|
||
self.assertEqual(user_device.spirit_id, spirit2.id)
|
||
|
||
|
||
# ==================== 管理端测试 ====================
|
||
|
||
class AdminAuthTests(APITestCase):
|
||
"""管理端认证测试"""
|
||
|
||
def setUp(self):
|
||
self.admin = AdminUser.objects.create_user(
|
||
username='admin',
|
||
password='admin123',
|
||
role='super_admin'
|
||
)
|
||
|
||
def test_admin_login(self):
|
||
"""测试管理员登录"""
|
||
url = '/api/admin/auth/login/'
|
||
data = {'username': 'admin', 'password': 'admin123'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertIn('token', response.data['data'])
|
||
self.assertEqual(response.data['data']['admin']['username'], 'admin')
|
||
|
||
def test_admin_login_wrong_password(self):
|
||
"""测试管理员登录 - 密码错误"""
|
||
url = '/api/admin/auth/login/'
|
||
data = {'username': 'admin', 'password': 'wrong'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_admin_login_user_not_exist(self):
|
||
"""测试管理员登录 - 用户不存在"""
|
||
url = '/api/admin/auth/login/'
|
||
data = {'username': 'notexist', 'password': 'password'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_admin_login_disabled(self):
|
||
"""测试管理员登录 - 账户禁用"""
|
||
disabled_admin = AdminUser.objects.create_user(
|
||
username='disabled', password='pass123', is_active=False
|
||
)
|
||
|
||
url = '/api/admin/auth/login/'
|
||
data = {'username': 'disabled', 'password': 'pass123'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
def test_admin_token_refresh(self):
|
||
"""测试管理员Token刷新"""
|
||
login_url = '/api/admin/auth/login/'
|
||
login_response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
|
||
)
|
||
refresh_token = login_response.data['data']['token']['refresh']
|
||
|
||
refresh_url = '/api/admin/auth/refresh/'
|
||
response = self.client.post(refresh_url, {'refresh': refresh_token}, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
|
||
class AdminProfileTests(APITestCase):
|
||
"""管理端个人信息测试"""
|
||
|
||
def setUp(self):
|
||
self.admin = AdminUser.objects.create_user(
|
||
username='admin', password='admin123', role='admin'
|
||
)
|
||
login_url = '/api/admin/auth/login/'
|
||
response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
|
||
)
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
def test_get_admin_profile(self):
|
||
"""测试获取管理员信息"""
|
||
url = '/api/admin/profile/me/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['username'], 'admin')
|
||
|
||
def test_change_password(self):
|
||
"""测试修改密码"""
|
||
url = '/api/admin/profile/change-password/'
|
||
data = {'old_password': 'admin123', 'new_password': 'newpass123'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
# 验证新密码可以登录
|
||
self.client.credentials()
|
||
login_url = '/api/admin/auth/login/'
|
||
login_response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'newpass123'}, format='json'
|
||
)
|
||
self.assertEqual(login_response.data['code'], 0)
|
||
|
||
def test_change_password_wrong_old(self):
|
||
"""测试修改密码 - 原密码错误"""
|
||
url = '/api/admin/profile/change-password/'
|
||
data = {'old_password': 'wrongpass', 'new_password': 'newpass123'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertNotEqual(response.data['code'], 0)
|
||
|
||
|
||
class AdminUserManageTests(APITestCase):
|
||
"""管理员用户管理测试"""
|
||
|
||
def setUp(self):
|
||
self.super_admin = AdminUser.objects.create_user(
|
||
username='superadmin', password='super123', role='super_admin'
|
||
)
|
||
login_url = '/api/admin/auth/login/'
|
||
response = self.client.post(
|
||
login_url, {'username': 'superadmin', 'password': 'super123'}, format='json'
|
||
)
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
def test_list_admins(self):
|
||
"""测试管理员列表"""
|
||
url = '/api/admin/admins/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
def test_create_admin(self):
|
||
"""测试创建管理员"""
|
||
url = '/api/admin/admins/'
|
||
data = {
|
||
'username': 'newadmin',
|
||
'password': 'newpass123',
|
||
'name': '新管理员',
|
||
'role': 'operator'
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['username'], 'newadmin')
|
||
|
||
def test_get_admin_detail(self):
|
||
"""测试获取管理员详情"""
|
||
admin = AdminUser.objects.create_user(username='testadmin', password='test123')
|
||
|
||
url = f'/api/admin/admins/{admin.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['username'], 'testadmin')
|
||
|
||
def test_toggle_admin_status(self):
|
||
"""测试启用/禁用管理员"""
|
||
admin = AdminUser.objects.create_user(username='testadmin2', password='test123')
|
||
|
||
url = f'/api/admin/admins/{admin.id}/toggle-status/'
|
||
response = self.client.post(url, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
admin.refresh_from_db()
|
||
self.assertFalse(admin.is_active)
|
||
|
||
def test_reset_admin_password(self):
|
||
"""测试重置管理员密码"""
|
||
admin = AdminUser.objects.create_user(username='testadmin3', password='oldpass')
|
||
|
||
url = f'/api/admin/admins/{admin.id}/reset-password/'
|
||
data = {'new_password': 'resetpass123'}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
|
||
|
||
class AdminDeviceTypeTests(APITestCase):
|
||
"""管理端设备类型测试"""
|
||
|
||
def setUp(self):
|
||
self.admin = AdminUser.objects.create_user(
|
||
username='admin', password='admin123', role='super_admin'
|
||
)
|
||
login_url = '/api/admin/auth/login/'
|
||
response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
|
||
)
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
def test_create_device_type_network(self):
|
||
"""测试创建联网设备类型"""
|
||
url = '/api/admin/device-types/'
|
||
data = {
|
||
'brand': 'AL',
|
||
'product_code': 'DZBJ-ON',
|
||
'name': '电子吧唧-联网版'
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertTrue(response.data['data']['is_network_required'])
|
||
|
||
def test_create_device_type_offline(self):
|
||
"""测试创建非联网设备类型"""
|
||
url = '/api/admin/device-types/'
|
||
data = {
|
||
'brand': 'AL',
|
||
'product_code': 'DZBJ-OFF',
|
||
'name': '电子吧唧-离线版'
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertFalse(response.data['data']['is_network_required'])
|
||
|
||
def test_list_device_types(self):
|
||
"""测试设备类型列表"""
|
||
DeviceType.objects.create(brand='AL', product_code='TEST-ON', name='测试设备')
|
||
|
||
url = '/api/admin/device-types/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
def test_get_device_type_detail(self):
|
||
"""测试设备类型详情"""
|
||
dt = DeviceType.objects.create(brand='AL', product_code='TEST-DT', name='测试类型')
|
||
|
||
url = f'/api/admin/device-types/{dt.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['name'], '测试类型')
|
||
|
||
def test_update_device_type(self):
|
||
"""测试更新设备类型"""
|
||
dt = DeviceType.objects.create(brand='AL', product_code='TEST-UP', name='原名称')
|
||
|
||
url = f'/api/admin/device-types/{dt.id}/'
|
||
data = {'name': '新名称'}
|
||
response = self.client.put(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['data']['name'], '新名称')
|
||
|
||
|
||
class AdminBatchTests(APITestCase):
|
||
"""管理端批次测试"""
|
||
|
||
def setUp(self):
|
||
self.admin = AdminUser.objects.create_user(
|
||
username='admin', password='admin123', role='super_admin'
|
||
)
|
||
login_url = '/api/admin/auth/login/'
|
||
response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
|
||
)
|
||
self.access_token = response.data['data']['token']['access']
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.access_token}')
|
||
|
||
self.device_type = DeviceType.objects.create(
|
||
brand='AL', product_code='DZBJ-ON', name='电子吧唧-联网版'
|
||
)
|
||
|
||
def test_create_batch_and_generate_sn(self):
|
||
"""测试创建批次并生成SN码"""
|
||
url = '/api/admin/device-batches/'
|
||
data = {
|
||
'device_type': self.device_type.id,
|
||
'batch_no': 'A01',
|
||
'production_date': '2026-01-28',
|
||
'quantity': 10,
|
||
'remark': '测试批次'
|
||
}
|
||
response = self.client.post(url, data, format='json')
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
self.assertEqual(response.data['data']['quantity'], 10)
|
||
self.assertIn('sn_range', response.data['data'])
|
||
|
||
# 验证SN码格式
|
||
sn_start = response.data['data']['sn_range']['start']
|
||
self.assertTrue(sn_start.startswith('AL-DZBJ-ON-'))
|
||
|
||
# 验证设备数量
|
||
device_count = Device.objects.filter(batch__batch_no='A01').count()
|
||
self.assertEqual(device_count, 10)
|
||
|
||
def test_list_batches(self):
|
||
"""测试批次列表"""
|
||
url = '/api/admin/device-batches/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(response.data['code'], 0)
|
||
|
||
def test_get_batch_detail(self):
|
||
"""测试批次详情"""
|
||
batch = DeviceBatch.objects.create(
|
||
device_type=self.device_type,
|
||
batch_no='B01',
|
||
production_date=date(2026, 1, 28),
|
||
quantity=5
|
||
)
|
||
|
||
url = f'/api/admin/device-batches/{batch.id}/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertIn('statistics', response.data['data'])
|
||
|
||
def test_get_batch_devices(self):
|
||
"""测试批次设备列表"""
|
||
batch = DeviceBatch.objects.create(
|
||
device_type=self.device_type,
|
||
batch_no='C01',
|
||
production_date=date(2026, 1, 28),
|
||
quantity=3
|
||
)
|
||
Device.objects.create(sn='TEST-001', batch=batch, device_type=self.device_type)
|
||
Device.objects.create(sn='TEST-002', batch=batch, device_type=self.device_type)
|
||
|
||
url = f'/api/admin/device-batches/{batch.id}/devices/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||
self.assertEqual(len(response.data['data']['items']), 2)
|
||
|
||
|
||
class AuthSeparationTests(APITestCase):
|
||
"""验证App和Admin认证分离"""
|
||
|
||
def setUp(self):
|
||
# 创建App用户
|
||
self.app_user = User.objects.create_user(phone='13800138099')
|
||
login_url = '/api/v1/auth/phone-login/'
|
||
response = self.client.post(login_url, {'phone': '13800138099'}, format='json')
|
||
self.app_token = response.data['data']['token']['access']
|
||
|
||
# 创建Admin用户
|
||
self.admin = AdminUser.objects.create_user(
|
||
username='admin', password='admin123', role='admin'
|
||
)
|
||
login_url = '/api/admin/auth/login/'
|
||
response = self.client.post(
|
||
login_url, {'username': 'admin', 'password': 'admin123'}, format='json'
|
||
)
|
||
self.admin_token = response.data['data']['token']['access']
|
||
|
||
def test_app_token_cannot_access_admin_api(self):
|
||
"""测试App Token无法访问管理端API"""
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.app_token}')
|
||
|
||
url = '/api/admin/device-types/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertNotEqual(response.data.get('code', 0), 0)
|
||
|
||
def test_admin_token_cannot_access_app_api(self):
|
||
"""测试Admin Token无法访问App端API"""
|
||
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.admin_token}')
|
||
|
||
url = '/api/v1/users/me/'
|
||
response = self.client.get(url)
|
||
|
||
self.assertNotEqual(response.data.get('code', 0), 0)
|
||
|
||
def test_no_token_cannot_access_protected_api(self):
|
||
"""测试无Token无法访问受保护API"""
|
||
self.client.credentials()
|
||
|
||
url = '/api/v1/users/me/'
|
||
response = self.client.get(url)
|
||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||
|
||
url = '/api/admin/device-types/'
|
||
response = self.client.get(url)
|
||
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||
|
||
|
||
# ==================== Log Center 集成测试 ====================
|
||
|
||
class LogCenterIntegrationTests(TestCase):
|
||
"""Log Center 错误上报集成测试"""
|
||
|
||
def test_report_to_log_center_function_exists(self):
|
||
"""测试 report_to_log_center 函数存在"""
|
||
from utils.exceptions import report_to_log_center
|
||
self.assertTrue(callable(report_to_log_center))
|
||
|
||
def test_report_to_log_center_with_exception(self):
|
||
"""测试上报异常到 Log Center"""
|
||
from utils.exceptions import report_to_log_center
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
# 创建测试异常
|
||
try:
|
||
raise ValueError("Test error message")
|
||
except ValueError as e:
|
||
test_exc = e
|
||
|
||
# Mock context
|
||
mock_request = MagicMock()
|
||
mock_request.path = '/api/test/'
|
||
mock_request.method = 'GET'
|
||
context = {'request': mock_request, 'view': 'TestView'}
|
||
|
||
# Mock requests.post
|
||
with patch('utils.exceptions.requests.post') as mock_post:
|
||
with patch('utils.exceptions.LOG_CENTER_ENABLED', True):
|
||
report_to_log_center(test_exc, context)
|
||
|
||
# 等待线程执行（测试中同步执行更可靠）
|
||
import time
|
||
time.sleep(0.5)
|
||
|
||
def test_report_payload_structure(self):
|
||
"""测试上报 payload 结构正确"""
|
||
from utils.exceptions import report_to_log_center
|
||
from unittest.mock import patch, MagicMock
|
||
import json
|
||
|
||
captured_payload = None
|
||
|
||
def capture_post(url, json=None, timeout=None):
|
||
nonlocal captured_payload
|
||
captured_payload = json
|
||
return MagicMock(status_code=200)
|
||
|
||
try:
|
||
raise TypeError("Type mismatch error")
|
||
except TypeError as e:
|
||
test_exc = e
|
||
|
||
mock_request = MagicMock()
|
||
mock_request.path = '/api/users/me/'
|
||
mock_request.method = 'POST'
|
||
context = {'request': mock_request, 'view': 'UserView'}
|
||
|
||
with patch('utils.exceptions.requests.post', side_effect=capture_post):
|
||
with patch('utils.exceptions.LOG_CENTER_ENABLED', True):
|
||
with patch('utils.exceptions.threading.Thread') as mock_thread:
|
||
# 直接调用 target 函数而不是启动线程
|
||
mock_thread_instance = MagicMock()
|
||
def run_target(*args, **kwargs):
|
||
target = kwargs.get('target') or args[0]
|
||
target()
|
||
mock_thread.side_effect = lambda *args, **kwargs: MagicMock(
|
||
start=lambda: run_target(*args, **kwargs),
|
||
daemon=True
|
||
)
|
||
|
||
report_to_log_center(test_exc, context)
|
||
|
||
# 验证 payload 结构
|
||
if captured_payload:
|
||
self.assertEqual(captured_payload['project_id'], 'rtc_backend')
|
||
self.assertEqual(captured_payload['level'], 'ERROR')
|
||
self.assertEqual(captured_payload['error']['type'], 'TypeError')
|
||
self.assertEqual(captured_payload['error']['message'], 'Type mismatch error')
|
||
self.assertIn('stack_trace', captured_payload['error'])
|
||
self.assertEqual(captured_payload['context']['url'], '/api/users/me/')
|
||
self.assertEqual(captured_payload['context']['method'], 'POST')
|
||
|
||
def test_report_disabled_when_flag_off(self):
|
||
"""测试关闭开关时不上报"""
|
||
from utils.exceptions import report_to_log_center
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
try:
|
||
raise Exception("Should not be reported")
|
||
except Exception as e:
|
||
test_exc = e
|
||
|
||
with patch('utils.exceptions.requests.post') as mock_post:
|
||
with patch('utils.exceptions.LOG_CENTER_ENABLED', False):
|
||
report_to_log_center(test_exc, {})
|
||
# 应该不调用 requests.post
|
||
mock_post.assert_not_called()
|
||
|
||
def test_report_silent_failure(self):
|
||
"""测试上报失败不抛异常"""
|
||
from utils.exceptions import report_to_log_center
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
try:
|
||
raise Exception("Test exception")
|
||
except Exception as e:
|
||
test_exc = e
|
||
|
||
with patch('utils.exceptions.requests.post', side_effect=Exception("Network error")):
|
||
with patch('utils.exceptions.LOG_CENTER_ENABLED', True):
|
||
# 不应抛出异常
|
||
try:
|
||
report_to_log_center(test_exc, {})
|
||
except Exception:
|
||
self.fail("report_to_log_center should not raise exceptions")
|
||
|
||
|
||
class ExceptionHandlerIntegrationTests(APITestCase):
|
||
"""异常处理器集成测试 - 验证异常时触发 Log Center 上报"""
|
||
|
||
def test_exception_triggers_log_center_report(self):
|
||
"""测试异常触发 Log Center 上报"""
|
||
from utils.exceptions import custom_exception_handler
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
# 创建异常
|
||
test_exc = ValueError("Database connection failed")
|
||
|
||
# Mock context
|
||
mock_request = MagicMock()
|
||
mock_request.path = '/api/test/'
|
||
mock_request.method = 'GET'
|
||
context = {'request': mock_request, 'view': MagicMock()}
|
||
|
||
with patch('utils.exceptions.report_to_log_center') as mock_report:
|
||
# 调用异常处理器
|
||
custom_exception_handler(test_exc, context)
|
||
|
||
# 验证调用了上报函数
|
||
mock_report.assert_called_once()
|
||
call_args = mock_report.call_args
|
||
self.assertEqual(call_args[0][0], test_exc)
|
||
|
||
def test_business_exception_not_reported(self):
|
||
"""测试业务异常不上报到 Log Center"""
|
||
from utils.exceptions import custom_exception_handler, BusinessException
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
# 创建业务异常
|
||
biz_exc = BusinessException(code=100, message="用户不存在")
|
||
|
||
context = {'request': MagicMock(), 'view': MagicMock()}
|
||
|
||
with patch('utils.exceptions.report_to_log_center') as mock_report:
|
||
custom_exception_handler(biz_exc, context)
|
||
|
||
# 业务异常不应触发上报
|
||
mock_report.assert_not_called()
|
||
|