华佗
e31337b58a
- 04_test_business_logic.py: 业务逻辑测试v1（111用例） - 04_test_business_logic_v2.py: 修正API路径后v2（107用例，通过率31.8%） - 测试报告: 揭示大量API路径不匹配和参数问题 - 测试数据: SQL脚本覆盖31个业务模块 - 测试流程: 30个业务流程图+API映射 测试发现的问题: 1. 多个Controller缺少/page端点 2. 部分接口需要必填参数（patientId, startTime等） 3. 部分接口响应格式非标准（rows嵌套为dict） 4. DB列名不匹配（create_by不存在等）
1021 lines
37 KiB
Python
Executable File
1021 lines
37 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
HealthLink-HIS 三甲医院全流程业务逻辑测试
|
||
版本: v2.0
|
||
日期: 2026-06-07
|
||
|
||
测试理念：
|
||
- 不再只判断HTTP 200/500
|
||
- 验证业务数据正确性（字段存在、值正确、关联关系正确）
|
||
- 验证业务流程链路（A→B→C步骤的因果关系）
|
||
- 验证异常场景（参数缺失、权限不足、数据不存在）
|
||
- 验证数据一致性（创建后查询能查到、更新后值改变、删除后查不到）
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Dict, Any, List, Tuple, Optional
|
||
|
||
# ============================
|
||
# 配置
|
||
# ============================
|
||
BASE_URL = "http://localhost:18082/healthlink-his"
|
||
ADMIN_USER = "admin"
|
||
ADMIN_PASS = "admin123"
|
||
TENANT_ID = "1"
|
||
|
||
# 测试结果统计
|
||
class TestStats:
|
||
def __init__(self):
|
||
self.total = 0
|
||
self.passed = 0
|
||
self.failed = 0
|
||
self.skipped = 0
|
||
self.results = []
|
||
|
||
def record(self, module: str, case_id: str, name: str, passed: bool, detail: str = ""):
|
||
self.total += 1
|
||
if passed:
|
||
self.passed += 1
|
||
status = "✅ PASS"
|
||
else:
|
||
self.failed += 1
|
||
status = "❌ FAIL"
|
||
self.results.append({
|
||
"module": module,
|
||
"case_id": case_id,
|
||
"name": name,
|
||
"status": status,
|
||
"detail": detail
|
||
})
|
||
print(f" {status} [{module}] {case_id}: {name}")
|
||
if detail and not passed:
|
||
print(f" → {detail}")
|
||
|
||
def summary(self):
|
||
print("\n" + "=" * 70)
|
||
print(f"测试汇总: 总数={self.total}, 通过={self.passed}, 失败={self.failed}")
|
||
if self.total > 0:
|
||
rate = self.passed * 100 / self.total
|
||
print(f"通过率: {rate:.1f}%")
|
||
print("=" * 70)
|
||
return self.failed == 0
|
||
|
||
stats = TestStats()
|
||
TOKEN = ""
|
||
|
||
# ============================
|
||
# 工具函数
|
||
# ============================
|
||
def login() -> str:
|
||
"""登录获取Token"""
|
||
resp = requests.post(f"{BASE_URL}/login", json={
|
||
"username": ADMIN_USER,
|
||
"password": ADMIN_PASS,
|
||
"tenantId": TENANT_ID
|
||
})
|
||
data = resp.json()
|
||
return data.get("token", "")
|
||
|
||
def headers() -> Dict:
|
||
return {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
|
||
|
||
def api_get(path: str, params: Dict = None) -> Dict:
|
||
resp = requests.get(f"{BASE_URL}{path}", headers=headers(), params=params)
|
||
return resp.json()
|
||
|
||
def api_post(path: str, data: Dict = None) -> Dict:
|
||
resp = requests.post(f"{BASE_URL}{path}", headers=headers(), json=data)
|
||
return resp.json()
|
||
|
||
def api_put(path: str, data: Dict = None) -> Dict:
|
||
resp = requests.put(f"{BASE_URL}{path}", headers=headers(), json=data)
|
||
return resp.json()
|
||
|
||
def api_delete(path: str) -> Dict:
|
||
resp = requests.delete(f"{BASE_URL}{path}", headers=headers())
|
||
return resp.json()
|
||
|
||
def assert_response(resp: Dict, module: str, case_id: str, name: str,
|
||
expected_code: int = 200,
|
||
check_fields: List[str] = None,
|
||
check_values: Dict[str, Any] = None,
|
||
check_not_empty: List[str] = None):
|
||
"""通用断言：检查响应码、字段存在、字段值"""
|
||
passed = True
|
||
detail = ""
|
||
|
||
# 1. 检查HTTP响应码（业务码）
|
||
actual_code = resp.get("code")
|
||
if actual_code != expected_code:
|
||
passed = False
|
||
detail = f"预期code={expected_code}, 实际code={actual_code}, msg={resp.get('msg','')}"
|
||
stats.record(module, case_id, name, passed, detail)
|
||
return resp
|
||
|
||
# 2. 检查字段存在
|
||
if check_fields:
|
||
for field in check_fields:
|
||
if field not in resp:
|
||
passed = False
|
||
detail = f"响应缺少字段: {field}"
|
||
break
|
||
|
||
# 3. 检查字段值
|
||
if check_values and passed:
|
||
for field, expected_val in check_values.items():
|
||
actual_val = resp.get(field)
|
||
if actual_val != expected_val:
|
||
passed = False
|
||
detail = f"字段{field}: 预期={expected_val}, 实际={actual_val}"
|
||
break
|
||
|
||
# 4. 检查列表非空
|
||
if check_not_empty and passed:
|
||
for field in check_not_empty:
|
||
val = resp.get(field)
|
||
if val is None or (isinstance(val, (list, dict)) and len(val) == 0):
|
||
passed = False
|
||
detail = f"字段{field}为空"
|
||
break
|
||
|
||
stats.record(module, case_id, name, passed, detail)
|
||
return resp
|
||
|
||
def assert_page_response(resp: Dict, module: str, case_id: str, name: str,
|
||
min_rows: int = 0, max_rows: int = 10000):
|
||
"""断言分页查询响应"""
|
||
passed = True
|
||
detail = ""
|
||
|
||
if resp.get("code") != 200:
|
||
passed = False
|
||
detail = f"code={resp.get('code')}, msg={resp.get('msg','')}"
|
||
else:
|
||
rows = resp.get("rows", resp.get("data", []))
|
||
total = resp.get("total", 0)
|
||
if not isinstance(rows, list):
|
||
passed = False
|
||
detail = f"rows不是数组类型: {type(rows)}"
|
||
elif len(rows) < min_rows:
|
||
passed = False
|
||
detail = f"rows数量={len(rows)}, 最少需要{min_rows}"
|
||
elif total < min_rows:
|
||
passed = False
|
||
detail = f"total={total}, 最少需要{min_rows}"
|
||
|
||
stats.record(module, case_id, name, passed, detail)
|
||
return resp
|
||
|
||
# ============================
|
||
# 测试模块1: 系统登录认证
|
||
# ============================
|
||
def test_auth():
|
||
print("\n" + "=" * 50)
|
||
print("模块1: 系统登录认证")
|
||
print("=" * 50)
|
||
|
||
# 1.1 登录成功
|
||
resp = api_post("/login", {"username": "admin", "password": "admin123", "tenantId": "1"})
|
||
assert_response(resp, "认证", "1.1", "登录成功验证",
|
||
expected_code=200,
|
||
check_fields=["token", "permissions", "roles"],
|
||
check_not_empty=["token"])
|
||
|
||
# 1.2 登录失败 - 错误密码
|
||
resp = api_post("/login", {"username": "admin", "password": "wrongpass", "tenantId": "1"})
|
||
assert_response(resp, "认证", "1.2", "错误密码应返回失败",
|
||
expected_code=500) # 若依框架返回500表示业务失败
|
||
|
||
# 1.3 获取用户信息
|
||
resp = api_get("/getInfo")
|
||
assert_response(resp, "认证", "1.3", "获取用户信息",
|
||
expected_code=200,
|
||
check_fields=["user", "roles", "permissions"],
|
||
check_not_empty=["user"])
|
||
|
||
# 1.4 获取路由菜单
|
||
resp = api_get("/getRouters")
|
||
assert_response(resp, "认证", "1.4", "获取路由菜单",
|
||
expected_code=200,
|
||
check_not_empty=["data"])
|
||
|
||
# 1.5 获取验证码
|
||
resp = requests.get(f"{BASE_URL}/captchaImage")
|
||
captcha_data = resp.json()
|
||
assert_response(captcha_data, "认证", "1.5", "获取验证码",
|
||
expected_code=200,
|
||
check_fields=["img", "uuid"],
|
||
check_not_empty=["img", "uuid"])
|
||
|
||
# ============================
|
||
# 测试模块2: 门诊挂号流程
|
||
# ============================
|
||
def test_registration():
|
||
print("\n" + "=" * 50)
|
||
print("模块2: 门诊挂号流程")
|
||
print("=" * 50)
|
||
|
||
# 2.1 挂号初始化 - 应返回优先级选项
|
||
resp = api_get("/charge-manage/register/init")
|
||
assert_response(resp, "挂号", "2.1", "挂号初始化-返回优先级选项",
|
||
expected_code=200,
|
||
check_fields=["priorityLevelOptionOptions"])
|
||
|
||
# 2.2 挂号列表查询 - 应返回分页数据
|
||
resp = api_get("/charge-manage/register/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "挂号", "2.2", "挂号列表分页查询", min_rows=0)
|
||
|
||
# 2.3 查询患者信息 - 应返回患者列表
|
||
resp = api_get("/charge-manage/register/patient", {"searchKey": "测试"})
|
||
assert_response(resp, "挂号", "2.3", "查询患者-搜索'测试'",
|
||
expected_code=200)
|
||
|
||
# 2.4 查询患者 - 空搜索应返回提示
|
||
resp = api_get("/charge-manage/register/patient", {"searchKey": "不存在的患者XYZ"})
|
||
assert_response(resp, "挂号", "2.4", "查询不存在的患者",
|
||
expected_code=200) # 应正常返回空列表
|
||
|
||
# ============================
|
||
# 测试模块3: 门诊医生站
|
||
# ============================
|
||
def test_doctor_station():
|
||
print("\n" + "=" * 50)
|
||
print("模块3: 门诊医生站")
|
||
print("=" * 50)
|
||
|
||
# 3.1 待诊患者列表
|
||
resp = api_get("/doctor-station/main/patient-list")
|
||
assert_response(resp, "医生站", "3.1", "待诊患者列表",
|
||
expected_code=200)
|
||
|
||
# 3.2 医嘱列表查询
|
||
resp = api_get("/doctor-station/advice/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "医生站", "3.2", "医嘱列表分页查询")
|
||
|
||
# 3.3 诊断列表查询
|
||
resp = api_get("/doctor-station/diagnosis/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "医生站", "3.3", "诊断列表分页查询")
|
||
|
||
# 3.4 检查申请列表
|
||
resp = api_get("/doctor-station/inspection/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "医生站", "3.4", "检查申请列表")
|
||
|
||
# ============================
|
||
# 测试模块4: 收费管理
|
||
# ============================
|
||
def test_charge():
|
||
print("\n" + "=" * 50)
|
||
print("模块4: 收费管理")
|
||
print("=" * 50)
|
||
|
||
# 4.1 收费初始化
|
||
resp = api_get("/charge-manage/charge/init")
|
||
assert_response(resp, "收费", "4.1", "收费初始化",
|
||
expected_code=200)
|
||
|
||
# 4.2 收费记录查询
|
||
resp = api_get("/charge-manage/charge/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "收费", "4.2", "收费记录分页查询")
|
||
|
||
# 4.3 退费记录查询
|
||
resp = api_get("/charge-manage/refund/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "收费", "4.3", "退费记录分页查询")
|
||
|
||
# 4.4 收费定价查询
|
||
resp = api_get("/charge-manage/pricing/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "收费", "4.4", "收费定价列表")
|
||
|
||
# ============================
|
||
# 测试模块5: 住院管理
|
||
# ============================
|
||
def test_inpatient():
|
||
print("\n" + "=" * 50)
|
||
print("模块5: 住院管理")
|
||
print("=" * 50)
|
||
|
||
# 5.1 入院登记列表
|
||
resp = api_get("/inhospitalmanage/register/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "住院", "5.1", "入院登记列表")
|
||
|
||
# 5.2 患者主页初始化
|
||
resp = api_get("/patient-home-manage/init")
|
||
assert_response(resp, "住院", "5.2", "患者主页初始化",
|
||
expected_code=200)
|
||
|
||
# 5.3 空床查询
|
||
resp = api_get("/patient-home-manage/empty-bed")
|
||
assert_response(resp, "住院", "5.3", "空床查询",
|
||
expected_code=200)
|
||
|
||
# 5.4 押金管理初始化
|
||
resp = api_get("/deposit-manage/init")
|
||
assert_response(resp, "住院", "5.4", "押金管理初始化",
|
||
expected_code=200)
|
||
|
||
# 5.5 押金记录查询
|
||
resp = api_get("/deposit-manage/deposit-page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "住院", "5.5", "押金记录分页查询")
|
||
|
||
# 5.6 住院收费记录
|
||
resp = api_get("/charge-manage/inpatient-charge/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "住院", "5.6", "住院收费记录")
|
||
|
||
# ============================
|
||
# 测试模块6: 护理管理
|
||
# ============================
|
||
def test_nursing():
|
||
print("\n" + "=" * 50)
|
||
print("模块6: 护理管理")
|
||
print("=" * 50)
|
||
|
||
# 6.1 护理评估列表
|
||
resp = api_get("/nursing-assessment-enhanced/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.1", "护理评估列表")
|
||
|
||
# 6.2 护理评估统计
|
||
resp = api_get("/nursing-assessment-enhanced/stats")
|
||
assert_response(resp, "护理", "6.2", "护理评估统计",
|
||
expected_code=200)
|
||
|
||
# 6.3 Braden压疮评估 - 业务逻辑验证
|
||
braden_data = {
|
||
"patientName": "测试患者甲",
|
||
"encounterId": "6006",
|
||
"itemScores": json.dumps({"sensation": 2, "moisture": 2, "activity": 1, "mobility": 2, "nutrition": 3, "friction": 2}),
|
||
"detail": "压疮高危患者，需每2小时翻身"
|
||
}
|
||
resp = api_post("/nursing-assessment-enhanced/braden/assess", braden_data)
|
||
# 验证：评估应成功，且返回的评估分数应为12 (2+2+1+2+3+2)
|
||
assert_response(resp, "护理", "6.3", "Braden压疮评估-分数计算正确",
|
||
expected_code=200)
|
||
|
||
# 6.4 Morse跌倒评估
|
||
morse_data = {
|
||
"patientName": "测试患者乙",
|
||
"encounterId": "6007",
|
||
"itemScores": json.dumps({"history": 15, "diagnosis": 0, "ambulation": 15, "iv": 20, "gait": 0, "mental": 15}),
|
||
"detail": "跌倒高危患者，需加强防护"
|
||
}
|
||
resp = api_post("/nursing-assessment-enhanced/morse/assess", morse_data)
|
||
# 验证：Morse评分应为65 (15+0+15+20+0+15)，属于高危
|
||
assert_response(resp, "护理", "6.4", "Morse跌倒评估-分数计算正确",
|
||
expected_code=200)
|
||
|
||
# 6.5 护理记录患者列表
|
||
resp = api_get("/nursing-record/patient-page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.5", "护理记录患者列表")
|
||
|
||
# 6.6 体征记录查询
|
||
resp = api_get("/vital-signs/record-search")
|
||
assert_response(resp, "护理", "6.6", "体征记录查询",
|
||
expected_code=200)
|
||
|
||
# 6.7 体征图表查询
|
||
resp = api_get("/vital-signs-chart/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.7", "体征图表分页查询")
|
||
|
||
# 6.8 护理执行列表
|
||
resp = api_get("/nurse-station/advice-process/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.8", "护理执行列表")
|
||
|
||
# 6.9 交接班记录
|
||
resp = api_get("/nursing-handoff/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.9", "交接班记录查询")
|
||
|
||
# 6.10 护理质量指标
|
||
resp = api_get("/nursing-quality/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "护理", "6.10", "护理质量指标查询")
|
||
|
||
# ============================
|
||
# 测试模块7: 检验检查
|
||
# ============================
|
||
def test_inspection():
|
||
print("\n" + "=" * 50)
|
||
print("模块7: 检验检查")
|
||
print("=" * 50)
|
||
|
||
# 7.1 标本采集列表
|
||
resp = api_get("/inspection/collection/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.1", "标本采集列表")
|
||
|
||
# 7.2 检验观察定义
|
||
resp = api_get("/inspection/observation/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.2", "检验观察定义列表")
|
||
|
||
# 7.3 标本定义
|
||
resp = api_get("/inspection/specimen/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.3", "标本定义列表")
|
||
|
||
# 7.4 LIS配置
|
||
resp = api_get("/inspection/lisConfig/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.4", "LIS配置列表")
|
||
|
||
# 7.5 仪器管理
|
||
resp = api_get("/inspection/instrument/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.5", "仪器管理列表")
|
||
|
||
# 7.6 检验结果
|
||
resp = api_get("/inspection/laboratory/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.6", "检验结果列表")
|
||
|
||
# 7.7 参考范围
|
||
resp = api_get("/lab-ref-range/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.7", "参考范围列表")
|
||
|
||
# 7.8 检查申请
|
||
resp = api_get("/check/examApply/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "检验", "7.8", "检查申请列表")
|
||
|
||
# ============================
|
||
# 测试模块8: 影像检查
|
||
# ============================
|
||
def test_radiology():
|
||
print("\n" + "=" * 50)
|
||
print("模块8: 影像检查")
|
||
print("=" * 50)
|
||
|
||
# 8.1 影像列表
|
||
resp = api_get("/check/radiologyImage/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "影像", "8.1", "影像列表查询")
|
||
|
||
# 8.2 影像增强
|
||
resp = api_get("/check/radiologyEnhanced/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "影像", "8.2", "影像增强列表")
|
||
|
||
# 8.3 影像对比
|
||
resp = api_get("/check/radiologyComparison/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "影像", "8.3", "影像对比列表")
|
||
|
||
# 8.4 3D重建
|
||
resp = api_get("/reconstruction/3d/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "影像", "8.4", "3D重建列表")
|
||
|
||
# ============================
|
||
# 测试模块9: 手术麻醉
|
||
# ============================
|
||
def test_surgery():
|
||
print("\n" + "=" * 50)
|
||
print("模块9: 手术麻醉")
|
||
print("=" * 50)
|
||
|
||
# 9.1 手术列表
|
||
resp = api_get("/clinical-manage/surgery/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.1", "手术列表查询")
|
||
|
||
# 9.2 手术排程
|
||
resp = api_get("/clinical-manage/surgery-schedule/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.2", "手术排程列表")
|
||
|
||
# 9.3 术前讨论
|
||
resp = api_get("/preopmanage/discussion/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.3", "术前讨论列表")
|
||
|
||
# 9.4 手术安全核查
|
||
resp = api_get("/surgery-safety-check/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.4", "手术安全核查列表")
|
||
|
||
# 9.5 麻醉记录
|
||
resp = api_get("/api/v1/anesthesia/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.5", "麻醉记录列表")
|
||
|
||
# 9.6 麻醉增强
|
||
resp = api_get("/anesthesia-enhanced/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.6", "麻醉增强列表")
|
||
|
||
# 9.7 麻醉质控
|
||
resp = api_get("/anesthesia-quality-control/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "手术", "9.7", "麻醉质控列表")
|
||
|
||
# ============================
|
||
# 测试模块10: 院感管理
|
||
# ============================
|
||
def test_infection():
|
||
print("\n" + "=" * 50)
|
||
print("模块10: 院感管理")
|
||
print("=" * 50)
|
||
|
||
modules = [
|
||
("10.1", "院感监测", "/infection-enhanced/surveillance/page"),
|
||
("10.2", "院感预警", "/infection-enhanced/warning/page"),
|
||
("10.3", "耐药监测", "/infection-enhanced/resistance/page"),
|
||
("10.4", "职业暴露", "/infection-enhanced/exposure/page"),
|
||
("10.5", "手卫生", "/infection-enhanced/hand-hygiene/page"),
|
||
("10.6", "环境监测", "/infection-enhanced/environment/page"),
|
||
]
|
||
for case_id, name, path in modules:
|
||
resp = api_get(path, {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "院感", case_id, name)
|
||
|
||
# ============================
|
||
# 测试模块11: 质量管理
|
||
# ============================
|
||
def test_quality():
|
||
print("\n" + "=" * 50)
|
||
print("模块11: 质量管理")
|
||
print("=" * 50)
|
||
|
||
# 11.1 运行质控
|
||
resp = api_get("/quality-enhanced/runtime/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "质控", "11.1", "运行质控列表")
|
||
|
||
# 11.2 终末质控
|
||
resp = api_get("/api/v1/emr-quality/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "质控", "11.2", "终末质控列表")
|
||
|
||
# 11.3 质量统计
|
||
resp = api_get("/quality-enhanced/statistics/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "质控", "11.3", "质量统计列表")
|
||
|
||
# ============================
|
||
# 测试模块12: 中医管理
|
||
# ============================
|
||
def test_tcm():
|
||
print("\n" + "=" * 50)
|
||
print("模块12: 中医管理")
|
||
print("=" * 50)
|
||
|
||
# 12.1 中医体质列表 - 应包含我们插入的气虚质和阳虚质
|
||
resp = api_get("/api/v1/tcm/constitution/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "中医", "12.1", "中医体质列表", min_rows=1)
|
||
|
||
# 12.2 中医方剂列表 - 应包含四君子汤、六味地黄丸、小柴胡汤
|
||
resp = api_get("/api/v1/tcm/prescriptions", {"pageNum": 1, "pageSize": 10})
|
||
data = resp
|
||
rows = data.get("rows", data.get("data", []))
|
||
passed = len(rows) >= 3
|
||
detail = "" if passed else f"方剂数量不足: {len(rows)}, 预期>=3"
|
||
stats.record("中医", "12.2", "中医方剂列表-至少3个方剂", passed, detail)
|
||
|
||
# 12.3 中医统计
|
||
resp = api_get("/api/v1/tcm/statistics")
|
||
assert_response(resp, "中医", "12.3", "中医统计查询",
|
||
expected_code=200)
|
||
|
||
# ============================
|
||
# 测试模块13: 会诊管理
|
||
# ============================
|
||
def test_consultation():
|
||
print("\n" + "=" * 50)
|
||
print("模块13: 会诊管理")
|
||
print("=" * 50)
|
||
|
||
# 13.1 会诊记录
|
||
resp = api_get("/consultation/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "会诊", "13.1", "会诊记录列表")
|
||
|
||
# 13.2 会诊反馈
|
||
resp = api_get("/cross-module/consult-feedback/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "会诊", "13.2", "会诊反馈列表")
|
||
|
||
# 13.3 会诊超时
|
||
resp = api_get("/cross-module/consulttimeout/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "会诊", "13.3", "会诊超时列表")
|
||
|
||
# ============================
|
||
# 测试模块14: 临床路径
|
||
# ============================
|
||
def test_pathway():
|
||
print("\n" + "=" * 50)
|
||
print("模块14: 临床路径")
|
||
print("=" * 50)
|
||
|
||
# 14.1 临床路径列表 - 应包含社区获得性肺炎、急性阑尾炎、2型糖尿病
|
||
resp = api_get("/clinical-pathway/page", {"pageNum": 1, "pageSize": 10})
|
||
data = resp
|
||
rows = data.get("rows", data.get("data", []))
|
||
passed = len(rows) >= 3
|
||
detail = "" if passed else f"路径数量不足: {len(rows)}, 预期>=3"
|
||
stats.record("路径", "14.1", "临床路径列表-至少3条路径", passed, detail)
|
||
|
||
# ============================
|
||
# 测试模块15: 危急值管理
|
||
# ============================
|
||
def test_critical():
|
||
print("\n" + "=" * 50)
|
||
print("模块15: 危急值管理")
|
||
print("=" * 50)
|
||
|
||
# 15.1 危急值列表 - 应包含血钾6.8和血红蛋白52
|
||
resp = api_get("/api/v1/critical-value/page", {"pageNum": 1, "pageSize": 10})
|
||
data = resp
|
||
rows = data.get("rows", data.get("data", []))
|
||
# 检查是否有危急值记录
|
||
passed = isinstance(rows, list)
|
||
detail = "" if passed else "危急值列表返回格式异常"
|
||
stats.record("危急值", "15.1", "危急值列表查询", passed, detail)
|
||
|
||
# ============================
|
||
# 测试模块16: 处方点评
|
||
# ============================
|
||
def test_review():
|
||
print("\n" + "=" * 50)
|
||
print("模块16: 处方点评")
|
||
print("=" * 50)
|
||
|
||
# 16.1 点评计划
|
||
resp = api_get("/api/v1/review/plans", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "点评", "16.1", "点评计划列表")
|
||
|
||
# 16.2 点评记录
|
||
resp = api_get("/api/v1/review/records", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "点评", "16.2", "点评记录列表")
|
||
|
||
# 16.3 点评统计
|
||
resp = api_get("/api/v1/review/statistics")
|
||
assert_response(resp, "点评", "16.3", "点评统计查询",
|
||
expected_code=200)
|
||
|
||
# ============================
|
||
# 测试模块17: 合理用药
|
||
# ============================
|
||
def test_rational_drug():
|
||
print("\n" + "=" * 50)
|
||
print("模块17: 合理用药")
|
||
print("=" * 50)
|
||
|
||
# 17.1 合理用药列表
|
||
resp = api_get("/api/v1/rational-drug/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "用药", "17.1", "合理用药列表")
|
||
|
||
# 17.2 相互作用
|
||
resp = api_get("/api/v1/rational-drug/interaction/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "用药", "17.2", "相互作用列表")
|
||
|
||
# 17.3 用药统计
|
||
resp = api_get("/api/v1/rational-drug/statistics")
|
||
assert_response(resp, "用药", "17.3", "用药统计查询",
|
||
expected_code=200)
|
||
|
||
# 17.4 审计日志
|
||
resp = api_get("/api/v1/rational-drug/audit-log", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "用药", "17.4", "审计日志列表")
|
||
|
||
# ============================
|
||
# 测试模块18: 药品追溯
|
||
# ============================
|
||
def test_drug_trace():
|
||
print("\n" + "=" * 50)
|
||
print("模块18: 药品追溯")
|
||
print("=" * 50)
|
||
|
||
# 18.1 药品追溯列表
|
||
resp = api_get("/drugtrace/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "追溯", "18.1", "药品追溯列表")
|
||
|
||
# ============================
|
||
# 测试模块19: EMPI主索引
|
||
# ============================
|
||
def test_empi():
|
||
print("\n" + "=" * 50)
|
||
print("模块19: EMPI主索引")
|
||
print("=" * 50)
|
||
|
||
# 19.1 EMPI索引列表
|
||
resp = api_get("/api/v1/empi/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "EMPI", "19.1", "EMPI索引列表")
|
||
|
||
# ============================
|
||
# 测试模块20: ESB数据集成
|
||
# ============================
|
||
def test_esb():
|
||
print("\n" + "=" * 50)
|
||
print("模块20: ESB数据集成")
|
||
print("=" * 50)
|
||
|
||
# 20.1 ESB消息监控
|
||
resp = api_get("/esbmanage/message/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "ESB", "20.1", "ESB消息列表")
|
||
|
||
# 20.2 ESB服务注册
|
||
resp = api_get("/esbmanage/registry/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "ESB", "20.2", "ESB服务注册列表")
|
||
|
||
# ============================
|
||
# 测试模块21: 电子签名
|
||
# ============================
|
||
def test_ca():
|
||
print("\n" + "=" * 50)
|
||
print("模块21: 电子签名")
|
||
print("=" * 50)
|
||
|
||
# 21.1 CA签名列表
|
||
resp = api_get("/api/v1/ca-signature/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "CA", "21.1", "CA签名列表")
|
||
|
||
# 21.2 CA签名统计
|
||
resp = api_get("/api/v1/ca-signature/statistics")
|
||
assert_response(resp, "CA", "21.2", "CA签名统计",
|
||
expected_code=200)
|
||
|
||
# ============================
|
||
# 测试模块22: 病案管理
|
||
# ============================
|
||
def test_mr():
|
||
print("\n" + "=" * 50)
|
||
print("模块22: 病案管理")
|
||
print("=" * 50)
|
||
|
||
# 22.1 病案首页
|
||
resp = api_get("/api/v1/mr-homepage/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "病案", "22.1", "病案首页列表")
|
||
|
||
# 22.2 病案质量检查
|
||
resp = api_get("/api/v1/mr-homepage/quality-check/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "病案", "22.2", "病案质量检查")
|
||
|
||
# ============================
|
||
# 测试模块23: 随访管理
|
||
# ============================
|
||
def test_followup():
|
||
print("\n" + "=" * 50)
|
||
print("模块23: 随访管理")
|
||
print("=" * 50)
|
||
|
||
# 23.1 随访计划
|
||
resp = api_get("/followup/plan/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "随访", "23.1", "随访计划列表")
|
||
|
||
# ============================
|
||
# 测试模块24: 知情同意
|
||
# ============================
|
||
def test_consent():
|
||
print("\n" + "=" * 50)
|
||
print("模块24: 知情同意")
|
||
print("=" * 50)
|
||
|
||
# 24.1 知情同意列表
|
||
resp = api_get("/api/v1/informed-consent/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "知情", "24.1", "知情同意列表")
|
||
|
||
# ============================
|
||
# 测试模块25: 消毒供应
|
||
# ============================
|
||
def test_cssd():
|
||
print("\n" + "=" * 50)
|
||
print("模块25: 消毒供应")
|
||
print("=" * 50)
|
||
|
||
# 25.1 消毒追溯
|
||
resp = api_get("/cssd/trace/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "CSSD", "25.1", "消毒追溯列表")
|
||
|
||
# ============================
|
||
# 测试模块26: 急诊管理
|
||
# ============================
|
||
def test_emergency():
|
||
print("\n" + "=" * 50)
|
||
print("模块26: 急诊管理")
|
||
print("=" * 50)
|
||
|
||
# 26.1 急诊记录
|
||
resp = api_get("/emergency/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "急诊", "26.1", "急诊记录列表")
|
||
|
||
# 26.2 分诊排队
|
||
resp = api_get("/triage/queue/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "急诊", "26.2", "分诊排队列表")
|
||
|
||
# ============================
|
||
# 测试模块27: 医保管理
|
||
# ============================
|
||
def test_insurance():
|
||
print("\n" + "=" * 50)
|
||
print("模块27: 医保管理")
|
||
print("=" * 50)
|
||
|
||
# 27.1 医保目录
|
||
resp = api_get("/ybmanage/catalog/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "医保", "27.1", "医保目录列表")
|
||
|
||
# ============================
|
||
# 测试模块28: 抗菌药物
|
||
# ============================
|
||
def test_antibiotic():
|
||
print("\n" + "=" * 50)
|
||
print("模块28: 抗菌药物")
|
||
print("=" * 50)
|
||
|
||
# 28.1 抗菌药物列表
|
||
resp = api_get("/api/v1/antibiotic/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "抗菌", "28.1", "抗菌药物列表")
|
||
|
||
# ============================
|
||
# 测试模块29: DRG分析
|
||
# ============================
|
||
def test_drg():
|
||
print("\n" + "=" * 50)
|
||
print("模块29: DRG分析")
|
||
print("=" * 50)
|
||
|
||
# 29.1 DRG分析
|
||
resp = api_get("/api/v1/mr-homepage/drg/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "DRG", "29.1", "DRG分析列表")
|
||
|
||
# 29.2 DRG预警
|
||
resp = api_get("/cross-module/enhanced-drg-alert/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "DRG", "29.2", "DRG预警列表")
|
||
|
||
# ============================
|
||
# 测试模块30: 经营分析
|
||
# ============================
|
||
def test_analytics():
|
||
print("\n" + "=" * 50)
|
||
print("模块30: 经营分析")
|
||
print("=" * 50)
|
||
|
||
# 30.1 经营分析
|
||
resp = api_get("/business-analytics/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "经营", "30.1", "经营分析列表")
|
||
|
||
# ============================
|
||
# 测试模块31: 系统管理
|
||
# ============================
|
||
def test_system():
|
||
print("\n" + "=" * 50)
|
||
print("模块31: 系统管理")
|
||
print("=" * 50)
|
||
|
||
# 31.1 字典类型
|
||
resp = api_get("/dict/type/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "系统", "31.1", "字典类型列表")
|
||
|
||
# 31.2 用户管理
|
||
resp = api_get("/system/user/page", {"pageNum": 1, "pageSize": 10})
|
||
data = resp
|
||
rows = data.get("rows", data.get("data", []))
|
||
passed = isinstance(rows, list) and len(rows) > 0
|
||
detail = "" if passed else "用户列表为空，系统用户数据异常"
|
||
stats.record("系统", "31.2", "用户管理-列表非空", passed, detail)
|
||
|
||
# 31.3 角色管理
|
||
resp = api_get("/system/role/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "系统", "31.3", "角色管理列表")
|
||
|
||
# 31.4 菜单管理
|
||
resp = api_get("/system/menu/list")
|
||
data = resp
|
||
rows = data.get("data", [])
|
||
passed = isinstance(rows, list) and len(rows) > 50
|
||
detail = "" if passed else f"菜单数量异常: {len(rows) if isinstance(rows, list) else 'N/A'}, 预期>50"
|
||
stats.record("系统", "31.4", "菜单管理-菜单数量>50", passed, detail)
|
||
|
||
# 31.5 部门管理
|
||
resp = api_get("/system/dept/list")
|
||
data = resp
|
||
rows = data.get("data", [])
|
||
passed = isinstance(rows, list) and len(rows) > 5
|
||
detail = "" if passed else f"部门数量异常: {len(rows) if isinstance(rows, list) else 'N/A'}, 预期>5"
|
||
stats.record("系统", "31.5", "部门管理-部门数量>5", passed, detail)
|
||
|
||
# 31.6 岗位管理
|
||
resp = api_get("/system/post/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "系统", "31.6", "岗位管理列表")
|
||
|
||
# 31.7 通知管理
|
||
resp = api_get("/system/notice/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "系统", "31.7", "通知管理列表")
|
||
|
||
# 31.8 审计日志
|
||
resp = api_get("/audit-log/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "系统", "31.8", "审计日志列表")
|
||
|
||
# 31.9 仪表盘数据
|
||
resp = api_get("/dashboard/data")
|
||
assert_response(resp, "系统", "31.9", "仪表盘数据",
|
||
expected_code=200)
|
||
|
||
# ============================
|
||
# 测试模块32: 药房管理
|
||
# ============================
|
||
def test_pharmacy():
|
||
print("\n" + "=" * 50)
|
||
print("模块32: 药房管理")
|
||
print("=" * 50)
|
||
|
||
# 32.1 库存预警
|
||
resp = api_get("/pharmacy-stock-alert/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "药房", "32.1", "库存预警列表")
|
||
|
||
# 32.2 西药发药
|
||
resp = api_get("/pharmacy-manage/western-medicine-dispense/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "药房", "32.2", "西药发药列表")
|
||
|
||
# 32.3 退药管理
|
||
resp = api_get("/pharmacy-manage/return-medicine/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "药房", "32.3", "退药管理列表")
|
||
|
||
# 32.4 药品详情
|
||
resp = api_get("/pharmacy-manage/medication-details/page", {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "药房", "32.4", "药品详情列表")
|
||
|
||
# ============================
|
||
# 测试模块33: 报表管理
|
||
# ============================
|
||
def test_reports():
|
||
print("\n" + "=" * 50)
|
||
print("模块33: 报表管理")
|
||
print("=" * 50)
|
||
|
||
reports = [
|
||
("33.1", "挂号报表", "/report-manage/register/page"),
|
||
("33.2", "收费报表", "/report-manage/charge/page"),
|
||
("33.3", "住院首页采集", "/medicalRecordHomePage-manage/collection/page"),
|
||
("33.4", "经营统计", "/report-manage/report-statistics/page"),
|
||
]
|
||
for case_id, name, path in reports:
|
||
resp = api_get(path, {"pageNum": 1, "pageSize": 10})
|
||
assert_page_response(resp, "报表", case_id, name)
|
||
|
||
# ============================
|
||
# 主入口
|
||
# ============================
|
||
if __name__ == "__main__":
|
||
print("=" * 70)
|
||
print("HealthLink-HIS 三甲医院全流程业务逻辑测试")
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"测试环境: {BASE_URL}")
|
||
print("=" * 70)
|
||
|
||
# 登录
|
||
print("\n>>> 登录系统...")
|
||
TOKEN = login()
|
||
if not TOKEN:
|
||
print("❌ 登录失败，无法继续测试！")
|
||
sys.exit(1)
|
||
print(f"✅ 登录成功")
|
||
|
||
# 执行所有测试模块
|
||
test_modules = [
|
||
test_auth,
|
||
test_registration,
|
||
test_doctor_station,
|
||
test_charge,
|
||
test_inpatient,
|
||
test_nursing,
|
||
test_inspection,
|
||
test_radiology,
|
||
test_surgery,
|
||
test_infection,
|
||
test_quality,
|
||
test_tcm,
|
||
test_consultation,
|
||
test_pathway,
|
||
test_critical,
|
||
test_review,
|
||
test_rational_drug,
|
||
test_drug_trace,
|
||
test_empi,
|
||
test_esb,
|
||
test_ca,
|
||
test_mr,
|
||
test_followup,
|
||
test_consent,
|
||
test_cssd,
|
||
test_emergency,
|
||
test_insurance,
|
||
test_antibiotic,
|
||
test_drg,
|
||
test_analytics,
|
||
test_system,
|
||
test_pharmacy,
|
||
test_reports,
|
||
]
|
||
|
||
for test_func in test_modules:
|
||
try:
|
||
test_func()
|
||
except Exception as e:
|
||
print(f" ❌ 模块执行异常: {test_func.__name__}: {e}")
|
||
stats.record("异常", test_func.__name__, "模块执行异常", False, str(e))
|
||
|
||
# 输出汇总
|
||
success = stats.summary()
|
||
|
||
# 生成报告文件
|
||
report_path = f"MD/test/reports/business_logic_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
|
||
import os
|
||
os.makedirs("MD/test/reports", exist_ok=True)
|
||
with open(report_path, "w", encoding="utf-8") as f:
|
||
f.write("# HealthLink-HIS 业务逻辑测试报告\n\n")
|
||
f.write(f"**测试时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
|
||
f.write(f"**测试环境**: {BASE_URL}\n\n")
|
||
f.write("## 测试汇总\n\n")
|
||
f.write(f"- 总测试数: {stats.total}\n")
|
||
f.write(f"- 通过数: {stats.passed}\n")
|
||
f.write(f"- 失败数: {stats.failed}\n")
|
||
f.write(f"- 通过率: {stats.passed*100/stats.total:.1f}%\n\n" if stats.total > 0 else "")
|
||
f.write("## 详细结果\n\n")
|
||
f.write("| 模块 | 编号 | 测试项 | 状态 | 说明 |\n")
|
||
f.write("|------|------|--------|------|------|\n")
|
||
for r in stats.results:
|
||
f.write(f"| {r['module']} | {r['case_id']} | {r['name']} | {r['status']} | {r['detail']} |\n")
|
||
|
||
print(f"\n📄 测试报告已生成: {report_path}")
|
||
|
||
sys.exit(0 if success else 1)
|