shihan11/medical_training
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: cms users institution department manager

Browse Source
This commit is contained in:
shihan11
2026-06-11 10:37:29 +08:00
parent 1dc9141856
commit 32915bc6b4
39 changed files with 2403 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/conftest.py
+3 -3
View File
@@ -123,11 +123,11 @@ def ensure_institution(name='测试医院', code='TEST-HOSP-001'):
return inst
def ensure_department(name='儿科', institution_name='测试医院'):
inst = ensure_institution(institution_name)
def ensure_department(name='儿科'):
"""科室为全局表，与机构无关。"""
dept, _ = Department.objects.get_or_create(
name=name,
defaults={'institution': inst, 'category': '临床'},
defaults={'category': '临床'},
)
return dept
test/swagger_cms.py
+245
View File
@@ -0,0 +1,245 @@
# -*- coding: utf-8 -*-
"""
CMS 端全接口 Swagger Try-it-out 脚本（真实 HTTP）。
- 覆盖：超级管理员(用户/医院/科室)、医院管理员(人员/师生关系)、带教医生(我的学生)。
- 详细请求/响应写入 logs/test-swagger-cms-YYYY-MM-DD.log。
- 导入/导出的 .xlsx 真实文件保存到 docx/CMS-excel样例/。
- 真实请求/响应样例写入 logs/cms-swagger-examples.json（供回填 CSV）。
前提：Django dev server 运行在 http://127.0.0.1:8000，Redis 已启动。
运行：.venv\\Scripts\\python.exe test/swagger_cms.py
"""
import io
import os
import sys
import json
import subprocess
from datetime import datetime
import requests
from openpyxl import Workbook
sys.stdout.reconfigure(encoding='utf-8')
BASE = 'http://127.0.0.1:8000'
PYTHON = r'D:\01Agent\medical_training\.venv\Scripts\python.exe'
CWD = r'D:\01Agent\medical_training'
LOG_DIR = os.path.join(CWD, 'logs')
EXCEL_DIR = os.path.join(CWD, 'docx', 'CMS-excel样例')
os.makedirs(EXCEL_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOG_DIR, f'test-swagger-cms-{datetime.now():%Y-%m-%d}.log')
EXAMPLES_FILE = os.path.join(LOG_DIR, 'cms-swagger-examples.json')
_fh = open(LOG_FILE, 'w', encoding='utf-8')
examples = {}
results = []
def w(text=''):
_fh.write(text + '\n'); _fh.flush(); print(text)
def django_eval(code):
pre = ('import django,os;os.environ.setdefault("DJANGO_SETTINGS_MODULE","config.settings");django.setup();')
p = subprocess.run([PYTHON, '-c', pre + code], capture_output=True, text=True, cwd=CWD, encoding='utf-8')
return (p.stdout or '').strip()
def make_xlsx_bytes(headers, rows):
wb = Workbook(); ws = wb.active
ws.append(headers)
for r in rows:
ws.append(r)
buf = io.BytesIO(); wb.save(buf); return buf.getvalue()
def call(code, name, method, path, token=None, json_body=None, params=None,
file_bytes=None, file_name=None, save_as=None, expect=200):
"""发起请求 + 记录日志 + 收集样例。"""
headers = {'Authorization': f'Bearer {token}'} if token else {}
files = {'file': (file_name, file_bytes, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')} if file_bytes else None
r = requests.request(method, BASE + path, headers=headers, json=json_body, params=params, files=files)
ct = r.headers.get('content-type', '')
is_json = ct.startswith('application/json')
resp = r.json() if is_json else f'<xlsx 文件 {len(r.content)} 字节>'
if save_as:
full = os.path.join(EXCEL_DIR, save_as)
with open(full, 'wb') as f:
f.write(r.content)
# 请求样例（给前端看：实际传了什么）
if json_body is not None:
req_example = json.dumps(json_body, ensure_ascii=False)
elif file_bytes is not None:
req_example = f'multipart/form-data: file={file_name}'
elif params:
req_example = '?' + '&'.join(f'{k}={v}' for k, v in params.items())
else:
req_example = '(无 body，仅 Header Authorization)' if '{' not in path else '(路径参数 + Header Authorization)'
ok = 'PASS' if r.status_code in (expect if isinstance(expect, (list, tuple)) else [expect]) else 'FAIL'
results.append((code, ok, r.status_code))
examples[code] = {'name': name, 'method': method, 'path': path, 'status': r.status_code,
'req': req_example, 'resp': resp, 'saved': save_as}
w(f'\n[{ok}] {code} {name} -> {method} {path} (status={r.status_code})')
w(f' 请求: {req_example}')
body_str = json.dumps(resp, ensure_ascii=False, indent=2) if is_json else resp
if len(body_str) > 1500:
body_str = body_str[:1500] + f' ...(截断, 共{len(body_str)}字符)'
w(f' 响应: {body_str}')
return r
# ─────────────────────────────────────────────────────────────────────────────
w('=' * 90)
w(f' CMS 全接口 Swagger 测试 {datetime.now():%Y-%m-%d %H:%M:%S}')
w('=' * 90)
# 1) 准备数据：两机构、两科室、超管/医院管理员/医生/学生/他院学生，并签发 token
w('\n[准备] 建测试机构/科室/用户、签发 token ...')
setup = django_eval(r'''
from apps.user.models import User, Institution, Department, TeacherStudentRelation
from rest_framework_simplejwt.tokens import RefreshToken
# 物理清理可能的残留（含软删），保证脚本可重复运行
ph = ["13800009000","13800009001","13800009002","13800009003","13800009004","13800009010","13800009011","13800009012","13800009020","13800009021","13800009022"]
for u in User.all_objects.filter(phone__in=ph): u.hard_delete()
Institution.all_objects.filter(code__startswith="SWG-CMS").hard_delete()
for d in Department.all_objects.filter(name__startswith="Swagger"): d.hard_delete()
instA,_ = Institution.objects.get_or_create(code="SWG-CMS-A", defaults={"name":"Swagger甲医院","type":"hospital","level":"三甲","province":"北京","city":"北京"})
instB,_ = Institution.objects.get_or_create(code="SWG-CMS-B", defaults={"name":"Swagger乙医院","type":"hospital"})
dN,_ = Department.objects.get_or_create(name="内科", defaults={"category":"临床"})
dW,_ = Department.objects.get_or_create(name="外科", defaults={"category":"临床"})
sup = User.objects.create_user(username="13800009000", password="x", phone="13800009000", real_name="Swagger超管", role_type="super_admin", status=1)
hos = User.objects.create_user(username="13800009001", password="x", phone="13800009001", real_name="Swagger院管", role_type="hospital_admin", institution=instA, status=1)
doc = User.objects.create_user(username="13800009002", password="x", phone="13800009002", real_name="Swagger医生", role_type="doctor", institution=instA, status=1)
stu = User.objects.create_user(username="13800009003", password="x", phone="13800009003", real_name="Swagger学生", role_type="student", institution=instA, department=dN, status=1)
stuB = User.objects.create_user(username="13800009004", password="x", phone="13800009004", real_name="他院学生", role_type="student", institution=instB, status=1)
TeacherStudentRelation.objects.create(teacher=doc, student=stu, relation_type="指导", status=1)
def tok(u): return str(RefreshToken.for_user(u).access_token)
print("|".join([tok(sup), tok(hos), tok(doc), str(instA.id), str(instB.id), str(dN.id), str(stu.id), str(doc.id)]))
''')
T_SUP, T_HOS, T_DOC, INST_A, INST_B, DEPT_N, STU_ID, DOC_ID = setup.split('|')
w(f'[准备] 完成 instA={INST_A} instB={INST_B} deptN={DEPT_N} student={STU_ID} doctor={DOC_ID}')
# ═══ 超级管理员 - 用户管理 ═══
w('\n' + '#' * 70 + '\n# 超级管理员 - 用户管理\n' + '#' * 70)
call('CMS-USER-1', '用户列表', 'GET', '/api/cms/users/', T_SUP, params={'role_type': 'doctor', 'page': 1})
r = call('CMS-USER-2', '新增用户', 'POST', '/api/cms/users/', T_SUP,
json_body={'phone': '13800009010', 'real_name': 'Swagger新医生', 'role_type': 'doctor', 'institution': int(INST_A)}, expect=201)
NEW_UID = r.json().get('id')
call('CMS-USER-3', '编辑用户', 'PATCH', f'/api/cms/users/{NEW_UID}/', T_SUP, json_body={'real_name': 'Swagger改名', 'title_name': '主治医师'})
call('CMS-USER-5', '重置密码', 'POST', f'/api/cms/users/{NEW_UID}/reset-password/', T_SUP, json_body={})
ufile = make_xlsx_bytes(['手机号', '姓名', '角色', '机构编码'],
[['13800009011', '导入医生', '医生', 'SWG-CMS-A'], ['13800009012', '导入学生', '学生', 'SWG-CMS-A']])
with open(os.path.join(EXCEL_DIR, '用户导入_样例.xlsx'), 'wb') as f:
f.write(ufile)
call('CMS-USER-6', '导入用户', 'POST', '/api/cms/users/import/', T_SUP, file_bytes=ufile, file_name='用户导入_样例.xlsx')
call('CMS-USER-7', '导入模板', 'GET', '/api/cms/users/import-template/', T_SUP, save_as='用户导入_模板.xlsx')
call('CMS-USER-8', '导出用户', 'GET', '/api/cms/users/export/', T_SUP, params={'role_type': 'doctor'}, save_as='用户导出_结果.xlsx')
call('CMS-USER-4', '停用用户', 'DELETE', f'/api/cms/users/{NEW_UID}/', T_SUP, expect=204)
# ═══ 超级管理员 - 医院管理 ═══
w('\n' + '#' * 70 + '\n# 超级管理员 - 医院管理\n' + '#' * 70)
call('CMS-INST-1', '机构列表', 'GET', '/api/cms/institutions/', T_SUP, params={'search': 'Swagger'})
r = call('CMS-INST-2', '新增机构', 'POST', '/api/cms/institutions/', T_SUP,
json_body={'code': 'SWG-CMS-NEW', 'name': 'Swagger新医院', 'type': 'hospital', 'level': '二甲', 'province': '上海', 'city': '上海'}, expect=201)
NEW_INST = r.json().get('id')
call('CMS-INST-3', '编辑机构', 'PATCH', f'/api/cms/institutions/{NEW_INST}/', T_SUP, json_body={'level': '三甲'})
ifile = make_xlsx_bytes(['机构编码', '名称', '类型', '等级', '省', '市'],
[['SWG-CMS-IMP1', '导入医院A', 'hospital', '三甲', '广东', '广州']])
with open(os.path.join(EXCEL_DIR, '机构导入_样例.xlsx'), 'wb') as f:
f.write(ifile)
call('CMS-INST-5', '导入机构', 'POST', '/api/cms/institutions/import/', T_SUP, file_bytes=ifile, file_name='机构导入_样例.xlsx')
call('CMS-INST-6', '导入模板', 'GET', '/api/cms/institutions/import-template/', T_SUP, save_as='机构导入_模板.xlsx')
call('CMS-INST-7', '导出机构', 'GET', '/api/cms/institutions/export/', T_SUP, save_as='机构导出_结果.xlsx')
call('CMS-INST-4', '停用机构', 'DELETE', f'/api/cms/institutions/{NEW_INST}/', T_SUP, expect=204)
# ═══ 超级管理员 - 科室管理 ═══
w('\n' + '#' * 70 + '\n# 超级管理员 - 科室管理\n' + '#' * 70)
call('CMS-DEPT-1', '科室列表', 'GET', '/api/cms/departments/', T_SUP)
r = call('CMS-DEPT-2', '新增科室', 'POST', '/api/cms/departments/', T_SUP, json_body={'name': 'Swagger儿科', 'category': '临床'}, expect=201)
NEW_DEPT = r.json().get('id')
call('CMS-DEPT-3', '编辑科室', 'PATCH', f'/api/cms/departments/{NEW_DEPT}/', T_SUP, json_body={'category': '儿童医学'})
dfile = make_xlsx_bytes(['科室名称', '分类'], [['Swagger放射科', '医技']])
with open(os.path.join(EXCEL_DIR, '科室导入_样例.xlsx'), 'wb') as f:
f.write(dfile)
call('CMS-DEPT-5', '导入科室', 'POST', '/api/cms/departments/import/', T_SUP, file_bytes=dfile, file_name='科室导入_样例.xlsx')
call('CMS-DEPT-7', '导入模板', 'GET', '/api/cms/departments/import-template/', T_SUP, save_as='科室导入_模板.xlsx')
call('CMS-DEPT-6', '导出科室', 'GET', '/api/cms/departments/export/', T_SUP, save_as='科室导出_结果.xlsx')
call('CMS-DEPT-4', '停用科室', 'DELETE', f'/api/cms/departments/{NEW_DEPT}/', T_SUP, expect=204)
# ═══ 医院管理员 - 人员管理（本院 doctor/student/content_admin）═══
w('\n' + '#' * 70 + '\n# 医院管理员 - 人员管理\n' + '#' * 70)
call('CMS-HUSER-1', '用户列表(本院)', 'GET', '/api/cms/users/', T_HOS, params={'role_type': 'student'})
r = call('CMS-HUSER-2', '新增用户(本院)', 'POST', '/api/cms/users/', T_HOS,
json_body={'phone': '13800009020', 'real_name': '院管建的内容员', 'role_type': 'content_admin'}, expect=201)
HU_UID = r.json().get('id')
call('CMS-HUSER-3', '编辑用户(本院)', 'PATCH', f'/api/cms/users/{HU_UID}/', T_HOS, json_body={'title_name': '内容主管'})
call('CMS-HUSER-5', '重置密码(本院)', 'POST', f'/api/cms/users/{HU_UID}/reset-password/', T_HOS, json_body={})
hufile = make_xlsx_bytes(['手机号', '姓名', '角色', '机构编码'], [['13800009021', '院管导入生', '学生', '忽略']])
with open(os.path.join(EXCEL_DIR, '本院用户导入_样例.xlsx'), 'wb') as f:
f.write(hufile)
call('CMS-HUSER-6', '导入用户(本院)', 'POST', '/api/cms/users/import/', T_HOS, file_bytes=hufile, file_name='本院用户导入_样例.xlsx')
call('CMS-HUSER-7', '导入模板', 'GET', '/api/cms/users/import-template/', T_HOS, save_as='本院用户导入_模板.xlsx')
call('CMS-HUSER-8', '导出用户(本院)', 'GET', '/api/cms/users/export/', T_HOS, save_as='本院用户导出_结果.xlsx')
call('CMS-HUSER-4', '停用用户(本院)', 'DELETE', f'/api/cms/users/{HU_UID}/', T_HOS, expect=204)
# ═══ 医院管理员 - 师生关系 ═══
w('\n' + '#' * 70 + '\n# 医院管理员 - 师生关系\n' + '#' * 70)
call('CMS-REL-1', '师生关系列表', 'GET', '/api/cms/teacher-student-relations/', T_HOS)
# 先把已存在的 doc×stu 关系拿来演示编辑/停用；新增用一对新的（再建一个本院学生）
new_stu = django_eval(r'''
from apps.user.models import User, Institution
instA = Institution.objects.get(code="SWG-CMS-A")
u,_ = User.objects.get_or_create(phone="13800009022", defaults=dict(username="13800009022", real_name="师生关系新学生", role_type="student", institution=instA, status=1))
print(u.id)
''')
r = call('CMS-REL-2', '新增师生关系', 'POST', '/api/cms/teacher-student-relations/', T_HOS,
json_body={'teacher': int(DOC_ID), 'student': int(new_stu), 'relation_type': '指导'}, expect=201)
REL_ID = r.json().get('id')
call('CMS-REL-3', '编辑师生关系', 'PATCH', f'/api/cms/teacher-student-relations/{REL_ID}/', T_HOS, json_body={'status': 0})
rfile = make_xlsx_bytes(['带教医生手机号', '学生手机号'], [['13800009002', '13800009003']])
with open(os.path.join(EXCEL_DIR, '师生关系导入_样例.xlsx'), 'wb') as f:
f.write(rfile)
call('CMS-REL-5', '导入师生关系', 'POST', '/api/cms/teacher-student-relations/import/', T_HOS, file_bytes=rfile, file_name='师生关系导入_样例.xlsx')
call('CMS-REL-6', '导入模板', 'GET', '/api/cms/teacher-student-relations/import-template/', T_HOS, save_as='师生关系导入_模板.xlsx')
call('CMS-REL-7', '导出师生关系', 'GET', '/api/cms/teacher-student-relations/export/', T_HOS, save_as='师生关系导出_结果.xlsx')
call('CMS-REL-4', '停用师生关系', 'DELETE', f'/api/cms/teacher-student-relations/{REL_ID}/', T_HOS, expect=204)
# ═══ 带教医生 - 我的学生 ═══
w('\n' + '#' * 70 + '\n# 带教医生 - 我的学生\n' + '#' * 70)
call('CMS-TEA-1', '我的学生列表', 'GET', '/api/cms/students/', T_DOC)
call('CMS-TEA-2', '学生基础信息', 'GET', f'/api/cms/students/{STU_ID}/', T_DOC)
# ─── 清理 ───
w('\n[清理] 删除测试数据 ...')
django_eval(r'''
from apps.user.models import User, Institution, Department, TeacherStudentRelation
ph = ["13800009000","13800009001","13800009002","13800009003","13800009004","13800009010","13800009011","13800009012","13800009020","13800009021","13800009022"]
TeacherStudentRelation.all_objects.filter(teacher__phone__in=ph).hard_delete()
for u in User.all_objects.filter(phone__in=ph): u.hard_delete()
Institution.all_objects.filter(code__startswith="SWG-CMS").hard_delete()
Department.all_objects.filter(name__startswith="Swagger").hard_delete()
print("cleaned")
''')
# ─── 落盘样例 JSON + 汇总 ───
with open(EXAMPLES_FILE, 'w', encoding='utf-8') as f:
json.dump(examples, f, ensure_ascii=False, indent=2)
w('\n' + '=' * 90)
total = len(results); passed = sum(1 for _, ok, _ in results if ok == 'PASS')
w(f' 总计 {total} 个接口 | 通过 {passed} | 失败 {total - passed}')
fails = [(c, s) for c, ok, s in results if ok == 'FAIL']
if fails:
w(' 失败: ' + ', '.join(f'{c}({s})' for c, s in fails))
w(f' 日志: {LOG_FILE}')
w(f' 样例 JSON: {EXAMPLES_FILE}')
w(f' Excel 文件目录: {EXCEL_DIR}')
_fh.close()
sys.exit(0 if not fails else 1)
test/swagger_profile_config.py
+2 -2
View File
@@ -57,8 +57,8 @@ setup = django_eval(
f'inst, _ = Institution.objects.get_or_create(code="{TRIAL_INST_CODE}", '
f' defaults={{"name":"北大医学部（实验室）试用","type":"hospital"}}); '
f'inst.banner_url = "institutions/default_hospital.png"; inst.save(update_fields=["banner_url"]); '
f'd1, _ = Department.objects.get_or_create(institution=inst, name="内科", defaults={{"category":"临床"}}); '
f'd2, _ = Department.objects.get_or_create(institution=inst, name="外科", defaults={{"category":"临床"}}); '
f'd1, _ = Department.objects.get_or_create(name="内科", defaults={{"category":"临床"}}); '
f'd2, _ = Department.objects.get_or_create(name="外科", defaults={{"category":"临床"}}); '
f'User.objects.filter(phone="{STUDENT_PHONE}").delete(); '
f'u = User.objects.create_user(username="{STUDENT_PHONE}", password=None, phone="{STUDENT_PHONE}", '
f' real_name="配置页测试学生", role_type="student", institution=inst, status=1); '
test/swagger_tryout.py
+1 -1
View File
@@ -528,7 +528,7 @@ django_eval(
f'from apps.user.models import Institution, Department; '
f'inst, _ = Institution.objects.get_or_create(code="{INST_CODE}", '
f' defaults={{"name":"{INST_NAME}","type":"hospital","province":"北京","city":"北京"}}); '
f'Department.objects.get_or_create(name="{DEPT_NAME}", institution=inst, '
f'Department.objects.get_or_create(name="{DEPT_NAME}", '
f' defaults={{"category":"临床"}}); '
f'print("OK")'
)
test/test_cms_department.py
+86
View File
@@ -0,0 +1,86 @@
"""CMS 超级管理员 - 科室管理接口测试（CMS-DEPT-1~6，全局科室）。"""
import io
from openpyxl import Workbook
from django.core.files.uploadedfile import SimpleUploadedFile
from rest_framework.test import APIClient
from apps.user.models import Department
from .conftest import CacheTestCase, create_test_user, get_auth_client
CMS_DEPT_URL = '/api/cms/departments/'
XLSX_CT = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
def d_detail(pk):
return f'/api/cms/departments/{pk}/'
def make_xlsx(headers, rows):
wb = Workbook(); ws = wb.active
ws.append(headers)
for r in rows:
ws.append(r)
buf = io.BytesIO(); wb.save(buf); buf.seek(0)
return SimpleUploadedFile('dept.xlsx', buf.read(), content_type=XLSX_CT)
def super_client(phone='13933300001'):
admin = create_test_user(phone=phone, password='Admin123', role_type='super_admin')
return get_auth_client(admin)
class CmsDepartmentTest(CacheTestCase):
def setUp(self):
super().setUp()
self.client = super_client()
def test_requires_super_admin(self):
self.assertEqual(APIClient().get(CMS_DEPT_URL).status_code, 401)
u = create_test_user(phone='13933300009', role_type='doctor')
self.assertEqual(get_auth_client(u).get(CMS_DEPT_URL).status_code, 403)
def test_crud(self):
# 新增
resp = self.client.post(CMS_DEPT_URL, {'name': '内科', 'category': '临床'})
self.assertEqual(resp.status_code, 201, resp.content)
did = resp.json()['id']
# 列表
resp = self.client.get(CMS_DEPT_URL)
self.assertEqual(resp.status_code, 200)
self.assertIn('results', resp.json())
# 编辑
resp = self.client.patch(d_detail(did), {'category': '医技'})
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['category'], '医技')
def test_duplicate_name(self):
Department.objects.create(name='外科', category='临床')
resp = self.client.post(CMS_DEPT_URL, {'name': '外科'})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_DEPARTMENT_NAME_EXISTS')
def test_soft_delete(self):
d = Department.objects.create(name='儿科', category='临床')
resp = self.client.delete(d_detail(d.id))
self.assertEqual(resp.status_code, 204, resp.content)
self.assertFalse(Department.objects.filter(id=d.id).exists())
self.assertTrue(Department.all_objects.get(id=d.id).is_deleted)
def test_import_and_export(self):
f = make_xlsx(['科室名称', '分类'], [['心内科', '临床'], ['', 'x'], ['心内科', '临床']])
resp = self.client.post('/api/cms/departments/import/', {'file': f}, format='multipart')
self.assertEqual(resp.status_code, 200, resp.content)
body = resp.json()
self.assertEqual(body['success'], 1) # 1 成功；空名 + 重复各 1 失败
self.assertEqual(body['failed'], 2)
# 导出
resp = self.client.get('/api/cms/departments/export/')
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_import_template(self):
resp = self.client.get('/api/cms/departments/import-template/')
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], XLSX_CT)
test/test_cms_huser.py
+88
View File
@@ -0,0 +1,88 @@
"""CMS 医院管理员 - 人员管理（用户范围）测试：CMS-HUSER-1~3。
医院管理员只能管理本院的医生/学生；机构强制本院、角色受限。
"""
from rest_framework.test import APIClient
from apps.user.models import User
from .conftest import CacheTestCase, create_test_user, get_auth_client, ensure_institution
CMS_USER_URL = '/api/cms/users/'
def u_detail(pk):
return f'/api/cms/users/{pk}/'
class HospitalAdminUserScopeTest(CacheTestCase):
def setUp(self):
super().setUp()
self.inst = ensure_institution(name='本院', code='HU-A')
self.other = ensure_institution(name='他院', code='HU-B')
self.admin = create_test_user(phone='13931000001', role_type='hospital_admin',
institution=self.inst)
self.client = get_auth_client(self.admin)
# 本院医生、学生；他院学生
self.doc = create_test_user(phone='13931000002', role_type='doctor', institution=self.inst)
self.stu = create_test_user(phone='13931000003', role_type='student', institution=self.inst)
self.other_stu = create_test_user(phone='13931000004', role_type='student', institution=self.other)
def test_list_scoped_to_own_doctor_student(self):
resp = self.client.get(CMS_USER_URL)
self.assertEqual(resp.status_code, 200, resp.content)
ids = {u['id'] for u in resp.json()['results']}
self.assertIn(self.doc.id, ids)
self.assertIn(self.stu.id, ids)
self.assertNotIn(self.other_stu.id, ids) # 他院不可见
self.assertNotIn(self.admin.id, ids) # 医院管理员自己(role=hospital_admin)不在 doctor/student 范围
def test_create_forces_own_institution(self):
# 传了他院机构，仍落本院
resp = self.client.post(CMS_USER_URL, {
'phone': '13931000010', 'real_name': '新医生', 'role_type': 'doctor',
'institution': self.other.id,
})
self.assertEqual(resp.status_code, 201, resp.content)
u = User.objects.get(phone='13931000010')
self.assertEqual(u.institution_id, self.inst.id) # 强制本院
def test_create_content_admin_allowed(self):
# 医院管理员可给本院授予内容管理员权限
resp = self.client.post(CMS_USER_URL, {
'phone': '13931000011', 'real_name': '内容员', 'role_type': 'content_admin',
})
self.assertEqual(resp.status_code, 201, resp.content)
u = User.objects.get(phone='13931000011')
self.assertEqual(u.role_type, 'content_admin')
self.assertEqual(u.institution_id, self.inst.id) # 强制本院
def test_create_role_restricted(self):
# 医院管理员不能建 hospital_admin / super_admin
for role in ('hospital_admin', 'super_admin'):
resp = self.client.post(CMS_USER_URL, {
'phone': '13931000012', 'real_name': 'x', 'role_type': role,
})
self.assertEqual(resp.status_code, 403, f'{role}: {resp.content}')
self.assertEqual(resp.json()['code'], 'CMS_ROLE_NOT_ALLOWED')
def test_cannot_touch_other_institution_user(self):
# 他院学生不在 queryset → 404
self.assertEqual(self.client.get(u_detail(self.other_stu.id)).status_code, 404)
self.assertEqual(self.client.delete(u_detail(self.other_stu.id)).status_code, 404)
def test_soft_delete_own_student(self):
resp = self.client.delete(u_detail(self.stu.id))
self.assertEqual(resp.status_code, 204, resp.content)
self.assertFalse(User.objects.filter(id=self.stu.id).exists())
def test_reset_password_own(self):
resp = self.client.post(f'/api/cms/users/{self.doc.id}/reset-password/', {})
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['password'], 'Pass13931000002')
def test_filter_by_role(self):
# 医生管理页 ?role_type=doctor
resp = self.client.get(CMS_USER_URL, {'role_type': 'doctor'})
self.assertEqual(resp.status_code, 200)
roles = {u['role_type'] for u in resp.json()['results']}
self.assertEqual(roles, {'doctor'})
test/test_cms_institution.py
+237
View File
@@ -0,0 +1,237 @@
"""CMS 超级管理员 - 机构（医院）管理接口测试（CMS-INST-1~6）。"""
import io
import tempfile
from openpyxl import Workbook
from django.core.files.uploadedfile import SimpleUploadedFile
from django.test import override_settings
from rest_framework.test import APIClient
from apps.user.models import Institution, Department
from .conftest import (
CacheTestCase,
create_test_user, get_auth_client, ensure_institution,
)
CMS_INST_URL = '/api/cms/institutions/'
XLSX_CT = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
def _xlsx(headers, rows):
wb = Workbook(); ws = wb.active
ws.append(headers)
for r in rows:
ws.append(r)
buf = io.BytesIO(); wb.save(buf); buf.seek(0)
return SimpleUploadedFile('inst.xlsx', buf.read(), content_type=XLSX_CT)
def inst_detail_url(pk):
return f'/api/cms/institutions/{pk}/'
def inst_banner_url(pk):
return f'/api/cms/institutions/{pk}/banner/'
def super_admin_client(phone='13911100001'):
admin = create_test_user(phone=phone, password='Admin123', role_type='super_admin')
return get_auth_client(admin), admin
# ── 权限 ──────────────────────────────────────────────────────────────────────
class CmsInstitutionPermissionTest(CacheTestCase):
def test_requires_auth(self):
"""未登录 → 401。"""
resp = APIClient().get(CMS_INST_URL)
self.assertEqual(resp.status_code, 401, resp.content)
def test_non_super_admin_forbidden(self):
"""非超管（学生/医院管理员）→ 403 CMS_PERMISSION_DENIED。"""
for role in ('student', 'hospital_admin', 'content_admin', 'doctor'):
user = create_test_user(phone=f'1391110100{("student hospital_admin content_admin doctor".split().index(role))}',
role_type=role)
client = get_auth_client(user)
resp = client.get(CMS_INST_URL)
self.assertEqual(resp.status_code, 403, f'{role}: {resp.content}')
self.assertEqual(resp.json()['code'], 'CMS_PERMISSION_DENIED')
# ── CRUD ──────────────────────────────────────────────────────────────────────
class CmsInstitutionCrudTest(CacheTestCase):
def setUp(self):
super().setUp()
self.client, self.admin = super_admin_client()
def test_list_paginated(self):
ensure_institution(name='协和医院', code='CMS-H001')
ensure_institution(name='同仁医院', code='CMS-H002')
resp = self.client.get(CMS_INST_URL)
self.assertEqual(resp.status_code, 200, resp.content)
data = resp.json()
self.assertIn('results', data) # DRF 分页
codes = {i['code'] for i in data['results']}
self.assertTrue({'CMS-H001', 'CMS-H002'} <= codes)
def test_list_search(self):
ensure_institution(name='北京协和医院', code='CMS-H010')
ensure_institution(name='上海瑞金医院', code='CMS-H011')
resp = self.client.get(CMS_INST_URL, {'search': '协和'})
self.assertEqual(resp.status_code, 200, resp.content)
results = resp.json()['results']
self.assertTrue(all('协和' in i['name'] for i in results))
self.assertTrue(any(i['code'] == 'CMS-H010' for i in results))
def test_create_success(self):
payload = {
'code': 'CMS-NEW-1', 'name': '新建示例医院',
'type': 'hospital', 'level': '三甲',
'province': '北京', 'city': '北京',
}
resp = self.client.post(CMS_INST_URL, payload)
self.assertEqual(resp.status_code, 201, resp.content)
body = resp.json()
self.assertEqual(body['code'], 'CMS-NEW-1')
self.assertEqual(body['name'], '新建示例医院')
self.assertEqual(body['banner_url'], '') # 未配图为空串
self.assertTrue(Institution.objects.filter(code='CMS-NEW-1').exists())
def test_create_duplicate_code(self):
ensure_institution(name='已存在', code='CMS-DUP')
resp = self.client.post(CMS_INST_URL, {'code': 'CMS-DUP', 'name': '重复编码'})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_INSTITUTION_CODE_EXISTS')
def test_retrieve(self):
inst = ensure_institution(name='详情医院', code='CMS-DET')
resp = self.client.get(inst_detail_url(inst.id))
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['id'], inst.id)
def test_update_patch(self):
inst = ensure_institution(name='旧名', code='CMS-UPD')
resp = self.client.patch(inst_detail_url(inst.id), {'name': '新名', 'level': '二甲'})
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['name'], '新名')
inst.refresh_from_db()
self.assertEqual(inst.name, '新名')
self.assertEqual(inst.level, '二甲')
def test_update_duplicate_code(self):
ensure_institution(name='A', code='CMS-A')
inst_b = ensure_institution(name='B', code='CMS-B')
resp = self.client.patch(inst_detail_url(inst_b.id), {'code': 'CMS-A'})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_INSTITUTION_CODE_EXISTS')
def test_update_same_code_ok(self):
"""编辑时传自己原 code 不算冲突。"""
inst = ensure_institution(name='自身', code='CMS-SELF')
resp = self.client.patch(inst_detail_url(inst.id), {'code': 'CMS-SELF', 'name': '改名'})
self.assertEqual(resp.status_code, 200, resp.content)
def test_delete_is_soft(self):
"""停用 = 逻辑删除：默认管理器查不到，但库里仍在（all_objects 可见）。"""
inst = ensure_institution(name='可停用', code='CMS-DEL')
resp = self.client.delete(inst_detail_url(inst.id))
self.assertEqual(resp.status_code, 204, resp.content)
# 默认管理器（已过滤 is_deleted）查不到
self.assertFalse(Institution.objects.filter(id=inst.id).exists())
# 实际未物理删除
obj = Institution.all_objects.get(id=inst.id)
self.assertTrue(obj.is_deleted)
self.assertIsNotNone(obj.deleted_at)
def test_deleted_not_in_list(self):
"""软删后不出现在列表。"""
inst = ensure_institution(name='停用后隐藏', code='CMS-HIDE')
self.client.delete(inst_detail_url(inst.id))
resp = self.client.get(CMS_INST_URL, {'search': 'CMS-HIDE'})
codes = {i['code'] for i in resp.json()['results']}
self.assertNotIn('CMS-HIDE', codes)
def test_put_not_allowed(self):
inst = ensure_institution(name='X', code='CMS-PUT')
resp = self.client.put(inst_detail_url(inst.id), {'code': 'CMS-PUT', 'name': 'Y'})
self.assertEqual(resp.status_code, 405, resp.content)
# ── Banner 上传（写临时静态目录，避免污染仓库）─────────────────────────────────
class CmsInstitutionBannerTest(CacheTestCase):
def setUp(self):
super().setUp()
self.client, self.admin = super_admin_client(phone='13911100050')
self.inst = ensure_institution(name='传图医院', code='CMS-BANNER')
self._tmp = tempfile.mkdtemp()
def _png(self, name='banner.png'):
# 最小合法 PNG 头 + 占位内容
content = b'\x89PNG\r\n\x1a\n' + b'0' * 64
return SimpleUploadedFile(name, content, content_type='image/png')
def test_upload_success(self):
with override_settings(STATICFILES_DIRS=[self._tmp]):
resp = self.client.post(inst_banner_url(self.inst.id),
{'file': self._png()}, format='multipart')
self.assertEqual(resp.status_code, 200, resp.content)
body = resp.json()
self.assertEqual(body['message'], '上传成功')
self.assertTrue(body['banner_url'].endswith(f'/static/institutions/inst_{self.inst.id}_banner.png'))
self.inst.refresh_from_db()
self.assertEqual(self.inst.banner_url, f'institutions/inst_{self.inst.id}_banner.png')
def test_upload_no_file(self):
with override_settings(STATICFILES_DIRS=[self._tmp]):
resp = self.client.post(inst_banner_url(self.inst.id), {}, format='multipart')
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_BANNER_FILE_REQUIRED')
def test_upload_bad_type(self):
bad = SimpleUploadedFile('x.txt', b'hello', content_type='text/plain')
with override_settings(STATICFILES_DIRS=[self._tmp]):
resp = self.client.post(inst_banner_url(self.inst.id),
{'file': bad}, format='multipart')
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_BANNER_BAD_TYPE')
class CmsInstitutionImportExportTest(CacheTestCase):
def setUp(self):
super().setUp()
self.client, self.admin = super_admin_client(phone='13911100090')
def test_import_template(self):
resp = self.client.get('/api/cms/institutions/import-template/')
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_export(self):
ensure_institution(name='导出医院', code='CMS-EXP-1')
resp = self.client.get('/api/cms/institutions/export/')
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_import(self):
ensure_institution(name='已存在', code='CMS-IMP-DUP')
f = _xlsx(
['机构编码', '名称', '类型', '等级', '省', '市'],
[
['CMS-IMP-1', '新医院A', 'hospital', '三甲', '北京', '北京'],
['', '无编码', 'hospital', '', '', ''], # 编码空 → 失败
['CMS-IMP-DUP', '重复编码', 'hospital', '', '', ''], # 重复 → 失败
],
)
resp = self.client.post('/api/cms/institutions/import/', {'file': f}, format='multipart')
self.assertEqual(resp.status_code, 200, resp.content)
body = resp.json()
self.assertEqual(body['success'], 1)
self.assertEqual(body['failed'], 2)
self.assertTrue(Institution.objects.filter(code='CMS-IMP-1').exists())
test/test_cms_relation.py
+99
View File
@@ -0,0 +1,99 @@
"""CMS 师生关系管理测试：CMS-REL-1~4（医院管理员本院 / 超管全平台）。"""
import io
from openpyxl import Workbook
from django.core.files.uploadedfile import SimpleUploadedFile
from rest_framework.test import APIClient
from apps.user.models import TeacherStudentRelation
from .conftest import CacheTestCase, create_test_user, get_auth_client, ensure_institution
REL_URL = '/api/cms/teacher-student-relations/'
XLSX_CT = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
def rel_detail(pk):
return f'/api/cms/teacher-student-relations/{pk}/'
def make_xlsx(headers, rows):
wb = Workbook(); ws = wb.active
ws.append(headers)
for r in rows:
ws.append(r)
buf = io.BytesIO(); wb.save(buf); buf.seek(0)
return SimpleUploadedFile('rel.xlsx', buf.read(), content_type=XLSX_CT)
class CmsRelationTest(CacheTestCase):
def setUp(self):
super().setUp()
self.inst = ensure_institution(name='本院', code='REL-A')
self.other = ensure_institution(name='他院', code='REL-B')
self.admin = create_test_user(phone='13932000001', role_type='hospital_admin', institution=self.inst)
self.client = get_auth_client(self.admin)
self.doc = create_test_user(phone='13932000002', role_type='doctor', institution=self.inst)
self.stu = create_test_user(phone='13932000003', role_type='student', institution=self.inst)
self.other_stu = create_test_user(phone='13932000004', role_type='student', institution=self.other)
def test_permission(self):
self.assertEqual(APIClient().get(REL_URL).status_code, 401)
stu_client = get_auth_client(self.stu)
self.assertEqual(stu_client.get(REL_URL).status_code, 403)
def test_create_and_list(self):
resp = self.client.post(REL_URL, {'teacher': self.doc.id, 'student': self.stu.id})
self.assertEqual(resp.status_code, 201, resp.content)
self.assertEqual(resp.json()['teacher_phone'], '13932000002')
self.assertEqual(resp.json()['student_phone'], '13932000003')
# 列表
resp = self.client.get(REL_URL)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.json()['results']), 1)
def test_create_duplicate(self):
self.client.post(REL_URL, {'teacher': self.doc.id, 'student': self.stu.id})
resp = self.client.post(REL_URL, {'teacher': self.doc.id, 'student': self.stu.id})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_REL_EXISTS')
def test_teacher_must_be_doctor(self):
# 用学生当 teacher → 无效 pk
resp = self.client.post(REL_URL, {'teacher': self.stu.id, 'student': self.stu.id})
self.assertEqual(resp.status_code, 400, resp.content)
def test_scope_other_institution_student_rejected(self):
resp = self.client.post(REL_URL, {'teacher': self.doc.id, 'student': self.other_stu.id})
self.assertEqual(resp.status_code, 403, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_REL_SCOPE_FORBIDDEN')
def test_soft_delete(self):
r = TeacherStudentRelation.objects.create(teacher=self.doc, student=self.stu, status=1)
resp = self.client.delete(rel_detail(r.id))
self.assertEqual(resp.status_code, 204, resp.content)
self.assertFalse(TeacherStudentRelation.objects.filter(id=r.id).exists())
self.assertTrue(TeacherStudentRelation.all_objects.get(id=r.id).is_deleted)
def test_import_and_export(self):
f = make_xlsx(['带教医生手机号', '学生手机号'], [
['13932000002', '13932000003'], # ✅
['13932000002', '13932000004'], # 学生他院 → 失败
['00000000000', '13932000003'], # 医生不存在 → 失败
])
resp = self.client.post('/api/cms/teacher-student-relations/import/', {'file': f}, format='multipart')
self.assertEqual(resp.status_code, 200, resp.content)
body = resp.json()
self.assertEqual(body['success'], 1)
self.assertEqual(body['failed'], 2)
# 导出
resp = self.client.get('/api/cms/teacher-student-relations/export/')
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_list_scoped_to_own(self):
# 他院的师生关系不可见
other_doc = create_test_user(phone='13932000005', role_type='doctor', institution=self.other)
TeacherStudentRelation.objects.create(teacher=other_doc, student=self.other_stu, status=1)
TeacherStudentRelation.objects.create(teacher=self.doc, student=self.stu, status=1)
resp = self.client.get(REL_URL)
self.assertEqual(len(resp.json()['results']), 1) # 只看到本院那条
test/test_cms_students.py
+128
View File
@@ -0,0 +1,128 @@
"""CMS 带教医生 - 我的学生测试：CMS-TEA-1~2。
带教医生（doctor）仅能看到 teacher_student_relation 中
teacher=自己 且 status=1 的学生；只读，不能新增/编辑/删除。
"""
from rest_framework.test import APIClient
from apps.user.models import TeacherStudentRelation
from .conftest import (
CacheTestCase,
create_test_user,
get_auth_client,
ensure_institution,
ensure_department,
create_teacher_student_relation,
)
STUDENTS_URL = '/api/cms/students/'
def student_detail(pk):
return f'/api/cms/students/{pk}/'
class CmsStudentsTest(CacheTestCase):
def setUp(self):
super().setUp()
self.inst = ensure_institution(name='本院', code='TEA-A')
self.dept = ensure_department(name='内科')
self.doc = create_test_user(
phone='13940000001', password='Doc12345',
real_name='张医生', role_type='doctor', institution=self.inst,
)
self.other_doc = create_test_user(
phone='13940000002', real_name='李医生',
role_type='doctor', institution=self.inst,
)
# 名下进行中学生
self.stu_own = create_test_user(
phone='13940000011', real_name='我的学生',
role_type='student', institution=self.inst,
)
self.stu_own.department = self.dept
self.stu_own.save(update_fields=['department'])
# 名下已结束学生（status=0，列表应排除）
self.stu_ended = create_test_user(
phone='13940000012', real_name='已毕业学生',
role_type='student', institution=self.inst,
)
# 其他医生的学生（当前医生不应看到）
self.stu_other = create_test_user(
phone='13940000013', real_name='其他学生',
role_type='student', institution=self.inst,
)
create_teacher_student_relation(self.doc, self.stu_own, status=1)
create_teacher_student_relation(self.doc, self.stu_ended, status=0)
create_teacher_student_relation(self.other_doc, self.stu_other, status=1)
self.client = get_auth_client(self.doc)
# ── 权限 ──────────────────────────────────────────────────────────────────
def test_unauthenticated_401(self):
self.assertEqual(APIClient().get(STUDENTS_URL).status_code, 401)
def test_non_doctor_403(self):
admin = create_test_user(phone='13940000091', role_type='hospital_admin',
institution=self.inst)
resp = get_auth_client(admin).get(STUDENTS_URL)
self.assertEqual(resp.status_code, 403, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_PERMISSION_DENIED')
# 学生本人也无权访问
resp2 = get_auth_client(self.stu_own).get(STUDENTS_URL)
self.assertEqual(resp2.status_code, 403, resp2.content)
# ── CMS-TEA-1 列表 ─────────────────────────────────────────────────────────
def test_list_own_active_students_only(self):
resp = self.client.get(STUDENTS_URL)
self.assertEqual(resp.status_code, 200, resp.content)
results = resp.json()['results']
ids = {u['id'] for u in results}
self.assertIn(self.stu_own.id, ids)
self.assertNotIn(self.stu_ended.id, ids) # status=0
self.assertNotIn(self.stu_other.id, ids) # 他医生
self.assertNotIn(self.doc.id, ids) # 不含自己
self.assertEqual(len(results), 1)
def test_list_search(self):
resp = self.client.get(STUDENTS_URL, {'search': '我的学生'})
self.assertEqual(resp.status_code, 200, resp.content)
ids = {u['id'] for u in resp.json()['results']}
self.assertEqual(ids, {self.stu_own.id})
def test_list_student_fields(self):
resp = self.client.get(STUDENTS_URL)
item = resp.json()['results'][0]
self.assertEqual(item['real_name'], '我的学生')
self.assertEqual(item['role_type'], 'student')
self.assertEqual(item['department_name'], '内科')
self.assertIn('total_training_count', item)
# ── CMS-TEA-2 详情 ─────────────────────────────────────────────────────────
def test_retrieve_own_student(self):
resp = self.client.get(student_detail(self.stu_own.id))
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['real_name'], '我的学生')
self.assertEqual(resp.json()['phone'], '13940000011')
def test_retrieve_other_student_404(self):
resp = self.client.get(student_detail(self.stu_other.id))
self.assertEqual(resp.status_code, 404, resp.content)
def test_retrieve_ended_student_404(self):
resp = self.client.get(student_detail(self.stu_ended.id))
self.assertEqual(resp.status_code, 404, resp.content)
# ── 软删除师生关系 ─────────────────────────────────────────────────────────
def test_soft_deleted_relation_excludes_student(self):
rel = TeacherStudentRelation.objects.get(teacher=self.doc, student=self.stu_own)
rel.delete() # SoftDeleteModel 逻辑删除
resp = self.client.get(STUDENTS_URL)
self.assertEqual(resp.status_code, 200, resp.content)
ids = {u['id'] for u in resp.json()['results']}
self.assertNotIn(self.stu_own.id, ids)
# ── 只读 ──────────────────────────────────────────────────────────────────
def test_readonly_methods_not_allowed(self):
self.assertEqual(self.client.post(STUDENTS_URL, {}).status_code, 405)
self.assertEqual(self.client.delete(student_detail(self.stu_own.id)).status_code, 405)
test/test_cms_user.py
+170
View File
@@ -0,0 +1,170 @@
"""CMS 超级管理员 - 用户管理接口测试（CMS-USER-1~8）。"""
import io
from openpyxl import Workbook
from django.core.files.uploadedfile import SimpleUploadedFile
from rest_framework.test import APIClient
from apps.user.models import User
from .conftest import CacheTestCase, create_test_user, get_auth_client, ensure_institution
CMS_USER_URL = '/api/cms/users/'
XLSX_CT = 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
def u_detail(pk):
return f'/api/cms/users/{pk}/'
def make_xlsx(headers, rows):
wb = Workbook(); ws = wb.active
ws.append(headers)
for r in rows:
ws.append(r)
buf = io.BytesIO(); wb.save(buf); buf.seek(0)
return SimpleUploadedFile('import.xlsx', buf.read(), content_type=XLSX_CT)
def super_client(phone='13922200001'):
admin = create_test_user(phone=phone, password='Admin123', role_type='super_admin')
return get_auth_client(admin), admin
class CmsUserPermissionTest(CacheTestCase):
def test_requires_auth(self):
self.assertEqual(APIClient().get(CMS_USER_URL).status_code, 401)
def test_non_manager_forbidden(self):
# 用户管理仅超管 + 医院管理员；其它角色 403
for role in ('student', 'doctor', 'content_admin'):
u = create_test_user(phone={'student': '13922200010', 'doctor': '13922200011',
'content_admin': '13922200012'}[role], role_type=role)
resp = get_auth_client(u).get(CMS_USER_URL)
self.assertEqual(resp.status_code, 403, f'{role}: {resp.content}')
self.assertEqual(resp.json()['code'], 'CMS_PERMISSION_DENIED')
class CmsUserCrudTest(CacheTestCase):
def setUp(self):
super().setUp()
self.client, self.admin = super_client()
self.inst = ensure_institution(name='测试医院', code='CU-H001')
def test_list(self):
create_test_user(phone='13922200021', real_name='学生甲', role_type='student')
resp = self.client.get(CMS_USER_URL)
self.assertEqual(resp.status_code, 200, resp.content)
self.assertIn('results', resp.json())
def test_create_success(self):
resp = self.client.post(CMS_USER_URL, {
'phone': '13922200030', 'real_name': '李医生',
'role_type': 'doctor', 'institution': self.inst.id,
})
self.assertEqual(resp.status_code, 201, resp.content)
self.assertEqual(resp.json()['phone'], '13922200030')
self.assertEqual(resp.json()['role_type'], 'doctor')
user = User.objects.get(phone='13922200030')
self.assertTrue(user.check_password('Pass13922200030')) # 默认密码
def test_create_role_required(self):
resp = self.client.post(CMS_USER_URL, {
'phone': '13922200031', 'real_name': '无角色', 'institution': self.inst.id})
self.assertEqual(resp.status_code, 400, resp.content)
def test_create_institution_required(self):
resp = self.client.post(CMS_USER_URL, {
'phone': '13922200032', 'real_name': '无机构', 'role_type': 'student'})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_VALIDATION_ERROR')
def test_create_bad_phone(self):
resp = self.client.post(CMS_USER_URL, {
'phone': '123', 'real_name': 'x', 'role_type': 'student', 'institution': self.inst.id})
self.assertEqual(resp.status_code, 400, resp.content)
def test_create_duplicate_phone(self):
create_test_user(phone='13922200040', role_type='student')
resp = self.client.post(CMS_USER_URL, {
'phone': '13922200040', 'real_name': 'dup', 'role_type': 'student',
'institution': self.inst.id})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_USER_PHONE_EXISTS')
def test_edit_partial_keeps_role_institution(self):
"""方案B：只改姓名、不带角色/机构 → 200，角色与机构保持原值。"""
u = create_test_user(phone='13922200050', real_name='原名', role_type='student',
institution=self.inst)
resp = self.client.patch(u_detail(u.id), {'real_name': '新名'})
self.assertEqual(resp.status_code, 200, resp.content)
u.refresh_from_db()
self.assertEqual(u.real_name, '新名')
self.assertEqual(u.role_type, 'student') # 未传 → 保持
self.assertEqual(u.institution_id, self.inst.id) # 未传 → 保持
def test_edit_cannot_blank_role(self):
"""方案B：传了 role_type 但为空 → 400（不可清空角色）。"""
u = create_test_user(phone='13922200051', real_name='x', role_type='student',
institution=self.inst)
resp = self.client.patch(u_detail(u.id), {'role_type': ''})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_VALIDATION_ERROR')
def test_edit_cannot_blank_institution(self):
"""方案B：传了 institution=null → 400（不可清空机构）。"""
u = create_test_user(phone='13922200052', real_name='x', role_type='student',
institution=self.inst)
resp = self.client.patch(u_detail(u.id), {'institution': None}, format='json')
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_VALIDATION_ERROR')
def test_soft_delete(self):
u = create_test_user(phone='13922200060', role_type='student')
resp = self.client.delete(u_detail(u.id))
self.assertEqual(resp.status_code, 204, resp.content)
self.assertFalse(User.objects.filter(id=u.id).exists()) # 默认管理器过滤
obj = User.all_objects.get(id=u.id)
self.assertTrue(obj.is_deleted) # 实际未物删
def test_reset_password(self):
u = create_test_user(phone='13922200070', password='OldPass1', role_type='student')
resp = self.client.post(f'/api/cms/users/{u.id}/reset-password/', {})
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['password'], 'Pass13922200070')
u.refresh_from_db()
self.assertTrue(u.check_password('Pass13922200070'))
def test_import_template(self):
resp = self.client.get('/api/cms/users/import-template/')
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_export(self):
create_test_user(phone='13922200080', role_type='student')
resp = self.client.get('/api/cms/users/export/')
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], XLSX_CT)
def test_import_users(self):
f = make_xlsx(
['手机号', '姓名', '角色', '机构编码'],
[
['13922200091', '导入甲', '医生', 'CU-H001'], # ✅
['13922200092', '导入乙', 'student', ''], # 机构编码空 → 失败
['bad', '格式错', '学生', 'CU-H001'], # 手机号格式错
['13922200093', '角色错', '不存在角色', 'CU-H001'], # 角色非法
],
)
resp = self.client.post('/api/cms/users/import/', {'file': f}, format='multipart')
self.assertEqual(resp.status_code, 200, resp.content)
body = resp.json()
self.assertEqual(body['total'], 4)
self.assertEqual(body['success'], 1)
self.assertEqual(body['failed'], 3)
self.assertTrue(User.objects.filter(phone='13922200091', role_type='doctor').exists())
def test_import_no_file(self):
resp = self.client.post('/api/cms/users/import/', {}, format='multipart')
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'CMS_IMPORT_FILE_REQUIRED')
test/test_profile_config.py
+21 -21
View File
@@ -69,16 +69,13 @@ class InstitutionInfoTest(CacheTestCase):
class MyDepartmentsTest(CacheTestCase):
"""所属机构科室列表接口（不分页）。"""
def test_returns_all_departments_no_pagination(self):
"""返回本机构全部科室，且不分页（直接为列表）。"""
inst = ensure_institution(name='测试医院', code='TEST-HOSP-001')
Department.objects.create(institution=inst, name='内科', category='临床')
Department.objects.create(institution=inst, name='外科', category='临床')
# 另一机构的科室不应出现
other = ensure_institution(name='其他医院', code='TEST-HOSP-OTHER')
Department.objects.create(institution=other, name='儿科', category='临床')
def test_returns_all_global_departments_no_pagination(self):
"""返回全部全局科室（与机构无关），不分页（直接为列表）。"""
Department.objects.create(name='内科', category='临床')
Department.objects.create(name='外科', category='临床')
Department.objects.create(name='儿科', category='临床')
user = create_test_user(phone='13900200010', institution=inst)
user = create_test_user(phone='13900200010')
client = get_auth_client(user)
resp = client.get(USER_MY_DEPARTMENTS_URL)
self.assertEqual(resp.status_code, 200, resp.content)
@@ -86,13 +83,16 @@ class MyDepartmentsTest(CacheTestCase):
data = resp.json()
self.assertIsInstance(data, list) # 不分页：顶层是列表
names = {d['name'] for d in data}
self.assertEqual(names, {'内科', '外科'})
self.assertEqual(names, {'内科', '外科', '儿科'})
def test_user_without_institution_404(self):
def test_no_institution_still_returns_departments(self):
"""科室全局，用户没机构也能拿到全部科室。"""
Department.objects.create(name='内科', category='临床')
user = create_test_user(phone='13900200011', institution=None)
client = get_auth_client(user)
resp = client.get(USER_MY_DEPARTMENTS_URL)
self.assertEqual(resp.status_code, 404, resp.content)
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual({d['name'] for d in resp.json()}, {'内科'})
class ProfileConfigTest(CacheTestCase):
@@ -101,7 +101,7 @@ class ProfileConfigTest(CacheTestCase):
def test_config_success(self):
"""录入 科室/职称/执业年限 → 落库成功。"""
inst = ensure_institution(name='测试医院', code='TEST-HOSP-001')
dept = Department.objects.create(institution=inst, name='内科', category='临床')
dept = Department.objects.create(name='内科', category='临床')
user = create_test_user(phone='13900200020', institution=inst)
client = get_auth_client(user)
@@ -118,20 +118,20 @@ class ProfileConfigTest(CacheTestCase):
self.assertEqual(user.practice_years, '1-3年')
self.assertEqual(resp.json()['user']['practice_years'], '1-3年')
def test_department_from_other_institution_rejected(self):
"""所选科室不属于本机构 → 校验失败 400。"""
inst = ensure_institution(name='测试医院', code='TEST-HOSP-001')
other = ensure_institution(name='其他医院', code='TEST-HOSP-OTHER')
other_dept = Department.objects.create(institution=other, name='外科', category='临床')
user = create_test_user(phone='13900200021', institution=inst)
def test_any_global_department_accepted(self):
"""科室全局：可选择任意科室（不再校验机构归属）。"""
dept = Department.objects.create(name='外科', category='临床')
user = create_test_user(phone='13900200021', institution=None)
client = get_auth_client(user)
resp = client.post(USER_PROFILE_CONFIG_URL, {
'department': other_dept.id,
'department': dept.id,
'title_name': '住院医师',
'practice_years': '1-3年',
})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.status_code, 200, resp.content)
user.refresh_from_db()
self.assertEqual(user.department_id, dept.id)
def test_missing_fields_rejected(self):
"""缺少必填字段 → 400。"""
test/test_profile_personal.py
+131
View File
@@ -0,0 +1,131 @@
"""移动端个人中心：个人信息获取(GET) / 更新(PATCH) 接口测试。"""
from rest_framework.test import APIClient
from apps.user.models import Department
from .conftest import (
CacheTestCase,
create_test_user, get_auth_client, ensure_institution,
)
PROFILE_URL = '/api/user/profile/'
class ProfileGetTest(CacheTestCase):
def test_get_full_info(self):
inst = ensure_institution(name='测试医院', code='PF-H001')
user = create_test_user(phone='13912300001', real_name='张三', institution=inst)
user.practice_years = '1-3年'
user.title_name = '住院医师'
user.save(update_fields=['practice_years', 'title_name'])
client = get_auth_client(user)
resp = client.get(PROFILE_URL)
self.assertEqual(resp.status_code, 200, resp.content)
data = resp.json()
# 全量信息
self.assertEqual(data['id'], user.id)
self.assertEqual(data['real_name'], '张三')
self.assertEqual(data['practice_years'], '1-3年')
self.assertEqual(data['title_name'], '住院医师')
# 机构/角色等只读信息也返回（供展示，但不可改）
self.assertEqual(data['institution'], inst.id)
self.assertEqual(data['role_type'], 'student')
def test_get_requires_auth(self):
resp = APIClient().get(PROFILE_URL)
self.assertEqual(resp.status_code, 401, resp.content)
class ProfileUpdateTest(CacheTestCase):
def setUp(self):
super().setUp()
self.inst = ensure_institution(name='测试医院', code='PF-H001')
self.user = create_test_user(phone='13912300010', real_name='原名', institution=self.inst)
self.client = get_auth_client(self.user)
def test_update_allowed_fields(self):
dept = Department.objects.create(name='内科', category='临床')
payload = {
'username': 'newuser001',
'real_name': '李四',
'avatar': 'https://cdn.x.com/a.png',
'gender': 1,
'department': dept.id,
'title_name': '主治医师',
'major': '心血管内科',
'practice_years': '3-5年',
'training_stage': '规培',
'learning_target': '提升问诊能力',
}
resp = self.client.patch(PROFILE_URL, payload)
self.assertEqual(resp.status_code, 200, resp.content)
self.assertEqual(resp.json()['message'], '更新成功')
self.user.refresh_from_db()
self.assertEqual(self.user.username, 'newuser001')
self.assertEqual(self.user.real_name, '李四')
self.assertEqual(self.user.avatar, 'https://cdn.x.com/a.png')
self.assertEqual(self.user.gender, 1)
self.assertEqual(self.user.department_id, dept.id)
self.assertEqual(self.user.title_name, '主治医师')
self.assertEqual(self.user.major, '心血管内科')
self.assertEqual(self.user.practice_years, '3-5年')
self.assertEqual(self.user.training_stage, '规培')
self.assertEqual(self.user.learning_target, '提升问诊能力')
def test_update_phone(self):
resp = self.client.patch(PROFILE_URL, {'phone': '13900008888'})
self.assertEqual(resp.status_code, 200, resp.content)
self.user.refresh_from_db()
self.assertEqual(self.user.phone, '13900008888')
def test_cannot_change_institution_role_superuser(self):
"""机构 / 角色 / is_superuser 不在白名单，传了也被忽略。"""
other = ensure_institution(name='其他医院', code='PF-OTHER')
resp = self.client.patch(PROFILE_URL, {
'real_name': '改个名',
'institution': other.id,
'role_type': 'super_admin',
'is_superuser': True,
})
self.assertEqual(resp.status_code, 200, resp.content)
self.user.refresh_from_db()
self.assertEqual(self.user.real_name, '改个名') # 白名单字段生效
self.assertEqual(self.user.institution_id, self.inst.id) # 机构未变
self.assertEqual(self.user.role_type, 'student') # 角色未变
self.assertFalse(self.user.is_superuser) # 仍非超管
def test_username_taken(self):
create_test_user(phone='13912300099', real_name='别人')
# 把别人的 username 抢过来
other_username = 'taken_name'
u = create_test_user(phone='13912300098', real_name='占名者')
u.username = other_username
u.save(update_fields=['username'])
resp = self.client.patch(PROFILE_URL, {'username': other_username})
self.assertEqual(resp.status_code, 400, resp.content)
self.assertEqual(resp.json()['code'], 'VALIDATION_ERROR')
def test_phone_invalid_format(self):
resp = self.client.patch(PROFILE_URL, {'phone': '12345'})
self.assertEqual(resp.status_code, 400, resp.content)
def test_phone_taken(self):
create_test_user(phone='13912300077', real_name='占号者')
resp = self.client.patch(PROFILE_URL, {'phone': '13912300077'})
self.assertEqual(resp.status_code, 400, resp.content)
def test_any_global_department_accepted(self):
"""科室全局：可更新为任意科室（不再校验机构归属）。"""
dept = Department.objects.create(name='外科', category='临床')
resp = self.client.patch(PROFILE_URL, {'department': dept.id})
self.assertEqual(resp.status_code, 200, resp.content)
self.user.refresh_from_db()
self.assertEqual(self.user.department_id, dept.id)
def test_update_requires_auth(self):
resp = APIClient().patch(PROFILE_URL, {'real_name': 'x'})
self.assertEqual(resp.status_code, 401, resp.content)