"""移动端个人中心 - 训练统计/智能分析三接口测试。 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} # 两条记录的维度评分（标准 5 维；检查利用→处置决策 用于验证归并）。得分率见行尾注释。 DIMS_98 = [_dim('信息获取', 20, 25), _dim('分析推理', 18, 20), _dim('检查利用', 9, 10), _dim('沟通人文', 7, 10), _dim('临床整合', 8, 10)] # 信息80/分析90/处置90/沟通70/整合80 DIMS_80 = [_dim('信息获取', 20, 25), _dim('分析推理', 16, 20), _dim('处置决策', 8, 10), _dim('沟通人文', 8, 10), _dim('临床整合', 8, 10)] # 信息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) # 检查利用90 + 处置决策80 → avg self.assertEqual(radar['临床整合'], 80) # avg(80,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)