""" 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()