Files
base/scripts/api_test_patient.py
iven 3772afd987 chore: 干净 ERP 基座 — 删除 health/ai/wechat 业务代码
删除内容:
- 前端: health/(67文件), ai/(2文件), Copilot, MediaPicker, 相关API/Store/Hook
- 后端: wechat_handler, wechat_service, wechat_user entity, analytics handler, ai_workflow_seed
- 配置: WechatConfig, AppConfig.wechat, AuthState wechat 字段
- 启动: 微信凭据检查块, ensure_ai_workflows() 调用
- 迁移: 新增 m20260613_000170_drop_wechat_users.rs
- 脚本: api_test_health_alert.py, api_test_mp.py, mpsync.sh/ps1
- E2E: health-data page, flows/ 目录

保留: erp-core/auth/workflow/message/config/plugin + 基座前端 + 通用组件
2026-06-13 00:32:50 +08:00

513 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""HMS 患者建档链路端到端 API 测试"""
import json
import sys
import time
import urllib.request
import urllib.error
BASE_URL = "http://localhost:3000/api/v1"
TOKEN = None
RESULTS = []
def log(test_id, name, status, detail):
"""记录测试结果"""
RESULTS.append({"id": test_id, "name": name, "status": status, "detail": detail})
icon = "PASS" if status == "PASS" else ("FAIL" if status == "FAIL" else "WARN")
print(f" [{icon}] {test_id}: {name} -- {detail}")
def api_call(method, path, data=None, token=None, expect_status=None):
"""发送 API 请求"""
url = f"{BASE_URL}{path}"
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
body = json.dumps(data).encode("utf-8") if data else None
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
status = resp.status
resp_data = json.loads(resp.read().decode("utf-8"))
if expect_status and status != expect_status:
return status, resp_data, f"Expected {expect_status}, got {status}"
return status, resp_data, None
except urllib.error.HTTPError as e:
resp_body = e.read().decode("utf-8", errors="replace")
try:
resp_data = json.loads(resp_body)
except:
resp_data = {"raw": resp_body}
if expect_status and e.code == expect_status:
return e.code, resp_data, None
return e.code, resp_data, f"HTTP {e.code}: {resp_body[:200]}"
except Exception as e:
return 0, None, str(e)
# ============================================================
# Step 0: Login
# ============================================================
print("\n" + "="*60)
print("Step 0: 登录获取 Token")
print("="*60)
status, resp, err = api_call("POST", "/auth/login",
{"username": "admin", "password": "Admin@2026"}, expect_status=200)
if err:
# 可能限流,等一下重试
print(f" 首次登录失败: {err}")
print(" 等待 20 秒后重试...")
time.sleep(20)
status, resp, err = api_call("POST", "/auth/login",
{"username": "admin", "password": "Admin@2026"}, expect_status=200)
if err or not resp or not resp.get("success"):
log("T0", "登录", "FAIL", f"登录失败: {err or resp}")
sys.exit(1)
TOKEN = resp["data"]["access_token"]
user = resp["data"]["user"]
log("T0", "登录", "PASS", f"用户: {user['display_name']}, 角色: {[r['name'] for r in user['roles']]}")
# ============================================================
# Test 1.1: Patient List
# ============================================================
print("\n" + "="*60)
print("Test 1.1: 患者列表")
print("="*60)
status, resp, err = api_call("GET", "/health/patients?page=1&page_size=10", token=TOKEN)
if err:
log("T1.1", "患者列表", "FAIL", err)
else:
total = resp.get("data", {}).get("total", 0)
items = resp.get("data", {}).get("items", [])
log("T1.1", "患者列表", "PASS", f"success={resp['success']}, total={total}, items={len(items)}")
if items:
p = items[0]
print(f" 首条: id={p.get('id')}, name={p.get('name')}, gender={p.get('gender')}")
# ============================================================
# Test 1.2: Create Patient (Valid)
# ============================================================
print("\n" + "="*60)
print("Test 1.2: 创建患者 - 完整有效数据")
print("="*60)
import random
_ts = str(int(time.time() * 1000))[-6:]
patient_data = {
"name": f"API测试患者_{_ts}",
"gender": "male",
"birth_date": "1990-05-15",
"phone": f"138{random.randint(10000000, 99999999)}",
"blood_type": "A",
"emergency_contact_name": "紧急联系人",
"emergency_contact_phone": f"139{random.randint(10000000, 99999999)}"
}
patient_id = None # 预定义,防止后续 NameError
status, resp, err = api_call("POST", "/health/patients", patient_data, token=TOKEN)
if err:
log("T1.2", "创建患者(有效)", "FAIL", err)
elif resp and resp.get("success"):
patient = resp["data"]
patient_id = patient.get("id")
log("T1.2", "创建患者(有效)", "PASS",
f"id={patient_id}, name={patient.get('name')}, gender={patient.get('gender')}, version={patient.get('version')}")
print(f" birth_date={patient.get('birth_date')}, phone={patient.get('phone')}")
print(f" blood_type={patient.get('blood_type')}, tenant_id={patient.get('tenant_id')}")
else:
log("T1.2", "创建患者(有效)", "FAIL", f"success={resp.get('success')}, error={resp.get('error')}")
# ============================================================
# Test 1.3: Create Patient - Empty Name (should fail)
# ============================================================
print("\n" + "="*60)
print("Test 1.3: 创建患者 - 空名称(应失败)")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": "", "gender": "male", "birth_date": "1990-01-01"}, token=TOKEN)
if status == 400 or (resp and not resp.get("success")):
log("T1.3", "创建患者(空名称)", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
elif status == 201 or (resp and resp.get("success")):
log("T1.3", "创建患者(空名称)", "FAIL", "空名称被接受,应该被拒绝")
else:
log("T1.3", "创建患者(空名称)", "FAIL", f"status={status}, resp={resp}")
# ============================================================
# Test 1.4: Create Patient - Future Birth Date (should fail)
# ============================================================
print("\n" + "="*60)
print("Test 1.4: 创建患者 - 未来出生日期(应失败)")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": "未来患者", "gender": "male", "birth_date": "2099-01-01"}, token=TOKEN)
if status == 400 or (resp and not resp.get("success")):
log("T1.4", "创建患者(未来日期)", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
elif status == 201 or (resp and resp.get("success")):
log("T1.4", "创建患者(未来日期)", "FAIL", "未来出生日期被接受,应该被拒绝")
else:
log("T1.4", "创建患者(未来日期)", "FAIL", f"status={status}, resp={resp}")
# ============================================================
# Test 2.1: Patient Detail + PII Check
# ============================================================
print("\n" + "="*60)
print("Test 2.1: 患者详情 + PII 脱敏验证")
print("="*60)
if patient_id:
status, resp, err = api_call("GET", f"/health/patients/{patient_id}", token=TOKEN)
if err:
log("T2.1", "患者详情", "FAIL", err)
else:
p = resp.get("data", {})
log("T2.1", "患者详情", "PASS", f"success={resp['success']}, name={p.get('name')}")
# PII 检查: phone 是否为明文或脱敏
phone = p.get("phone", "N/A")
emergency_phone = p.get("emergency_contact_phone", "N/A")
print(f" phone={phone}")
print(f" emergency_contact_phone={emergency_phone}")
print(f" id_card_number={p.get('id_card_number', 'N/A')}")
# 检查标准字段
has_tenant_id = "tenant_id" in p
has_created_at = "created_at" in p
has_version = "version" in p
print(f" tenant_id={'存在' if has_tenant_id else '缺失'}, "
f"created_at={'存在' if has_created_at else '缺失'}, "
f"version={'存在' if has_version else '缺失'}")
if not (has_tenant_id and has_created_at and has_version):
log("T2.1b", "标准字段检查", "WARN", "部分标准字段缺失")
else:
print(f" 标准字段检查: 通过")
else:
log("T2.1", "患者详情", "SKIP", "无 patient_id (创建患者失败)")
# ============================================================
# Test 2.2: Patient Detail - Non-existent ID (should 404)
# ============================================================
print("\n" + "="*60)
print("Test 2.2: 患者详情 - 不存在的ID(应404)")
print("="*60)
status, resp, err = api_call("GET", "/health/patients/00000000-0000-0000-0000-000000000000", token=TOKEN)
if status == 404:
log("T2.2", "患者详情(不存在)", "PASS", f"正确返回 404")
elif err:
log("T2.2", "患者详情(不存在)", "WARN", f"status={status}, err={err}")
else:
log("T2.2", "患者详情(不存在)", "FAIL", f"status={status}, 应为 404")
# ============================================================
# Test 3.1: Patient Tags - Create Tag
# ============================================================
print("\n" + "="*60)
print("Test 3.1: 患者标签 - 创建标签")
print("="*60)
tag_id = None
status, resp, err = api_call("POST", "/health/patient-tags",
{"name": f"API测试标签_{_ts}", "color": "#FF5500"}, token=TOKEN)
if err:
log("T3.1", "创建标签", "FAIL", err)
elif resp and resp.get("success"):
tag = resp["data"]
tag_id = tag.get("id")
log("T3.1", "创建标签", "PASS", f"id={tag_id}, name={tag.get('name')}, color={tag.get('color')}")
else:
log("T3.1", "创建标签", "FAIL", f"status={status}, error={resp}")
# ============================================================
# Test 3.2: Patient Tags - List
# ============================================================
print("\n" + "="*60)
print("Test 3.2: 患者标签 - 列表")
print("="*60)
status, resp, err = api_call("GET", "/health/patient-tags", token=TOKEN)
if err:
log("T3.2", "标签列表", "FAIL", err)
else:
raw_data = resp.get("data")
if isinstance(raw_data, list):
total = len(raw_data)
log("T3.2", "标签列表", "PASS", f"success={resp['success']}, count={total}")
elif isinstance(raw_data, dict):
items = raw_data.get("items", [])
total = raw_data.get("total", len(items))
log("T3.2", "标签列表", "PASS", f"success={resp['success']}, total={total}")
else:
log("T3.2", "标签列表", "PASS", f"success={resp['success']}, data_type={type(raw_data)}")
# ============================================================
# Test 3.3: Patient Tags - Assign to Patient
# ============================================================
print("\n" + "="*60)
print("Test 3.3: 患者标签 - 关联标签到患者")
print("="*60)
if patient_id and tag_id:
# 尝试关联标签 - 可能的 API 路径
status, resp, err = api_call("POST", f"/health/patients/{patient_id}/tags",
{"tag_ids": [tag_id]}, token=TOKEN)
if err and status == 404:
# 尝试替代路径
status2, resp2, err2 = api_call("POST", "/health/patient-tag-relations",
{"patient_id": patient_id, "tag_id": tag_id}, token=TOKEN)
if err2:
log("T3.3", "关联标签", "WARN", f"标签关联路径未确认: {err[:100]}")
else:
log("T3.3", "关联标签", "PASS", f"通过 /patient-tag-relations: {resp2}")
elif err:
log("T3.3", "关联标签", "WARN", f"status={status}, err={err[:150]}")
else:
log("T3.3", "关联标签", "PASS", f"success={resp.get('success')}")
else:
log("T3.3", "关联标签", "SKIP", "缺少 patient_id 或 tag_id")
# ============================================================
# Test 4.1: Patient Update - Normal
# ============================================================
print("\n" + "="*60)
print("Test 4.1: 患者更新 - 正常更新")
print("="*60)
updated_version = None
if patient_id:
# 先获取当前 version
status, resp, err = api_call("GET", f"/health/patients/{patient_id}", token=TOKEN)
if not err and resp.get("success"):
current_version = resp["data"].get("version")
print(f" 当前 version: {current_version}")
update_data = {
"name": "API测试患者-已更新",
"phone": "13800003333",
"version": current_version
}
status, resp, err = api_call("PUT", f"/health/patients/{patient_id}", update_data, token=TOKEN)
if err:
log("T4.1", "患者更新", "FAIL", err)
elif resp and resp.get("success"):
updated = resp["data"]
updated_version = updated.get("version")
log("T4.1", "患者更新", "PASS",
f"name={updated.get('name')}, version={current_version}->{updated_version}")
else:
log("T4.1", "患者更新", "FAIL", f"success={resp.get('success')}, error={resp}")
else:
log("T4.1", "患者更新", "FAIL", f"获取患者失败: {err}")
else:
log("T4.1", "患者更新", "SKIP", "无 patient_id")
# ============================================================
# Test 4.2: Patient Update - Optimistic Lock (should fail)
# ============================================================
print("\n" + "="*60)
print("Test 4.2: 患者更新 - 乐观锁冲突(应失败)")
print("="*60)
if patient_id:
# 使用旧 version 触发乐观锁冲突
stale_update = {
"name": "API测试患者-冲突更新",
"version": 1 # 旧版本号
}
status, resp, err = api_call("PUT", f"/health/patients/{patient_id}", stale_update, token=TOKEN)
if status == 409 or (resp and not resp.get("success")):
log("T4.2", "乐观锁冲突", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
elif resp and resp.get("success"):
log("T4.2", "乐观锁冲突", "FAIL", "旧版本号更新被接受,乐观锁未生效")
else:
log("T4.2", "乐观锁冲突", "WARN", f"status={status}, resp={str(resp)[:200]}")
else:
log("T4.2", "乐观锁冲突", "SKIP", "无 patient_id")
# ============================================================
# Test 5.1: Security - No Auth (should 401)
# ============================================================
print("\n" + "="*60)
print("Test 5.1: 安全测试 - 无认证访问(应401)")
print("="*60)
status, resp, err = api_call("GET", "/health/patients")
if status == 401:
log("T5.1", "无认证访问", "PASS", "正确返回 401")
elif err:
log("T5.1", "无认证访问", "FAIL", f"status={status}, 应为 401")
else:
log("T5.1", "无认证访问", "FAIL", f"status={status}, 成功访问但应该被拒绝")
# ============================================================
# Test 5.2: Security - SQL Injection Attempt
# ============================================================
print("\n" + "="*60)
print("Test 5.2: 安全测试 - SQL 注入尝试")
print("="*60)
status, resp, err = api_call("GET", "/health/patients?search=%27%3B%20DROP%20TABLE%20patients%3B%20--", token=TOKEN)
if status == 500:
log("T5.2", "SQL注入防护", "FAIL", "服务器返回 500可能存在注入风险")
elif resp and resp.get("success"):
# 正常返回搜索结果 = 注入被参数化查询防住了
log("T5.2", "SQL注入防护", "PASS", f"注入被参数化查询拦截,正常返回数据")
else:
log("T5.2", "SQL注入防护", "PASS", f"status={status}, 注入未导致服务异常")
# ============================================================
# Test 5.3: Security - Invalid Token (should 401)
# ============================================================
print("\n" + "="*60)
print("Test 5.3: 安全测试 - 无效 Token(应401)")
print("="*60)
status, resp, err = api_call("GET", "/health/patients", token="invalid_token_here")
if status == 401:
log("T5.3", "无效Token", "PASS", "正确返回 401")
else:
log("T5.3", "无效Token", "FAIL", f"status={status}, 应为 401")
# ============================================================
# Test 6.1: Pagination
# ============================================================
print("\n" + "="*60)
print("Test 6.1: 分页查询")
print("="*60)
status, resp, err = api_call("GET", "/health/patients?page=1&page_size=2", token=TOKEN)
if err:
log("T6.1", "分页查询", "FAIL", err)
else:
total = resp.get("data", {}).get("total", 0)
items = resp.get("data", {}).get("items", [])
page = resp.get("data", {}).get("page", "N/A")
page_size = resp.get("data", {}).get("page_size", resp.get("data", {}).get("per_page", "N/A"))
log("T6.1", "分页查询", "PASS",
f"total={total}, page={page}, page_size={page_size}, returned={len(items)}")
# ============================================================
# Test 7.1: Create patient with minimal data
# ============================================================
print("\n" + "="*60)
print("Test 7.1: 创建患者 - 最小必填数据")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": f"最小数据患者_{_ts}", "gender": "female", "birth_date": "2000-01-01"}, token=TOKEN)
if err:
log("T7.1", "创建患者(最小数据)", "FAIL", err)
elif resp and resp.get("success"):
p = resp["data"]
log("T7.1", "创建患者(最小数据)", "PASS",
f"id={p.get('id')}, name={p.get('name')}, optional fields: phone={p.get('phone')}, blood_type={p.get('blood_type')}")
else:
log("T7.1", "创建患者(最小数据)", "FAIL", f"success={resp.get('success')}, error={resp}")
# ============================================================
# Test 7.2: Create patient with whitespace-only name (should fail)
# ============================================================
print("\n" + "="*60)
print("Test 7.2: 创建患者 - 纯空格名称(应失败)")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": " ", "gender": "male", "birth_date": "1990-01-01"}, token=TOKEN)
if status == 400 or (resp and not resp.get("success")):
log("T7.2", "创建患者(空格名称)", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
else:
log("T7.2", "创建患者(空格名称)", "FAIL", f"纯空格名称被接受: status={status}")
# ============================================================
# Test 8.1: Invalid gender
# ============================================================
print("\n" + "="*60)
print("Test 8.1: 创建患者 - 无效性别值")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": "无效性别", "gender": "invalid_gender", "birth_date": "1990-01-01"}, token=TOKEN)
if status == 400 or (resp and not resp.get("success")):
log("T8.1", "创建患者(无效性别)", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
elif resp and resp.get("success"):
log("T8.1", "创建患者(无效性别)", "WARN", f"无效性别被接受(可能是开放枚举): gender={resp['data'].get('gender')}")
else:
log("T8.1", "创建患者(无效性别)", "WARN", f"status={status}")
# ============================================================
# Test 9.1: Invalid birth_date format
# ============================================================
print("\n" + "="*60)
print("Test 9.1: 创建患者 - 无效日期格式")
print("="*60)
status, resp, err = api_call("POST", "/health/patients",
{"name": "无效日期", "gender": "male", "birth_date": "not-a-date"}, token=TOKEN)
if status == 400 or (resp and not resp.get("success")):
log("T9.1", "创建患者(无效日期格式)", "PASS",
f"正确拒绝: status={status}, error={resp.get('error', resp.get('message', 'N/A'))}")
else:
log("T9.1", "创建患者(无效日期格式)", "FAIL", f"无效日期格式被接受: status={status}")
# ============================================================
# Summary
# ============================================================
print("\n" + "="*60)
print("测试汇总")
print("="*60)
passed = sum(1 for r in RESULTS if r["status"] == "PASS")
failed = sum(1 for r in RESULTS if r["status"] == "FAIL")
warned = sum(1 for r in RESULTS if r["status"] == "WARN")
skipped = sum(1 for r in RESULTS if r["status"] == "SKIP")
total = len(RESULTS)
print(f"\n 总计: {total} 项测试")
print(f" PASS: {passed}")
print(f" FAIL: {failed}")
print(f" WARN: {warned}")
print(f" SKIP: {skipped}")
print(f" 通过率: {passed/total*100:.1f}%")
if failed > 0:
print("\n 失败项:")
for r in RESULTS:
if r["status"] == "FAIL":
print(f" [{r['id']}] {r['name']}: {r['detail']}")
if warned > 0:
print("\n 警告项:")
for r in RESULTS:
if r["status"] == "WARN":
print(f" [{r['id']}] {r['name']}: {r['detail']}")
print("\n" + "="*60)
if failed == 0:
print("结论: 所有测试通过")
elif failed <= 2:
print(f"结论: 基本通过,有 {failed} 项失败需要关注")
else:
print(f"结论: 有 {failed} 项失败,需要修复")
print("="*60)