299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""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',
|
||
)
|