"""CMS 各角色概览大屏聚合接口测试（第八章 / CMS-STATS-*）。 `training_record` / `training_session` 是 managed=False 只读表，迁移无占位结构， 故在 setUpClass 按真实列建表（仅测试库），用 TransactionTestCase 允许 DDL。 权限类用例（401/403）在权限层即返回、不触表，放普通 TestCase。 """ 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, TeacherStudentRelation from apps.case.models import CaseBase from .conftest import ( CacheTestCase, create_test_user, get_auth_client, ensure_institution, ) OVERVIEW = '/api/cms/stats/overview/' HOSPITAL = '/api/cms/stats/hospital/overview/' CONTENT = '/api/cms/stats/content/overview/' TEACHING = '/api/cms/stats/teaching/overview/' 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 """ CREATE_TS = """ CREATE TABLE training_session ( id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, case_id BIGINT NULL, case_type VARCHAR(30) NULL, training_mode VARCHAR(50) NULL, status VARCHAR(30) NULL, external_user_id VARCHAR(128) NULL, started_at DATETIME NULL, completed_at DATETIME NULL, created_at DATETIME NULL, updated_at DATETIME NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ DIMS = [{'dimension': '信息获取', 'score': 8, 'max_score': 10}, {'dimension': '分析推理', 'score': 9, 'max_score': 10}, {'dimension': '处置决策', 'score': 8, 'max_score': 10}, {'dimension': '沟通人文', 'score': 7, 'max_score': 10}, {'dimension': '临床整合', 'score': 8, 'max_score': 10}] class StatsOverviewTest(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('DROP TABLE IF EXISTS training_session') c.execute(CREATE_TR) c.execute(CREATE_TS) c.execute('SET FOREIGN_KEY_CHECKS=1') def _record(self, user_id, case_id, total_score, status='completed'): now = 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, created_at, updated_at) " "VALUES (%s,%s,'practice','traditional',%s,%s,1800,%s,%s,'percentage','good',%s,%s,%s)", [user_id, case_id, status, total_score, now, now, json.dumps({'dimension_scores': DIMS}, ensure_ascii=False), now, now], ) def _session(self, case_id, completed=True): now = timezone.now() with connection.cursor() as c: c.execute( "INSERT INTO training_session (case_id, case_type, status, created_at, updated_at, " "started_at, completed_at) VALUES (%s,'traditional',%s,%s,%s,%s,%s)", [case_id, 'completed' if completed else 'inquiry', now, now, now, now if completed else None], ) def setUp(self): cache.clear() with connection.cursor() as c: c.execute('DELETE FROM training_record') c.execute('DELETE FROM training_session') self.instA = ensure_institution(name='A院', code='STAT-A') self.instB = ensure_institution(name='B院', code='STAT-B') self.dept = Department.objects.create(name='心内科', category='临床') self.caseA1 = CaseBase.objects.create(title='病例甲', case_type='traditional', institution=self.instA, department=self.dept, publish_status=1) self.caseA2 = CaseBase.objects.create(title='病例乙', case_type='traditional', institution=self.instA, department=self.dept, publish_status=0) self.superu = create_test_user(phone='13900000001', role_type='super_admin', institution=self.instA) self.hosp = create_test_user(phone='13900000002', role_type='hospital_admin', institution=self.instA) self.content = create_test_user(phone='13900000003', role_type='content_admin', institution=self.instA) self.doctor = create_test_user(phone='13900000004', role_type='doctor', institution=self.instA) self.s1 = create_test_user(phone='13900000005', role_type='student', institution=self.instA) self.s2 = create_test_user(phone='13900000006', role_type='student', institution=self.instA) TeacherStudentRelation.objects.create(teacher=self.doctor, student=self.s1, status=1) TeacherStudentRelation.objects.create(teacher=self.doctor, student=self.s2, status=1) # 4 个会话（3 完成）、2 条训练记录（s1=90、s2=80，均在 A 院病例甲） self._session(self.caseA1.id, completed=True) self._session(self.caseA1.id, completed=True) self._session(self.caseA1.id, completed=True) self._session(self.caseA1.id, completed=False) self._record(self.s1.id, self.caseA1.id, 90) self._record(self.s2.id, self.caseA1.id, 80) # ── 权限 ───────────────────────────────────────────────────────────── def test_permissions(self): anon = APIClient() for url in (OVERVIEW, HOSPITAL, CONTENT, TEACHING): self.assertEqual(anon.get(url).status_code, 401, url) # 角色错配 → 403 self.assertEqual(get_auth_client(self.hosp).get(OVERVIEW).status_code, 403) # 非超管 self.assertEqual(get_auth_client(self.superu).get(HOSPITAL).status_code, 403) # 非医院管理员 self.assertEqual(get_auth_client(self.doctor).get(CONTENT).status_code, 403) # 非内容管理员 self.assertEqual(get_auth_client(self.content).get(TEACHING).status_code, 403) # 非带教医生 # ── 8.1 平台总览 ───────────────────────────────────────────────────── def test_platform_overview(self): d = get_auth_client(self.superu).get(OVERVIEW).json() self.assertEqual(d['kpi']['institution_count'], 2) self.assertEqual(d['kpi']['user_total'], 6) # 6 个 status=1 用户 self.assertEqual(d['core']['train_total'], 4) # 4 个会话 self.assertEqual(d['core']['complete_rate'], 75.0) # 3/4 self.assertEqual(d['core']['avg_score'], 85.0) # avg(90,80) self.assertEqual(d['case_asset']['total'], 2) self.assertEqual(set(x['dimension'] for x in d['competency']['radar']) if 'competency' in d else {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}, {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}) # ── 8.2 医院驾驶舱 ─────────────────────────────────────────────────── def test_hospital_overview(self): d = get_auth_client(self.hosp).get(HOSPITAL).json() self.assertEqual(d['profile']['name'], 'A院') self.assertEqual(d['summary']['student_count'], 2) self.assertEqual(d['summary']['doctor_count'], 1) self.assertEqual(d['summary']['train_total'], 2) # s1+s2 记录 self.assertEqual(d['summary']['avg_score'], 85.0) self.assertEqual(d['case_asset']['total'], 2) self.assertEqual(set(x['dimension'] for x in d['competency']['radar']), {'信息获取', '分析推理', '处置决策', '沟通人文', '临床整合'}) # ── 8.3 内容概览 ───────────────────────────────────────────────────── def test_content_overview(self): d = get_auth_client(self.content).get(CONTENT).json() self.assertEqual(d['summary']['case_total'], 2) self.assertEqual(d['summary']['pending_publish'], 1) # caseA1 publish_status=1 self.assertEqual(d['summary']['train_total'], 2) # ── 8.4 教学概览 ───────────────────────────────────────────────────── def test_teaching_overview(self): d = get_auth_client(self.doctor).get(TEACHING).json() self.assertEqual(d['overview']['student_count'], 2) self.assertEqual(len(d['students']), 2) byid = {s['id']: s for s in d['students']} self.assertEqual(byid[self.s1.id]['avg_score'], 90.0) self.assertEqual(byid[self.s2.id]['avg_score'], 80.0) self.assertEqual(byid[self.s1.id]['most_trained_type'], 'traditional') # 任务相关无数据源 → None self.assertIsNone(byid[self.s1.id]['pending_tasks']) self.assertIsNone(d['overview']['task_summary'])