179 lines
9.9 KiB
Python
179 lines
9.9 KiB
Python
"""移动端个人中心 - 训练统计/智能分析三接口测试。
|
||
|
||
training_record 是 managed=False 只读表，迁移里的占位结构缺新列，故在 setUpClass 里
|
||
按真实列重建该表（仅测试库），用 TransactionTestCase 允许 DDL。
|
||
"""
|
||
import json
|
||
|
||
from django.core.cache import cache
|
||
from django.db import connection
|
||
from django.test import TransactionTestCase
|
||
from django.utils import timezone
|
||
from rest_framework.test import APIClient
|
||
|
||
from apps.user.models import Department, Institution, User
|
||
from apps.case.models import CaseBase
|
||
from .conftest import create_test_user, get_auth_client, ensure_institution
|
||
|
||
CREATE_TR = """
|
||
CREATE TABLE training_record (
|
||
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
|
||
user_id BIGINT NULL, case_id BIGINT NULL, teacher_id BIGINT NULL,
|
||
training_mode VARCHAR(50) NULL, case_type VARCHAR(30) NULL,
|
||
start_time DATETIME NULL, end_time DATETIME NULL, duration_seconds INT NULL,
|
||
total_score DECIMAL(5,2) NULL, ai_score DECIMAL(5,2) NULL, teacher_score DECIMAL(5,2) NULL,
|
||
evaluation_level VARCHAR(20) NULL, status VARCHAR(30) NULL,
|
||
feedback TEXT NULL, thinking_chain TEXT NULL, diagnosis_path TEXT NULL,
|
||
wrong_points JSON NULL, missed_questions JSON NULL, recommendation_result JSON NULL,
|
||
ai_feedback_structured JSON NULL, osce_station_score JSON NULL,
|
||
interruption_count INT NULL, emotion_analysis JSON NULL,
|
||
prompt_version VARCHAR(50) NULL, rag_context_version VARCHAR(50) NULL,
|
||
external_user_id VARCHAR(128) NULL, session_id BIGINT NULL, evaluation_record_id BIGINT NULL,
|
||
score_type VARCHAR(20) NULL, pdf_file_path VARCHAR(512) NULL,
|
||
created_at DATETIME NULL, updated_at DATETIME NULL
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
"""
|
||
|
||
COMP_URL = '/api/user/competency-metrics/'
|
||
LIST_URL = '/api/user/training-records/'
|
||
ANALYSIS_URL = '/api/user/analysis/'
|
||
|
||
|
||
def _dim(name, score, mx):
|
||
return {'dimension': name, 'score': score, 'max_score': mx}
|
||
|
||
|
||
# 两条记录的维度评分（得分率：见注释）
|
||
DIMS_98 = [_dim('信息获取', 20, 25), _dim('体格检查', 8, 10), _dim('检查决策', 9, 10),
|
||
_dim('诊断推理', 18, 20), _dim('治疗决策', 8, 10), _dim('医患沟通', 7, 10)] # 80/80/90/90/80/70
|
||
DIMS_80 = [_dim('信息获取', 20, 25), _dim('体格检查', 8, 10), _dim('检查决策', 8, 10),
|
||
_dim('诊断推理', 16, 20), _dim('治疗决策', 8, 10), _dim('医患沟通', 8, 10)] # 80/80/80/80/80/80
|
||
|
||
|
||
class TrainingStatsTest(TransactionTestCase):
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
super().setUpClass()
|
||
with connection.cursor() as c:
|
||
c.execute('SET FOREIGN_KEY_CHECKS=0')
|
||
c.execute('DROP TABLE IF EXISTS training_record')
|
||
c.execute(CREATE_TR)
|
||
c.execute('SET FOREIGN_KEY_CHECKS=1')
|
||
|
||
def _insert(self, user_id, case_id, total_score, duration, dims, status='completed', end_time=None,
|
||
score_type='percentage'):
|
||
end_time = end_time or timezone.now()
|
||
with connection.cursor() as c:
|
||
c.execute(
|
||
"INSERT INTO training_record (user_id, case_id, training_mode, case_type, status, "
|
||
"total_score, duration_seconds, end_time, start_time, score_type, evaluation_level, "
|
||
"ai_feedback_structured, pdf_file_path, created_at, updated_at) "
|
||
"VALUES (%s,%s,'practice','diagnosis_treatment',%s,%s,%s,%s,%s,%s,'good',%s,%s,%s,%s)",
|
||
[user_id, case_id, status, total_score, duration, end_time, end_time, score_type,
|
||
json.dumps({'dimension_scores': dims}, ensure_ascii=False),
|
||
'/app/storage/reports/r.pdf', end_time, end_time],
|
||
)
|
||
return c.lastrowid
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
with connection.cursor() as c:
|
||
c.execute('DELETE FROM training_record')
|
||
self.inst = ensure_institution(name='测试医院', code='TS-H1')
|
||
self.dept = Department.objects.create(name='心内科', category='临床')
|
||
self.case = CaseBase.objects.create(title='急性心肌梗死', case_type='traditional', department=self.dept)
|
||
self.user = create_test_user(phone='13955500001', role_type='student', institution=self.inst)
|
||
self.other = create_test_user(phone='13955500002', role_type='student', institution=self.inst)
|
||
self.client = get_auth_client(self.user)
|
||
# 本人 2 条已完成 + 1 条进行中；他人 1 条
|
||
now = timezone.now()
|
||
self.rec98 = self._insert(self.user.id, self.case.id, 98, 3600, DIMS_98, end_time=now)
|
||
self.rec80 = self._insert(self.user.id, self.case.id, 80, 1800, DIMS_80, end_time=now - timezone.timedelta(days=1))
|
||
self._insert(self.user.id, self.case.id, 50, 600, DIMS_80, status='in_progress')
|
||
self._insert(self.other.id, self.case.id, 10, 600, DIMS_80)
|
||
|
||
# ── 4.3 临床核心能力指标 ──────────────────────────────────────────────────
|
||
def test_competency_metrics(self):
|
||
resp = self.client.get(COMP_URL)
|
||
self.assertEqual(resp.status_code, 200, resp.content)
|
||
d = resp.json()
|
||
self.assertEqual(d['completed_cases'], 2) # 仅本人已完成，排除进行中/他人
|
||
self.assertEqual(d['total_hours'], 1.5) # (3600+1800)/3600
|
||
self.assertEqual(d['avg_score'], 89.0) # avg(98,80)
|
||
self.assertEqual(d['diagnosis_accuracy'], 85) # 诊断推理 avg(90%,80%)
|
||
|
||
def test_competency_requires_auth(self):
|
||
self.assertEqual(APIClient().get(COMP_URL).status_code, 401)
|
||
|
||
def test_score_type_normalized(self):
|
||
# 五分制(0-5)记录应 ×20 归一到百分制后再与百分制平均，避免量纲混算
|
||
with connection.cursor() as c:
|
||
c.execute('DELETE FROM training_record')
|
||
u = create_test_user(phone='13955500007', role_type='student', institution=self.inst)
|
||
self._insert(u.id, self.case.id, 90, 3600, DIMS_80, score_type='percentage') # 90 分
|
||
self._insert(u.id, self.case.id, 4, 3600, DIMS_80, score_type='five_point') # 4×20=80 分
|
||
client = get_auth_client(u)
|
||
comp = client.get(COMP_URL).json()
|
||
self.assertEqual(comp['avg_score'], 85.0) # avg(90, 80) 而非 avg(90, 4)=47
|
||
ana = client.get(ANALYSIS_URL).json()
|
||
self.assertEqual(ana['current_score'], 85.0)
|
||
self.assertTrue(all(0 <= x['score'] <= 100 for x in ana['recent_trend']))
|
||
|
||
# ── 4.4 训练记录（统计信息）──────────────────────────────────────────────
|
||
def test_training_records(self):
|
||
resp = self.client.get(LIST_URL)
|
||
self.assertEqual(resp.status_code, 200, resp.content)
|
||
d = resp.json()
|
||
self.assertEqual(d['count'], 2)
|
||
self.assertEqual(d['summary'], {'total_cases': 2, 'total_hours': 1.5, 'avg_accuracy': 85})
|
||
first = d['results'][0] # 按 end_time 倒序 → 98 分那条
|
||
self.assertEqual(first['score'], 98.0)
|
||
self.assertEqual(first['case_title'], '急性心肌梗死')
|
||
self.assertEqual(first['department'], '心内科')
|
||
self.assertEqual(first['score_type'], 'percentage')
|
||
self.assertNotIn('pdf_file_path', first) # fastapi 内部路径，Django 不返回
|
||
|
||
def test_training_records_search(self):
|
||
resp = self.client.get(LIST_URL, {'search': '心肌'})
|
||
self.assertEqual(resp.json()['count'], 2)
|
||
resp = self.client.get(LIST_URL, {'search': '不存在的病例'})
|
||
self.assertEqual(resp.json()['count'], 0)
|
||
|
||
# ── 4.5 智能分析（关联评价表）─────────────────────────────────────────────
|
||
def test_analysis(self):
|
||
resp = self.client.get(ANALYSIS_URL)
|
||
self.assertEqual(resp.status_code, 200, resp.content)
|
||
d = resp.json()
|
||
self.assertEqual(d['current_score'], 89.0)
|
||
radar = {x['dimension']: x['score'] for x in d['radar']}
|
||
self.assertEqual(set(radar), {'病史采集', '查体能力', '检查决策', '诊断能力', '治疗决策', '医患沟通'})
|
||
self.assertEqual(radar['病史采集'], 80) # avg(80,80)
|
||
self.assertEqual(radar['检查决策'], 85) # avg(90,80)
|
||
self.assertEqual(radar['诊断能力'], 85) # 诊断推理 avg(90,80)
|
||
self.assertEqual(radar['医患沟通'], 75) # avg(70,80) → 最低
|
||
self.assertEqual(d['weak_dimensions'], ['医患沟通'])
|
||
self.assertIn('医患沟通', d['comment']) # 强/弱不同 → 走对比文案
|
||
self.assertIn('突出', d['comment'])
|
||
|
||
def test_analysis_balanced_single_record(self):
|
||
# 单条记录、各维度并列：强==弱，文案应走「均衡」分支而非自相矛盾
|
||
with connection.cursor() as c:
|
||
c.execute('DELETE FROM training_record')
|
||
u = create_test_user(phone='13955500008', role_type='student', institution=self.inst)
|
||
flat = [_dim('信息获取', 8, 10), _dim('诊断推理', 8, 10), _dim('治疗决策', 8, 10)] # 全 80%
|
||
self._insert(u.id, self.case.id, 80, 1800, flat)
|
||
d = get_auth_client(u).get(ANALYSIS_URL).json()
|
||
# 最强与最弱同分，不应出现「您的X表现突出，但X仍有提升」式自相矛盾
|
||
self.assertNotIn('表现突出', d['comment'])
|
||
self.assertIn('均衡', d['comment'])
|
||
|
||
def test_analysis_empty_user(self):
|
||
# 无训练记录的用户 → 全 0、不报错
|
||
client = get_auth_client(self.other) if False else None
|
||
u = create_test_user(phone='13955500009', role_type='student', institution=self.inst)
|
||
resp = get_auth_client(u).get(ANALYSIS_URL)
|
||
self.assertEqual(resp.status_code, 200, resp.content)
|
||
self.assertEqual(resp.json()['current_score'], 0)
|
||
self.assertEqual(resp.json()['radar'][0]['score'], 0)
|