"""用户域 4 条 happy-path 流程测试。""" import time from contextlib import ExitStack from unittest.mock import patch from django.core.cache import cache from rest_framework.test import APIClient from apps.user.throttling import ( SmsPhoneMinuteThrottle, SmsPhoneDayThrottle, SmsIpThrottle, RegisterIpThrottle, ResetPhoneThrottle, ) from .conftest import ( CacheTestCase, USER_SEND_CODE_URL, USER_REGISTER_URL, USER_LOGIN_URL, USER_LOGIN_CODE_URL, USER_CHANGE_PWD_URL, USER_RESET_PWD_URL, USER_ME_URL, USER_LIST_URL, user_detail_url, DEFAULT_INSTITUTION_CODE, DEFAULT_INSTITUTION_NAME, inject_sms_code, create_test_user, get_auth_client, get_tokens, create_teacher_student_relation, ) def _bypass_all_auth_throttles(stack): """在 ExitStack 中注册所有认证相关限流 bypass。""" for cls in (SmsPhoneMinuteThrottle, SmsPhoneDayThrottle, SmsIpThrottle, RegisterIpThrottle, ResetPhoneThrottle): stack.enter_context(patch.object(cls, 'allow_request', return_value=True)) class UserAuthHappyPathTest(CacheTestCase): """用户认证 happy-path 测试。""" def setUp(self): super().setUp() self.client = APIClient() # ── HP-1: 注册 → 密码登录 → /me ────────────────────────────────────── def test_flow_register_login_me(self): """HP-1: U2 register(管理员代注册,默认密码) → U3 login(默认密码) → GET /me""" phone = '13900000001' default_password = f'Pass{phone}' real_name = '张三' with ExitStack() as stack: _bypass_all_auth_throttles(stack) # U2: register（管理员代注册，无需验证码，密码自动为 Pass+手机号） resp = self.client.post(USER_REGISTER_URL, { 'phone': phone, 'real_name': real_name, 'institution_code': DEFAULT_INSTITUTION_CODE, 'institution_name': DEFAULT_INSTITUTION_NAME, }) self.assertEqual(resp.status_code, 201, resp.content) data = resp.json() self.assertIn('tokens', data) self.assertEqual(data['user']['phone'], phone) self.assertEqual(data['user']['real_name'], real_name) self.assertEqual(data['user']['institution_name'], DEFAULT_INSTITUTION_NAME) self.assertEqual(data['user']['institution_code'], DEFAULT_INSTITUTION_CODE) # U3: login (默认密码 Pass+手机号) resp = self.client.post(USER_LOGIN_URL, { 'phone': phone, 'password': default_password, }) self.assertEqual(resp.status_code, 200, resp.content) tokens = resp.json()['tokens'] # GET /me self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {tokens["access"]}') resp = self.client.get(USER_ME_URL) self.assertEqual(resp.status_code, 200, resp.content) self.assertEqual(resp.json()['phone'], phone) self.assertEqual(resp.json()['real_name'], real_name) # ── HP-2: 验证码登录 ───────────────────────────────────────────────── def test_flow_code_login(self): """HP-2: 预创建用户 → U1 send-code(login) → U4 login-code → /me""" phone = '13900000002' user = create_test_user(phone=phone, password='TestPass1') with ExitStack() as stack: _bypass_all_auth_throttles(stack) # U1: send-code (login) resp = self.client.post(USER_SEND_CODE_URL, { 'phone': phone, 'scene': 'login', }) self.assertEqual(resp.status_code, 200, resp.content) code = cache.get(f'sms:login:{phone}') self.assertIsNotNone(code) # U4: login-code（需要 institution 字段） resp = self.client.post(USER_LOGIN_CODE_URL, { 'phone': phone, 'code': str(code), 'institution_name': DEFAULT_INSTITUTION_NAME, 'institution_code': DEFAULT_INSTITUTION_CODE, }) self.assertEqual(resp.status_code, 200, resp.content) tokens = resp.json()['tokens'] self.assertIn('access', tokens) self.assertIn('refresh', tokens) self.assertFalse(resp.json()['is_new_user']) # GET /me self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {tokens["access"]}') resp = self.client.get(USER_ME_URL) self.assertEqual(resp.status_code, 200, resp.content) self.assertEqual(resp.json()['phone'], phone) # ── HP-3: 重置密码 ────────────────────────────────────────────────── def test_flow_reset_password(self): """HP-3: U1 send-code(reset) → U5 reset-password → U3 login(新密码)""" phone = '13900000003' old_pwd = 'OldPass1' new_pwd = 'NewPass1' create_test_user(phone=phone, password=old_pwd) with ExitStack() as stack: _bypass_all_auth_throttles(stack) # U1: send-code (reset) resp = self.client.post(USER_SEND_CODE_URL, { 'phone': phone, 'scene': 'reset', }) self.assertEqual(resp.status_code, 200, resp.content) code = cache.get(f'sms:reset:{phone}') self.assertIsNotNone(code) # U5: reset-password resp = self.client.post(USER_RESET_PWD_URL, { 'phone': phone, 'code': str(code), 'new_password': new_pwd, }) self.assertEqual(resp.status_code, 200, resp.content) # 新密码登录成功 resp = self.client.post(USER_LOGIN_URL, { 'phone': phone, 'password': new_pwd, }) self.assertEqual(resp.status_code, 200, resp.content) # 旧密码登录失败 resp = self.client.post(USER_LOGIN_URL, { 'phone': phone, 'password': old_pwd, }) self.assertIn(resp.status_code, (400, 401)) # ── HP-4: 修改密码 + 旧 token 失效 ────────────────────────────────── def test_flow_change_password(self): """HP-4: login → U6 change-password → 旧 token 失效 → 新密码 login""" phone = '13900000004' old_pwd = 'OldPass1' new_pwd = 'NewPass1' user = create_test_user(phone=phone, password=old_pwd) # U3: login resp = self.client.post(USER_LOGIN_URL, { 'phone': phone, 'password': old_pwd, }) self.assertEqual(resp.status_code, 200, resp.content) old_access = resp.json()['tokens']['access'] # U6: change-password self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {old_access}') resp = self.client.post(USER_CHANGE_PWD_URL, { 'old_password': old_pwd, 'new_password': new_pwd, }) self.assertEqual(resp.status_code, 200, resp.content) # 等待 1 秒：invalidate_user_tokens 写入 time()+1 time.sleep(1) # 旧 token 应被拒绝 self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {old_access}') resp = self.client.get(USER_ME_URL) self.assertEqual(resp.status_code, 401, f'旧 token 应失效: {resp.content}') # 新密码登录 self.client.credentials() # 清除旧 auth resp = self.client.post(USER_LOGIN_URL, { 'phone': phone, 'password': new_pwd, }) self.assertEqual(resp.status_code, 200, resp.content) new_access = resp.json()['tokens']['access'] # 新 token 正常 self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {new_access}') resp = self.client.get(USER_ME_URL) self.assertEqual(resp.status_code, 200, resp.content) class UserListDetailHappyPathTest(CacheTestCase): """U9 用户列表 + U10 用户详情 happy-path 测试。""" # ── HP-5: 管理员获取全部用户列表 ───────────────────────────────────── def test_admin_list_all_users(self): """HP-5: admin GET /users/ → 200，可见全部用户""" admin = create_test_user( phone='13900100001', password='Admin123', real_name='管理员', role_type='super_admin', ) stu1 = create_test_user( phone='13900100002', password='Stu12345', real_name='学生A', role_type='student', ) stu2 = create_test_user( phone='13900100003', password='Stu12345', real_name='学生B', role_type='student', ) client = get_auth_client(admin) resp = client.get(USER_LIST_URL) self.assertEqual(resp.status_code, 200, resp.content) data = resp.json() # DRF 分页：results 列表 results = data.get('results', data) result_ids = [u['id'] for u in results] self.assertIn(stu1.id, result_ids) self.assertIn(stu2.id, result_ids) self.assertIn(admin.id, result_ids) # ── HP-6: 教师仅看到自己名下学生 ───────────────────────────────────── def test_teacher_list_own_students_only(self): """HP-6: teacher GET /users/ → 200，仅包含名下活跃学生""" teacher = create_test_user( phone='13900100010', password='Teacher1', real_name='王老师', role_type='teacher', ) stu_own = create_test_user( phone='13900100011', password='Stu12345', real_name='我的学生', role_type='student', ) stu_other = create_test_user( phone='13900100012', password='Stu12345', real_name='其他学生', role_type='student', ) create_teacher_student_relation(teacher, stu_own, status=1) # stu_other 无关系 client = get_auth_client(teacher) resp = client.get(USER_LIST_URL) self.assertEqual(resp.status_code, 200, resp.content) data = resp.json() results = data.get('results', data) result_ids = [u['id'] for u in results] self.assertIn(stu_own.id, result_ids) self.assertNotIn(stu_other.id, result_ids) # 教师自己也不应出现在列表（queryset 过滤 role_type='student'） self.assertNotIn(teacher.id, result_ids) # ── HP-7: 教师不可见已结束关系的学生 ───────────────────────────────── def test_teacher_list_excludes_ended_relation(self): """HP-7: 已结束(status=0)的师生关系学生不出现在列表""" teacher = create_test_user( phone='13900100020', password='Teacher1', real_name='李老师', role_type='teacher', ) stu_active = create_test_user( phone='13900100021', password='Stu12345', real_name='活跃学生', role_type='student', ) stu_ended = create_test_user( phone='13900100022', password='Stu12345', real_name='已毕业学生', role_type='student', ) create_teacher_student_relation(teacher, stu_active, status=1) create_teacher_student_relation(teacher, stu_ended, status=0) client = get_auth_client(teacher) resp = client.get(USER_LIST_URL) self.assertEqual(resp.status_code, 200, resp.content) results = resp.json().get('results', resp.json()) result_ids = [u['id'] for u in results] self.assertIn(stu_active.id, result_ids) self.assertNotIn(stu_ended.id, result_ids) # ── HP-8: 管理员查看任意用户详情 ───────────────────────────────────── def test_admin_retrieve_any_user(self): """HP-8: admin GET /users/{id}/ → 200，可查看任意用户""" admin = create_test_user( phone='13900100030', password='Admin123', real_name='管理员', role_type='super_admin', ) student = create_test_user( phone='13900100031', password='Stu12345', real_name='某学生', role_type='student', ) client = get_auth_client(admin) resp = client.get(user_detail_url(student.id)) self.assertEqual(resp.status_code, 200, resp.content) self.assertEqual(resp.json()['id'], student.id) self.assertEqual(resp.json()['real_name'], '某学生') # ── HP-9: 用户查看自己的详情 ───────────────────────────────────────── def test_self_retrieve(self): """HP-9: student GET /users/{self.id}/ → 200，可查看自己""" student = create_test_user( phone='13900100040', password='Stu12345', real_name='自查学生', role_type='student', ) client = get_auth_client(student) resp = client.get(user_detail_url(student.id)) self.assertEqual(resp.status_code, 200, resp.content) self.assertEqual(resp.json()['id'], student.id) # ── HP-10: 教师查看名下学生详情 ────────────────────────────────────── def test_teacher_retrieve_own_student(self): """HP-10: teacher GET /users/{student.id}/ → 200，可查看名下学生""" teacher = create_test_user( phone='13900100050', password='Teacher1', real_name='赵老师', role_type='teacher', ) student = create_test_user( phone='13900100051', password='Stu12345', real_name='赵的学生', role_type='student', ) create_teacher_student_relation(teacher, student, status=1) client = get_auth_client(teacher) resp = client.get(user_detail_url(student.id)) self.assertEqual(resp.status_code, 200, resp.content) self.assertEqual(resp.json()['id'], student.id) # ── HP-11: 管理员列表支持过滤和搜索 ────────────────────────────────── def test_admin_list_filter_and_search(self): """HP-11: admin GET /users/?role_type=student&search=张 → 过滤生效""" admin = create_test_user( phone='13900100060', password='Admin123', real_name='管理员', role_type='super_admin', ) stu_zhang = create_test_user( phone='13900100061', password='Stu12345', real_name='张同学', role_type='student', ) stu_li = create_test_user( phone='13900100062', password='Stu12345', real_name='李同学', role_type='student', ) teacher = create_test_user( phone='13900100063', password='Teacher1', real_name='张老师', role_type='teacher', ) client = get_auth_client(admin) # 按 role_type 过滤 + search resp = client.get(USER_LIST_URL, {'role_type': 'student', 'search': '张'}) self.assertEqual(resp.status_code, 200, resp.content) results = resp.json().get('results', resp.json()) result_ids = [u['id'] for u in results] self.assertIn(stu_zhang.id, result_ids) self.assertNotIn(stu_li.id, result_ids) # 张老师 role_type=teacher，被 role_type=student 过滤掉 self.assertNotIn(teacher.id, result_ids)