"""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) def test_recreate_soft_deleted_code_returns_400(self): """软删后用相同编码重建：返回 400 CMS_INSTITUTION_CODE_EXISTS（而非 500）。 编码唯一约束对软删行仍生效，须按 all_objects 校验，避免写库时撞约束抛 500。 """ inst = ensure_institution(name='待停用', code='CMS-SOFT-DUP') self.client.delete(inst_detail_url(inst.id)) self.assertFalse(Institution.objects.filter(code='CMS-SOFT-DUP').exists()) resp = self.client.post(CMS_INST_URL, {'code': 'CMS-SOFT-DUP', 'name': '重建'}) self.assertEqual(resp.status_code, 400, resp.content) self.assertEqual(resp.json()['code'], 'CMS_INSTITUTION_CODE_EXISTS') # 不应产生重复行（仍只有那条已软删的） self.assertEqual(Institution.all_objects.filter(code='CMS-SOFT-DUP').count(), 1) # ── 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())