"""D8 测试共享工具:URL 常量、用户/科室夹具、SMS 注入、载荷构建。""" from django.core.cache import cache from django.core.files.uploadedfile import SimpleUploadedFile from django.test import TestCase, TransactionTestCase from rest_framework.test import APIClient from rest_framework_simplejwt.tokens import RefreshToken from apps.user.models import User, Institution, Department, TeacherStudentRelation class CacheTestCase(TestCase): """所有测试基类:setUp 自动 clear Redis,隔离 SMS 验证码/限流/JWT 黑名单。""" def setUp(self): super().setUp() cache.clear() class CacheTransactionTestCase(TransactionTestCase): """事务测试基类:setUp 自动 clear Redis。""" def setUp(self): super().setUp() cache.clear() # ─── URL 常量 ───────────────────────────────────────────────────────────────── # 用户认证 USER_SEND_CODE_URL = '/api/user/auth/send-code/' USER_REGISTER_URL = '/api/user/auth/register/' USER_LOGIN_URL = '/api/user/auth/login/' USER_LOGIN_CODE_URL = '/api/user/auth/login-code/' USER_LOGOUT_URL = '/api/user/auth/logout/' USER_REFRESH_URL = '/api/user/auth/refresh/' USER_RESET_PWD_URL = '/api/user/auth/reset-password/' USER_CHANGE_PWD_URL = '/api/user/users/change-password/' USER_ME_URL = '/api/user/users/me/' USER_LIST_URL = '/api/user/users/' USER_INSTITUTION_LIST_URL = '/api/user/institution_list/' USER_INSTITUTION_INFO_URL = '/api/user/institution_info/' USER_MY_DEPARTMENTS_URL = '/api/user/my_departments/' USER_PROFILE_CONFIG_URL = '/api/user/profile/config/' # 病例 CASE_PARSE_URL = '/api/case/cases/parse-pdf/' CASE_GENERATE_RULES_URL = '/api/case/cases/generate-scoring-rules/' CASE_FULL_CREATE_URL = '/api/case/cases/full-create/' def user_detail_url(user_id): return f'/api/user/users/{user_id}/' def case_full_url(case_id): return f'/api/case/cases/{case_id}/full/' # ─── SMS 注入 ───────────────────────────────────────────────────────────────── def inject_sms_code(phone, scene, code='123456'): """直接将验证码写入 Redis,绕过 SMS 服务。""" cache.set(f'sms:{scene}:{phone}', code, timeout=300) # ─── 默认测试机构 ───────────────────────────────────────────────────────────── DEFAULT_INSTITUTION_CODE = 'TEST-HOSP-001' DEFAULT_INSTITUTION_NAME = '测试医院' # ─── 用户工具 ───────────────────────────────────────────────────────────────── def create_test_user(phone='13900000001', password='TestPass1', real_name='测试用户', role_type='student', status=1, institution=None): """创建测试用户(已知密码),返回 User 实例。""" user = User.objects.create_user( username=phone, password=password, phone=phone, real_name=real_name, role_type=role_type, status=status, institution=institution, ) return user def get_tokens(user): """返回 {'access': '...', 'refresh': '...'} 字符串。""" refresh = RefreshToken.for_user(user) return {'access': str(refresh.access_token), 'refresh': str(refresh)} def get_auth_client(user): """返回已携带 JWT Bearer 的 APIClient。""" tokens = get_tokens(user) client = APIClient() client.credentials(HTTP_AUTHORIZATION=f'Bearer {tokens["access"]}') return client # ─── 师生关系 ───────────────────────────────────────────────────────────────── def create_teacher_student_relation(teacher, student, status=1): """创建师生关系记录。""" return TeacherStudentRelation.objects.create( teacher=teacher, student=student, relation_type='指导', status=status, ) # ─── 科室工具 ───────────────────────────────────────────────────────────────── def ensure_institution(name='测试医院', code='TEST-HOSP-001'): inst, _ = Institution.objects.get_or_create( code=code, defaults={'name': name, 'type': 'hospital', 'province': '北京', 'city': '北京'}, ) return inst def ensure_department(name='儿科'): """科室为全局表,与机构无关。""" dept, _ = Department.objects.get_or_create( name=name, defaults={'category': '临床'}, ) return dept # ─── 病例载荷构建 ───────────────────────────────────────────────────────────── def sample_exam_items(): """示例检查项(用于 C3 full-create)。""" return [ { 'item_code': 'blood_routine', 'item_name': '血常规', 'item_type': 'lab', 'category': '实验室检查', 'result_text': 'WBC 10×10^9/L', 'result_structured': {'wbc': '10×10^9/L'}, 'is_key': True, 'is_abnormal': False, 'score_weight': 1.0, 'display_order': 1, }, ] def build_traditional_payload(department_name='儿科', scoring_rules_count=2, with_exam_items=False): """构建合法的传统病例 full-create 载荷。""" rules = [ { 'dimension': f'测试维度{i + 1}', 'score_weight': round(1.0 / scoring_rules_count, 2), 'ai_auto_score': True, 'scoring_standard': f'评分标准{i + 1}', } for i in range(scoring_rules_count) ] payload = { 'title': '测试传统病例-表单录入', 'case_type': 'traditional', 'difficulty': 'medium', 'chief_complaint': '发热 3 天', 'description': '患儿,男,4 岁,因发热 3 天就诊。', 'patient_age': 4, 'patient_gender': 'male', 'department_name': department_name, 'estimated_minutes': 30, 'osce_enabled': False, 'tags': '儿科,发热', 'traditional': { 'standard_diagnosis': '上呼吸道感染', 'standard_treatment': '对症治疗,退热处理', 'guideline_reference': '《儿科学》第 9 版', }, 'scoring_rules': rules, } if with_exam_items: payload['exam_items'] = sample_exam_items() return payload def build_teaching_payload(department_name='儿科', scoring_rules_count=2): """构建合法的教学病例 full-create 载荷。""" rules = [ { 'dimension': f'教学维度{i + 1}', 'score_weight': round(1.0 / scoring_rules_count, 2), 'ai_auto_score': False, 'scoring_standard': f'教学评分标准{i + 1}', } for i in range(scoring_rules_count) ] return { 'title': '测试教学病例-表单录入', 'case_type': 'teaching', 'difficulty': 'hard', 'chief_complaint': '腹痛 2 天', 'description': '患者,女,28 岁,因腹痛 2 天就诊。', 'patient_age': 28, 'patient_gender': 'female', 'department_name': department_name, 'estimated_minutes': 45, 'osce_enabled': False, 'teaching': { 'teaching_goal': '掌握急腹症鉴别诊断', 'discussion_questions': '如何鉴别急性阑尾炎与其他急腹症?', 'teacher_guide': '引导学生按 SOAP 格式分析', 'scoring_focus': '鉴别诊断思路', }, 'scoring_rules': rules, } # ─── AI Mock 数据 ───────────────────────────────────────────────────────────── MOCK_C1_PARSE_RESULT = { 'data': { 'title': 'Mock-儿科发热病例', 'case_type': 'traditional', 'difficulty': 'medium', 'chief_complaint': '发热 3 天', 'description': '患儿,男,4 岁,因发热 3 天就诊。', 'patient_age': 4, 'patient_gender': 'male', 'department_name': '儿科', 'estimated_minutes': 30, 'osce_enabled': False, 'tags': '儿科,发热', 'traditional': { 'standard_diagnosis': 'Mock 上呼吸道感染', 'standard_treatment': 'Mock 对症治疗', 'guideline_reference': 'Mock 指南', }, 'exam_items': [ { 'item_code': 'blood_routine', 'item_name': '血常规', 'item_type': 'lab', 'category': '实验室检查', 'result_text': 'WBC 10×10^9/L', 'is_key': True, 'is_abnormal': False, 'score_weight': 1.0, 'display_order': 1, }, ], }, 'usage': {'prompt_tokens': 100, 'completion_tokens': 200, 'total_tokens': 300}, } MOCK_C2_SCORING_RULES = { 'data': { 'scoring_rules': [ { 'dimension': '诊断准确性', 'competency_dimension': '临床推理', 'score_weight': 0.4, 'ai_auto_score': True, 'osce_dimension': False, 'scoring_standard': '能准确判断上呼吸道感染', }, { 'dimension': '治疗方案合理性', 'competency_dimension': '治疗决策', 'score_weight': 0.3, 'ai_auto_score': True, 'osce_dimension': False, 'scoring_standard': '治疗方案符合指南推荐', }, { 'dimension': '医患沟通', 'competency_dimension': '沟通技巧', 'score_weight': 0.3, 'ai_auto_score': False, 'osce_dimension': True, 'scoring_standard': '能向家属解释病情和治疗方案', }, ], }, 'usage': {'prompt_tokens': 150, 'completion_tokens': 250, 'total_tokens': 400}, } def make_fake_pdf(): """创建一个假 PDF 上传文件(仅用于触发 multipart 解析)。""" return SimpleUploadedFile( 'test.pdf', b'%PDF-1.4 fake content for testing', content_type='application/pdf', )