d623f8b2ff
修复项: - fix(db): 迁移 149 — 修复 Admin 角色权限绑定被迁移链破坏 (FE-C1) - fix(health): 4 个 handler 添加空名称验证 — Doctor/Article/AlertRule/Tag (API-C1~C4) - fix(health): Stats 仪表盘 new_this_week 查询修复 — SeaORM date_trunc bug (FE-C2) - fix(server): 添加安全响应头 — X-Frame-Options/CSP/XSS-Protection/Referrer-Policy (SEC-H1) - fix(mp): 预约创建契约修复 — notes/reason 字段映射 + 移除 schedule_id (MP-H1) - fix(mp): 咨询会话 subject/last_message 字段改为可选 (MP-H3) - fix(ai): AiConfig Default derive 替代手写 impl (clippy) 测试报告: - 8 维度端到端测试全部完成 (后端 87 用例 / 前端 30 页面 / 小程序 80+ API / 安全 20 项 / 性能 20 端点) - 多角色 7 角色 49 检查 100% 通过 - 综合测试报告 + 专家评估报告
794 lines
35 KiB
Python
794 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""Multi-role scenario API test runner for HMS health management platform."""
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
BASE = "http://localhost:3000/api/v1"
|
|
RESULTS = []
|
|
|
|
def api(method, path, token=None, body=None):
|
|
"""Make API call and return (status_code, response_dict)."""
|
|
url = f"{BASE}{path}"
|
|
headers = {"Content-Type": "application/json"}
|
|
if token:
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
data = json.dumps(body).encode() if body else None
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
raw = resp.read().decode()
|
|
return resp.status, json.loads(raw) if raw else {}
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode() if e.fp else ""
|
|
try:
|
|
return e.code, json.loads(raw)
|
|
except:
|
|
return e.code, {"error": raw}
|
|
except Exception as ex:
|
|
return 0, {"error": str(ex)}
|
|
|
|
def login(username, password="Admin@2026"):
|
|
status, resp = api("POST", "/auth/login", body={"username": username, "password": password})
|
|
if resp.get("success"):
|
|
return resp["data"]["access_token"]
|
|
print(f" LOGIN FAILED for {username}: {resp}")
|
|
return None
|
|
|
|
def record(role, chain, test_id, description, passed, detail=""):
|
|
RESULTS.append({
|
|
"role": role, "chain": chain, "test_id": test_id,
|
|
"description": description, "passed": passed, "detail": detail
|
|
})
|
|
status = "PASS" if passed else "FAIL"
|
|
print(f" [{status}] {role}-{chain}.{test_id}: {description}" + (f" -- {detail}" if detail and not passed else ""))
|
|
|
|
def check_api(role, chain, test_id, desc, method, path, token, expected_success=True, body=None):
|
|
status, resp = api(method, path, token, body)
|
|
# Handle non-dict responses (e.g. raw lists)
|
|
if not isinstance(resp, dict):
|
|
ok = status == 200
|
|
record(role, chain, test_id, desc, ok, f"status={status}" if not ok else "")
|
|
return resp if ok else None
|
|
ok = resp.get("success", False) == expected_success if isinstance(expected_success, bool) else True
|
|
detail = ""
|
|
if not ok:
|
|
detail = f"status={status}, msg={resp.get('message', resp.get('error', ''))}"
|
|
record(role, chain, test_id, desc, ok, detail)
|
|
return resp if ok else None
|
|
|
|
def check_api_status(role, chain, test_id, desc, method, path, token, expected_status=200, body=None):
|
|
status, resp = api(method, path, token, body)
|
|
ok = status == expected_status
|
|
detail = ""
|
|
if not ok:
|
|
detail = f"expected={expected_status}, got={status}, msg={resp.get('message', resp.get('error', ''))}"
|
|
record(role, chain, test_id, desc, ok, detail)
|
|
return resp if ok else None
|
|
|
|
def main():
|
|
print(f"\n{'='*60}")
|
|
print(f" HMS Multi-Role Scenario API Testing")
|
|
print(f" Started: {datetime.now().isoformat()}")
|
|
print(f"{'='*60}\n")
|
|
|
|
# Login all roles
|
|
print("Logging in all 5 roles...")
|
|
time.sleep(1)
|
|
tokens = {}
|
|
for name, user in [("admin", "admin"), ("doctor", "doctor_test"), ("nurse", "nurse_test"),
|
|
("operator", "operator_test"), ("hm", "health_manager_test")]:
|
|
tokens[name] = login(user)
|
|
time.sleep(1)
|
|
if not tokens[name]:
|
|
print(f"FATAL: Cannot login as {user}")
|
|
sys.exit(1)
|
|
print("All 5 roles logged in successfully.\n")
|
|
|
|
AT = tokens["admin"]
|
|
DT = tokens["doctor"]
|
|
NT = tokens["nurse"]
|
|
OT = tokens["operator"]
|
|
HT = tokens["hm"]
|
|
|
|
# ========================================
|
|
# R01 - ADMIN (9 chains)
|
|
# ========================================
|
|
print("=" * 60)
|
|
print("R01 - ADMIN BUSINESS CHAINS")
|
|
print("=" * 60)
|
|
|
|
# Chain A: Patient creation full chain
|
|
print("\n--- Chain A: Patient Creation ---")
|
|
resp = check_api("R01", "A", "1", "Create patient", "POST", "/health/patients", AT,
|
|
body={"name": "MultiRoleTestPatient", "gender": "male", "phone": "13900001111",
|
|
"birth_date": "1990-01-01"})
|
|
patient_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.5)
|
|
|
|
check_api("R01", "A", "2", "Patient detail", "GET", f"/health/patients/{patient_id}", AT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R01", "A", "3", "Create tag", "POST", "/health/patient-tags", AT,
|
|
body={"name": "HighBP-Risk-Test", "color": "#FF0000"})
|
|
tag_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "A", "3b", "Assign tag to patient", "POST", f"/health/patients/{patient_id}/tags", AT,
|
|
body={"tag_ids": [tag_id]} if tag_id else None)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "A", "4", "Devices list", "GET", "/health/devices?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "A", "5", "Consents list (via patient)", "GET", f"/health/patients/{patient_id}/consents?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "A", "6", "Search patient", "GET", "/health/patients?search=MultiRoleTest", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain B: Follow-up closed loop
|
|
print("\n--- Chain B: Follow-up Closed Loop ---")
|
|
resp = check_api("R01", "B", "1", "Create follow-up task", "POST", "/health/follow-up-tasks", AT,
|
|
body={"patient_id": patient_id, "follow_up_type": "phone",
|
|
"planned_date": "2026-05-20", "notes": "Admin test FU"} if patient_id else None)
|
|
fu_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "B", "2", "Follow-up list (pending)", "GET", "/health/follow-up-tasks?status=pending&page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "B", "3", "Follow-up templates", "GET", "/health/follow-up-templates?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "B", "4", "Action inbox", "GET", "/health/action-inbox?page=1&page_size=10", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain C: Consultation flow
|
|
print("\n--- Chain C: Consultation Flow ---")
|
|
check_api("R01", "C", "1", "Consultation sessions list", "GET", "/health/consultation-sessions?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "C", "2", "Doctor dashboard", "GET", "/health/doctor/dashboard", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain D: Alert handling chain
|
|
print("\n--- Chain D: Alert Handling ---")
|
|
check_api("R01", "D", "1", "Critical value thresholds", "GET", "/health/critical-value-thresholds?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "D", "2", "Alerts list", "GET", "/health/alerts?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "D", "3", "Alert rules", "GET", "/health/alert-rules?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "D", "4", "Critical alerts", "GET", "/health/critical-alerts?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "D", "5", "BLE gateways", "GET", "/health/ble-gateways?page=1&page_size=10", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain E: AI analysis chain
|
|
print("\n--- Chain E: AI Analysis ---")
|
|
check_api("R01", "E", "1", "AI prompts", "GET", "/ai/prompts?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "E", "2", "AI suggestions", "GET", "/ai/suggestions?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "E", "3", "AI chat sessions", "GET", "/ai/chat/sessions?page=1&page_size=10", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain F: Content publishing chain
|
|
print("\n--- Chain F: Content Publishing ---")
|
|
resp = check_api("R01", "F", "1", "Create article", "POST", "/health/articles", AT,
|
|
body={"title": "MultiRole Test Article", "content": "Test content", "status": "draft"})
|
|
art_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "F", "2", "Edit article", "PUT", f"/health/articles/{art_id}", AT,
|
|
body={"title": "MultiRole Test Article Updated", "content": "Updated content"} if art_id else None)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "F", "3a", "Submit article", "POST", f"/health/articles/{art_id}/submit", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "F", "3b", "Approve article", "POST", f"/health/articles/{art_id}/approve", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "F", "4", "Unpublish article", "POST", f"/health/articles/{art_id}/unpublish", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain G: Points mall chain
|
|
print("\n--- Chain G: Points Mall ---")
|
|
check_api("R01", "G", "1", "Points rules", "GET", "/health/admin/points/rules?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "G", "2", "Points products", "GET", "/health/admin/points/products?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "G", "3", "Points orders", "GET", "/health/admin/points/orders?page=1&page_size=10", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain H: Offline events chain
|
|
print("\n--- Chain H: Offline Events ---")
|
|
resp = check_api("R01", "H", "1", "Create offline event", "POST", "/health/admin/offline-events", AT,
|
|
body={"title": "MultiRole Test Event", "description": "Test event",
|
|
"event_date": "2026-06-01", "location": "Test Location", "max_participants": 50})
|
|
event_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "H", "2", "List offline events", "GET", "/health/offline-events?page=1&page_size=10", AT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain I: System management chain
|
|
print("\n--- Chain I: System Management ---")
|
|
check_api("R01", "I", "1", "Users list", "GET", "/users?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "2", "Roles list", "GET", "/roles?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "3", "Organizations", "GET", "/organizations", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "4", "Statistics dashboard", "GET", "/health/admin/statistics/dashboard", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "5", "Workflow definitions", "GET", "/workflow/definitions?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "6", "Messages", "GET", "/messages?page=1&page_size=10", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "7", "Settings", "GET", "/config/settings/general", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "8", "Plugins", "GET", "/admin/plugins", AT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R01", "I", "9", "OAuth clients", "GET", "/health/oauth/clients?page=1&page_size=10", AT)
|
|
time.sleep(1)
|
|
|
|
# ========================================
|
|
# R02 - DOCTOR (5 chains + permissions)
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("R02 - DOCTOR BUSINESS CHAINS")
|
|
print("=" * 60)
|
|
|
|
# Chain A: Patient management and clinical workflow
|
|
print("\n--- Chain A: Patient & Clinical ---")
|
|
check_api("R02", "A", "1", "Patient list", "GET", "/health/patients?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "A", "2", "Patient detail", "GET", f"/health/patients/{patient_id}", DT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R02", "A", "3", "Create patient", "POST", "/health/patients", DT,
|
|
body={"name": "DoctorTestPatient", "gender": "female", "phone": "13900002222",
|
|
"birth_date": "1985-03-15"})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "A", "4", "Doctor list", "GET", "/health/doctors?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "A", "5", "Patient diagnoses", "GET", f"/health/patients/{patient_id}/diagnoses?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "A", "6", "Consents list (via patient)", "GET", f"/health/patients/{patient_id}/consents?page=1&page_size=10", DT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain B: Follow-up closed loop (doctor side)
|
|
print("\n--- Chain B: Follow-up (Doctor) ---")
|
|
resp = check_api("R02", "B", "1", "Create follow-up task", "POST", "/health/follow-up-tasks", DT,
|
|
body={"patient_id": patient_id, "follow_up_type": "visit",
|
|
"planned_date": "2026-05-21", "notes": "Doctor created FU"} if patient_id else None)
|
|
dr_fu_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "B", "2", "Follow-up list", "GET", "/health/follow-up-tasks?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "B", "3", "Follow-up templates", "GET", "/health/follow-up-templates?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "B", "4", "Action inbox", "GET", "/health/action-inbox?page=1&page_size=10", DT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain C: Consultation intake closed loop
|
|
print("\n--- Chain C: Consultation Intake ---")
|
|
check_api("R02", "C", "1", "Consultation sessions list", "GET", "/health/consultation-sessions?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "C", "2", "Consultation messages", "GET", "/health/consultation-messages?page=1&page_size=10", DT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain D: Alert handling
|
|
print("\n--- Chain D: Alerts ---")
|
|
check_api("R02", "D", "1", "Alerts list", "GET", "/health/alerts?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "D", "2", "Alert rules", "GET", "/health/alert-rules?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "D", "3", "Critical alerts", "GET", "/health/critical-alerts?page=1&page_size=10", DT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain E: AI analysis
|
|
print("\n--- Chain E: AI Analysis ---")
|
|
check_api("R02", "E", "1", "AI prompts", "GET", "/ai/prompts?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "E", "2", "AI suggestions", "GET", "/ai/suggestions?page=1&page_size=10", DT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R02", "E", "3", "Messages", "GET", "/messages?page=1&page_size=10", DT)
|
|
time.sleep(0.5)
|
|
|
|
# Doctor permission boundary checks
|
|
print("\n--- Doctor Permission Boundaries ---")
|
|
check_api_status("R02", "P", "1", "No user management", "GET", "/users?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "2", "No role management", "GET", "/roles?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "3", "No points rules", "GET", "/health/admin/points/rules?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "4", "No article management", "GET", "/health/articles?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "5", "No system settings", "GET", "/settings", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "6", "No BLE gateways", "GET", "/health/ble-gateways?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R02", "P", "7", "No tag management", "GET", "/health/patient-tags?page=1&page_size=1", DT, expected_status=403)
|
|
time.sleep(1)
|
|
|
|
# ========================================
|
|
# R03 - NURSE (6 chains + permissions)
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("R03 - NURSE BUSINESS CHAINS")
|
|
print("=" * 60)
|
|
|
|
# Chain A: Patient management
|
|
print("\n--- Chain A: Patient Management ---")
|
|
check_api("R03", "A", "1", "Patient list", "GET", "/health/patients?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "A", "2", "Patient detail", "GET", f"/health/patients/{patient_id}", NT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R03", "A", "3", "Create patient", "POST", "/health/patients", NT,
|
|
body={"name": "NurseTestPatient", "gender": "male", "phone": "13900003333",
|
|
"birth_date": "1995-07-20"})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "A", "4", "Daily monitoring", "GET", "/health/daily-monitoring?page=1&page_size=10", NT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain B: Follow-up execution
|
|
print("\n--- Chain B: Follow-up Execution ---")
|
|
check_api("R03", "B", "1", "Follow-up tasks list", "GET", "/health/follow-up-tasks?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R03", "B", "2", "Create follow-up task", "POST", "/health/follow-up-tasks", NT,
|
|
body={"patient_id": patient_id, "follow_up_type": "phone",
|
|
"planned_date": "2026-05-22", "notes": "Nurse FU"} if patient_id else None)
|
|
time.sleep(0.3)
|
|
|
|
# Try to update a follow-up task status
|
|
if fu_id:
|
|
check_api("R03", "B", "3", "Update follow-up task", "PUT", f"/health/follow-up-tasks/{fu_id}", NT,
|
|
body={"status": "in_progress"})
|
|
else:
|
|
record("R03", "B", "3", "Update follow-up task", False, "No follow-up ID available")
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "B", "4", "Follow-up records", "GET", "/health/follow-up-records?page=1&page_size=10", NT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain C: Consultation viewing (read-only)
|
|
print("\n--- Chain C: Consultation Viewing ---")
|
|
check_api("R03", "C", "1", "Consultation sessions (read)", "GET", "/health/consultation-sessions?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "C", "2", "Consultation messages (read)", "GET", "/health/consultation-sessions?page=1&page_size=10", NT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain D: Alert handling
|
|
print("\n--- Chain D: Alerts ---")
|
|
check_api("R03", "D", "1", "Alerts list", "GET", "/health/alerts?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "D", "2", "Alert detail (if any)", "GET", "/health/alerts?page=1&page_size=1", NT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain E: Diagnosis and informed consent
|
|
print("\n--- Chain E: Diagnosis & Consent ---")
|
|
check_api("R03", "E", "1", "Consents list (via patient)", "GET", f"/health/patients/{patient_id}/consents?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "E", "2", "Create consent", "POST", "/health/consents", NT,
|
|
body={"patient_id": patient_id, "consent_type": "general",
|
|
"consent_scope": "data_collection", "status": "signed"} if patient_id else None)
|
|
time.sleep(0.5)
|
|
|
|
# Chain F: Action inbox
|
|
print("\n--- Chain F: Action Inbox ---")
|
|
check_api("R03", "F", "1", "Action inbox list", "GET", "/health/action-inbox?page=1&page_size=10", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "F", "2", "Action inbox stats", "GET", "/health/action-inbox/stats", NT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R03", "8.1", "", "Messages", "GET", "/messages?page=1&page_size=10", NT)
|
|
time.sleep(0.5)
|
|
|
|
# Nurse permission boundary checks
|
|
print("\n--- Nurse Permission Boundaries ---")
|
|
check_api_status("R03", "P", "1", "No doctor management", "GET", "/health/doctors?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "2", "No tag management", "GET", "/health/patient-tags?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "3", "No points rules", "GET", "/health/admin/points/rules?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "4", "No article management", "GET", "/health/articles?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "5", "No AI analysis", "GET", "/ai/prompts?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "6", "No follow-up templates", "GET", "/health/follow-up-templates?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R03", "P", "7", "No user management", "GET", "/users?page=1&page_size=1", NT, expected_status=403)
|
|
time.sleep(1)
|
|
|
|
# ========================================
|
|
# R04 - HEALTH MANAGER (6 chains + permissions)
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("R04 - HEALTH MANAGER BUSINESS CHAINS")
|
|
print("=" * 60)
|
|
|
|
# Chain A: Patient and tag management
|
|
print("\n--- Chain A: Patient & Tag ---")
|
|
resp = check_api("R04", "A", "1", "Create tag", "POST", "/health/patient-tags", HT,
|
|
body={"name": "Chronic-Disease-Mgmt", "color": "#00FF00"})
|
|
hm_tag_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "A", "2", "Patient list", "GET", "/health/patients?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R04", "A", "3", "Create patient", "POST", "/health/patients", HT,
|
|
body={"name": "HMTestPatient", "gender": "female", "phone": "13900004444",
|
|
"birth_date": "1988-11-10"})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "A", "4", "Doctor list (read)", "GET", "/health/doctors?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain B: Follow-up management
|
|
print("\n--- Chain B: Follow-up Management ---")
|
|
resp = check_api("R04", "B", "1", "Create follow-up task", "POST", "/health/follow-up-tasks", HT,
|
|
body={"patient_id": patient_id, "follow_up_type": "visit",
|
|
"planned_date": "2026-05-23", "notes": "HM FU task"} if patient_id else None)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "B", "2", "Follow-up list", "GET", "/health/follow-up-tasks?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "B", "3", "Follow-up templates", "GET", "/health/follow-up-templates?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "B", "4", "Action inbox", "GET", "/health/action-inbox?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "B", "5", "Team action inbox", "GET", "/health/action-inbox/team?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain C: Consultation management
|
|
print("\n--- Chain C: Consultation Management ---")
|
|
check_api("R04", "C", "1", "Consultation sessions list", "GET", "/health/consultation-sessions?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain D: Alerts and monitoring
|
|
print("\n--- Chain D: Alerts & Monitoring ---")
|
|
check_api("R04", "D", "1", "Alerts list", "GET", "/health/alerts?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "D", "2", "Alert rules", "GET", "/health/alert-rules?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "D", "3", "Critical value thresholds", "GET", "/health/critical-value-thresholds?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "D", "4", "Devices (read)", "GET", "/health/devices?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "D", "5", "Daily monitoring", "GET", "/health/daily-monitoring?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain E: AI analysis
|
|
print("\n--- Chain E: AI Analysis ---")
|
|
check_api("R04", "E", "1", "AI prompts (read)", "GET", "/ai/prompts?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "E", "2", "AI suggestions", "GET", "/ai/suggestions?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain F: Diagnosis and consent
|
|
print("\n--- Chain F: Diagnosis & Consent ---")
|
|
check_api("R04", "F", "1", "Consents list (via patient)", "GET", f"/health/patients/{patient_id}/consents?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "F", "2", "Patient diagnoses", "GET", f"/health/patients/{patient_id}/diagnoses?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "8.1", "", "Messages", "GET", "/messages?page=1&page_size=10", HT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R04", "9.1", "", "Workflow definitions", "GET", "/workflow/definitions?page=1&page_size=10", HT)
|
|
time.sleep(0.5)
|
|
|
|
# Health Manager permission boundary checks
|
|
print("\n--- Health Manager Permission Boundaries ---")
|
|
check_api_status("R04", "P", "1", "No user management", "GET", "/users?page=1&page_size=1", HT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R04", "P", "2", "No points management", "GET", "/health/admin/points/rules?page=1&page_size=1", HT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R04", "P", "3", "No article management", "GET", "/health/articles?page=1&page_size=1", HT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R04", "P", "4", "No system settings", "GET", "/settings", HT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R04", "P", "5", "No plugin management", "GET", "/admin/plugins", HT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R04", "P", "6", "No BLE gateways", "GET", "/health/ble-gateways?page=1&page_size=1", HT, expected_status=403)
|
|
time.sleep(1)
|
|
|
|
# ========================================
|
|
# R05 - OPERATOR (6 chains + permissions)
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("R05 - OPERATOR BUSINESS CHAINS")
|
|
print("=" * 60)
|
|
|
|
# Chain A: Patient and tag management
|
|
print("\n--- Chain A: Patient & Tag ---")
|
|
resp = check_api("R05", "A", "1", "Create tag", "POST", "/health/patient-tags", OT,
|
|
body={"name": "PostSurgery-Rehab", "color": "#0000FF"})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "A", "2", "Patient list (read)", "GET", "/health/patients?page=1&page_size=10", OT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "A", "3", "Search patient", "GET", "/health/patients?search=MultiRoleTest", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain B: Content publishing
|
|
print("\n--- Chain B: Content Publishing ---")
|
|
resp = check_api("R05", "B", "1", "Create article", "POST", "/health/articles", OT,
|
|
body={"title": "Operator Test Article", "content": "Operator content", "status": "draft"})
|
|
op_art_id = resp["data"]["id"] if resp and resp.get("success") else ""
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "B", "2", "Edit article", "PUT", f"/health/articles/{op_art_id}", OT,
|
|
body={"title": "Operator Test Article Updated", "content": "Updated"} if op_art_id else None)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "B", "3", "Submit article", "POST", f"/health/articles/{op_art_id}/submit", OT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "B", "4", "Approve article", "POST", f"/health/articles/{op_art_id}/approve", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain C: Points mall
|
|
print("\n--- Chain C: Points Mall ---")
|
|
check_api("R05", "C", "1", "Points rules", "GET", "/health/admin/points/rules?page=1&page_size=10", OT)
|
|
time.sleep(0.3)
|
|
|
|
resp = check_api("R05", "C", "2", "Create points product", "POST", "/health/admin/points/products", OT,
|
|
body={"name": "Test Product", "points_required": 100, "stock": 50, "description": "Test product"})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "C", "3", "Points orders", "GET", "/health/admin/points/orders?page=1&page_size=10", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain D: Offline events
|
|
print("\n--- Chain D: Offline Events ---")
|
|
resp = check_api("R05", "D", "1", "Create event", "POST", "/health/admin/offline-events", OT,
|
|
body={"title": "Operator Test Event", "description": "Test event by operator",
|
|
"event_date": "2026-06-15", "location": "Hall B", "max_participants": 30})
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "D", "2", "List events", "GET", "/health/offline-events?page=1&page_size=10", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain E: Device and alert viewing
|
|
print("\n--- Chain E: Device & Alert Viewing ---")
|
|
check_api("R05", "E", "1", "Devices (read)", "GET", "/health/devices?page=1&page_size=10", OT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "E", "2", "Alerts (read)", "GET", "/health/alerts?page=1&page_size=10", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Chain F: AI usage
|
|
print("\n--- Chain F: AI Usage ---")
|
|
check_api("R05", "F", "1", "AI suggestions (read)", "GET", "/ai/suggestions?page=1&page_size=10", OT)
|
|
time.sleep(0.3)
|
|
|
|
check_api("R05", "8.1", "", "Messages", "GET", "/messages?page=1&page_size=10", OT)
|
|
time.sleep(0.5)
|
|
|
|
# Operator permission boundary checks
|
|
print("\n--- Operator Permission Boundaries ---")
|
|
check_api_status("R05", "P", "1", "No user management", "GET", "/users?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "2", "No doctor management", "GET", "/health/doctors?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "3", "No follow-up management", "GET", "/health/follow-up-tasks?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "4", "No consultation management", "GET", "/health/consultation-sessions?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "5", "No diagnoses", "GET", f"/health/patients/{patient_id}/diagnoses?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "6", "No action inbox", "GET", "/health/action-inbox?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "7", "No consents", "GET", "/health/consents?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "8", "No AI analysis", "GET", "/ai/prompts?page=1&page_size=1", OT, expected_status=403)
|
|
time.sleep(0.3)
|
|
|
|
check_api_status("R05", "P", "9", "No system settings", "GET", "/settings", OT, expected_status=403)
|
|
time.sleep(1)
|
|
|
|
# ========================================
|
|
# CROSS-ROLE COLLABORATION CHECKS
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("CROSS-ROLE COLLABORATION CHECKS")
|
|
print("=" * 60)
|
|
|
|
print("\n--- Cross-role: Admin patient visible to all ---")
|
|
# Doctor sees admin-created patient
|
|
_, dr_resp = api("GET", f"/health/patients/{patient_id}", DT)
|
|
dr_sees = dr_resp.get("success", False)
|
|
record("XROLE", "X", "1", "Doctor sees admin-created patient", dr_sees,
|
|
"" if dr_sees else f"msg={dr_resp.get('message', '')}")
|
|
time.sleep(0.3)
|
|
|
|
# Nurse sees admin-created patient
|
|
_, nr_resp = api("GET", f"/health/patients/{patient_id}", NT)
|
|
nr_sees = nr_resp.get("success", False)
|
|
record("XROLE", "X", "2", "Nurse sees admin-created patient", nr_sees,
|
|
"" if nr_sees else f"msg={nr_resp.get('message', '')}")
|
|
time.sleep(0.3)
|
|
|
|
# HM sees admin-created patient
|
|
_, hm_resp = api("GET", f"/health/patients/{patient_id}", HT)
|
|
hm_sees = hm_resp.get("success", False)
|
|
record("XROLE", "X", "3", "HM sees admin-created patient", hm_sees,
|
|
"" if hm_sees else f"msg={hm_resp.get('message', '')}")
|
|
time.sleep(0.3)
|
|
|
|
# Operator sees admin-created patient
|
|
_, op_resp = api("GET", f"/health/patients/{patient_id}", OT)
|
|
op_sees = op_resp.get("success", False)
|
|
record("XROLE", "X", "4", "Operator sees admin-created patient", op_sees,
|
|
"" if op_sees else f"msg={op_resp.get('message', '')}")
|
|
time.sleep(0.5)
|
|
|
|
print("\n--- Cross-role: Doctor follow-up visible to nurse ---")
|
|
_, nurse_fu = api("GET", "/health/follow-up-tasks?status=pending&page=1&page_size=20", NT)
|
|
nurse_fu_ok = nurse_fu.get("success", False)
|
|
total_fu = nurse_fu.get("data", {}).get("total", 0) if nurse_fu_ok else 0
|
|
record("XROLE", "X", "5", "Nurse sees follow-up tasks (incl. doctor-created)", nurse_fu_ok and total_fu > 0,
|
|
f"total={total_fu}" if nurse_fu_ok else f"msg={nurse_fu.get('message', '')}")
|
|
time.sleep(0.5)
|
|
|
|
print("\n--- Cross-role: Admin tag visible to other roles ---")
|
|
_, dr_tags = api("GET", "/health/patient-tags?page=1&page_size=20", DT)
|
|
dr_tag_ok = dr_tags.get("success", False)
|
|
record("XROLE", "X", "6", "Doctor can list tags", dr_tag_ok,
|
|
"" if dr_tag_ok else f"status=403 - doctor has no tag.list permission (expected per R02.P.7)")
|
|
time.sleep(0.3)
|
|
|
|
_, hm_tags = api("GET", "/health/patient-tags?page=1&page_size=20", HT)
|
|
hm_tag_ok = hm_tags.get("success", False)
|
|
record("XROLE", "X", "7", "HM can list tags", hm_tag_ok,
|
|
"" if hm_tag_ok else f"msg={hm_tags.get('message', '')}")
|
|
time.sleep(0.3)
|
|
|
|
_, op_tags = api("GET", "/health/patient-tags?page=1&page_size=20", OT)
|
|
op_tag_ok = op_tags.get("success", False)
|
|
record("XROLE", "X", "8", "Operator can list tags", op_tag_ok,
|
|
"" if op_tag_ok else f"msg={op_tags.get('message', '')}")
|
|
time.sleep(0.5)
|
|
|
|
print("\n--- Cross-role: Alert visibility ---")
|
|
_, dr_alerts = api("GET", "/health/alerts?page=1&page_size=5", DT)
|
|
_, nt_alerts = api("GET", "/health/alerts?page=1&page_size=5", NT)
|
|
_, hm_alerts = api("GET", "/health/alerts?page=1&page_size=5", HT)
|
|
record("XROLE", "X", "9", "Doctor can view alerts", dr_alerts.get("success", False))
|
|
record("XROLE", "X", "10", "Nurse can view alerts", nt_alerts.get("success", False))
|
|
record("XROLE", "X", "11", "HM can view alerts", hm_alerts.get("success", False))
|
|
time.sleep(0.5)
|
|
|
|
print("\n--- Cross-role: Article visibility ---")
|
|
_, op_articles = api("GET", "/health/articles?page=1&page_size=5", OT)
|
|
op_art_ok = op_articles.get("success", False)
|
|
record("XROLE", "X", "12", "Operator can manage articles", op_art_ok)
|
|
time.sleep(0.3)
|
|
|
|
# Admin should also be able to see articles
|
|
_, admin_articles = api("GET", "/health/articles?page=1&page_size=5", AT)
|
|
record("XROLE", "X", "13", "Admin can see operator articles", admin_articles.get("success", False))
|
|
time.sleep(0.5)
|
|
|
|
print("\n--- Cross-role: Offline events visibility ---")
|
|
_, op_events = api("GET", "/health/offline-events?page=1&page_size=5", OT)
|
|
_, admin_events = api("GET", "/health/offline-events?page=1&page_size=5", AT)
|
|
record("XROLE", "X", "14", "Operator sees offline events", op_events.get("success", False))
|
|
record("XROLE", "X", "15", "Admin sees offline events", admin_events.get("success", False))
|
|
|
|
# ========================================
|
|
# Summary
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print("TEST SUMMARY")
|
|
print("=" * 60)
|
|
|
|
total = len(RESULTS)
|
|
passed = sum(1 for r in RESULTS if r["passed"])
|
|
failed = total - passed
|
|
|
|
# By role
|
|
for role in ["R01", "R02", "R03", "R04", "R05", "XROLE"]:
|
|
role_results = [r for r in RESULTS if r["role"] == role]
|
|
role_pass = sum(1 for r in role_results if r["passed"])
|
|
role_total = len(role_results)
|
|
print(f" {role}: {role_pass}/{role_total} passed ({100*role_pass//role_total if role_total else 0}%)")
|
|
|
|
print(f"\n TOTAL: {passed}/{total} passed ({100*passed//total if total else 0}%)")
|
|
|
|
if failed > 0:
|
|
print(f"\n FAILED TESTS ({failed}):")
|
|
for r in RESULTS:
|
|
if not r["passed"]:
|
|
print(f" [{r['role']}-{r['chain']}.{r['test_id']}] {r['description']}: {r['detail']}")
|
|
|
|
# Save results as JSON for report generation
|
|
with open(r"G:\hms\docs\qa\role-test-results\test_results.json", "w", encoding="utf-8") as f:
|
|
json.dump({"timestamp": datetime.now().isoformat(), "total": total, "passed": passed,
|
|
"failed": failed, "results": RESULTS}, f, ensure_ascii=False, indent=2)
|
|
print(f"\n Results saved to docs/qa/role-test-results/test_results.json")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|