Files
nj/scripts/e2e_test_followup_consultation_points.py
iven c539e6fd83 feat: initialize Nuanji (Warm Notes) project
- Base platform from base.git (ERP base: auth, core, config, message, workflow, plugin)
- Created erp-diary module skeleton (lib.rs, dto.rs, error.rs, event.rs, state.rs)
- Integrated erp-diary into workspace and erp-server
- Added DiaryModule registration in main.rs
- Added DiaryState FromRef in state.rs
- Diary routes mounted (empty routes, ready for implementation)
- Product design spec v1.2 preserved in docs/
- Implementation plan preserved in plans/

Cargo check: OK
Cargo test: OK (78+ base tests passing)
2026-05-31 20:52:19 +08:00

551 lines
20 KiB
Python

#!/usr/bin/env python3
"""
HMS 健康管理平台 - 随访管理 / 咨询管理 / 积分商城 端到端 API 测试
使用 subprocess + curl 避免Windows urllib兼容问题
"""
import json
import subprocess
import sys
import time
from datetime import datetime, timedelta
BASE = "http://localhost:3000/api/v1"
results = {
"follow_up": [],
"consultation": [],
"points": [],
}
def api(method, path, body=None, token=None):
"""通过 curl 发送 HTTP 请求"""
url = f"{BASE}{path}"
cmd = ["curl", "-s", "-X", method, url, "-H", "Content-Type: application/json"]
if token:
cmd += ["-H", f"Authorization: Bearer {token}"]
if body:
cmd += ["-d", json.dumps(body)]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
text = result.stdout.strip()
if not text:
return {"success": False, "error": f"Empty response (stderr: {result.stderr[:100]})"}, 0
data = json.loads(text)
# Infer status code from response
status = 200
if data.get("success") is False:
error_msg = data.get("error", "") or data.get("message", "")
if "not found" in error_msg.lower() or "404" in str(data.get("status", "")):
status = 404
elif "unauthorized" in error_msg.lower() or "authentication" in error_msg.lower():
status = 401
elif "validation" in error_msg.lower() or "required" in error_msg.lower():
status = 400
elif "too many" in error_msg.lower() or "frequent" in error_msg.lower():
status = 429
elif "forbidden" in error_msg.lower():
status = 403
else:
status = 400
return data, status
except subprocess.TimeoutExpired:
return {"success": False, "error": "Request timeout"}, 0
except json.JSONDecodeError as e:
return {"success": False, "error": f"JSON decode error: {e}"}, 0
except Exception as e:
return {"success": False, "error": str(e)}, 0
def api_raw(method, path, body=None, token=None):
"""通过 curl 发送请求并获取原始 HTTP 状态码,带重试"""
url = f"{BASE}{path}"
max_retries = 3
for attempt in range(max_retries):
cmd = ["curl", "-s", "-w", "\n%{http_code}", "-X", method, url, "-H", "Content-Type: application/json"]
if token:
cmd += ["-H", f"Authorization: Bearer {token}"]
if body:
cmd += ["-d", json.dumps(body)]
try:
result = subprocess.run(cmd, capture_output=True, timeout=15)
output = result.stdout.decode("utf-8", errors="replace").strip()
if not output:
return {"success": False, "error": "Empty response"}, 0
# Split last line as status code
lines = output.rsplit("\n", 1)
if len(lines) == 2 and lines[-1].strip().isdigit():
body_text = lines[0]
status = int(lines[-1].strip())
else:
body_text = output
status = 200
if not body_text:
return {"success": False, "error": "Empty body"}, status
try:
parsed = json.loads(body_text)
except json.JSONDecodeError:
return {"success": False, "error": body_text[:200]}, status
# Retry on 503 (Redis fail-close transient)
if status == 503 and attempt < max_retries - 1:
time.sleep(5)
continue
return parsed, status
except subprocess.TimeoutExpired:
return {"success": False, "error": "Request timeout"}, 0
except Exception as e:
return {"success": False, "error": str(e)}, 0
return {"success": False, "error": "Max retries exceeded (503)"}, 503
def record(section, name, response, status_code, expect_success=True):
"""记录测试结果"""
success = response.get("success") == expect_success
entry = {
"name": name,
"status_code": status_code,
"success": success,
"response_success": response.get("success"),
"error": response.get("error") or response.get("message", ""),
"detail": "",
}
data = response.get("data", {})
if isinstance(data, dict):
if "total" in data:
entry["detail"] = f"total={data['total']}"
if "id" in data:
entry["detail"] = f"id={data['id']}"
results[section].append(entry)
icon = "PASS" if success else "FAIL"
detail_str = f" | {entry['detail']}" if entry["detail"] else ""
err_str = f" | err={entry['error'][:60]}" if entry["error"] and not success else ""
print(f" [{icon}] {name}: HTTP {status_code}{detail_str}{err_str}")
return data
# ============================================================
# 0. 认证
# ============================================================
print("=" * 70)
print("0. Authentication")
print("=" * 70)
resp, code = api_raw("POST", "/auth/login", {"username": "admin", "password": "Admin@2026"})
# Retry login up to 5 times with delay
for _ in range(5):
if code == 200 and resp.get("success"):
break
print(f" Login attempt failed (HTTP {code}), retrying in 10s...")
time.sleep(10)
resp, code = api_raw("POST", "/auth/login", {"username": "admin", "password": "Admin@2026"})
if code != 200 or not resp.get("success"):
print(f" [FAIL] Login failed after retries: HTTP {code}, {json.dumps(resp)[:200]}")
sys.exit(1)
TOKEN = resp["data"]["access_token"]
print(f" [PASS] Login OK: HTTP {code}, token={TOKEN[:30]}...")
# ============================================================
# Base data
# ============================================================
print()
print("=" * 70)
print("Base Data Preparation")
print("=" * 70)
resp, code = api_raw("GET", "/health/patients", token=TOKEN)
PATIENT_ID = None
if resp.get("success") and resp.get("data", {}).get("items"):
PATIENT_ID = resp["data"]["items"][0]["id"]
print(f" Patients total: {resp['data']['total']}, selected ID: {PATIENT_ID}")
else:
# Create a test patient
print(f" No patients found, creating test patient...")
time.sleep(1)
test_patient = {
"name": "E2E Test Patient",
"gender": "male",
"birth_date": "1990-01-15",
"phone": "13800000001",
"id_card_number": "110101199001150011",
}
resp2, code2 = api_raw("POST", "/health/patients", test_patient, token=TOKEN)
if resp2.get("success") and isinstance(resp2.get("data"), dict) and "id" in resp2["data"]:
PATIENT_ID = resp2["data"]["id"]
print(f" Created test patient: {PATIENT_ID}")
else:
print(f" [WARN] Failed to create test patient: HTTP {code2}, {resp2.get('error', resp2.get('message', ''))[:80]}")
resp, code = api_raw("GET", "/health/doctors", token=TOKEN)
DOCTOR_ID = None
if resp.get("success") and resp.get("data", {}).get("items"):
DOCTOR_ID = resp["data"]["items"][0]["id"]
print(f" Doctors total: {resp['data']['total']}, selected ID: {DOCTOR_ID}")
else:
# Create a test doctor - needs user_id
print(f" No doctors found, trying to create...")
# List users to get a user_id for doctor
time.sleep(1)
resp_u, code_u = api_raw("GET", "/users", token=TOKEN)
if resp_u.get("success") and resp_u.get("data", {}).get("items"):
user_id = resp_u["data"]["items"][0]["id"]
test_doctor = {
"user_id": user_id,
"name": "E2E Test Doctor",
"department": "General",
"title": "Attending Physician",
"specialty": "Internal Medicine",
}
time.sleep(1)
resp2, code2 = api_raw("POST", "/health/doctors", test_doctor, token=TOKEN)
if resp2.get("success") and isinstance(resp2.get("data"), dict) and "id" in resp2["data"]:
DOCTOR_ID = resp2["data"]["id"]
print(f" Created test doctor: {DOCTOR_ID}")
else:
print(f" [WARN] Failed to create test doctor: HTTP {code2}, {resp2.get('error', resp2.get('message', ''))[:80]}")
else:
print(f" [WARN] No users to create doctor from")
# ============================================================
# Part 1: Follow-up Management
# ============================================================
print()
print("=" * 70)
print("Part 1: Follow-up Management")
print("=" * 70)
# 1.1 Task list
print()
print("1.1 List follow-up tasks")
resp, code = api_raw("GET", "/health/follow-up-tasks", token=TOKEN)
record("follow_up", "List follow-up tasks", resp, code)
# 1.2 Create task
print()
print("1.2 Create follow-up task")
if PATIENT_ID:
time.sleep(1)
next_week = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")
body = {
"patient_id": PATIENT_ID,
"follow_up_type": "phone",
"planned_date": next_week,
"content_template": "E2E test follow-up task",
}
resp, code = api_raw("POST", "/health/follow-up-tasks", body, token=TOKEN)
task_data = record("follow_up", "Create follow-up task", resp, code)
if resp.get("success") and isinstance(task_data, dict) and "id" in task_data:
task_id = task_data["id"]
task_version = task_data.get("version", 1)
print()
print("1.2a Get follow-up task detail")
resp, code = api_raw("GET", f"/health/follow-up-tasks/{task_id}", token=TOKEN)
record("follow_up", "Get follow-up task detail", resp, code)
print()
print("1.2b Update follow-up task")
time.sleep(1)
update_body = {"content_template": "E2E test updated", "status": "in_progress", "version": task_version}
resp, code = api_raw("PUT", f"/health/follow-up-tasks/{task_id}", update_body, token=TOKEN)
record("follow_up", "Update follow-up task", resp, code)
print()
print("1.2c Create follow-up record")
time.sleep(1)
today = datetime.now().strftime("%Y-%m-%d")
record_body = {
"task_id": task_id,
"executed_date": today,
"result": "contacted",
"patient_condition": "stable",
"medical_advice": "continue medication",
}
resp, code = api_raw("POST", f"/health/follow-up-tasks/{task_id}/records", record_body, token=TOKEN)
record("follow_up", "Create follow-up record", resp, code)
else:
print(" [SKIP] No patient ID")
# 1.3 Template list
print()
print("1.3 List follow-up templates")
resp, code = api_raw("GET", "/health/follow-up-templates", token=TOKEN)
record("follow_up", "List follow-up templates", resp, code)
# 1.4 Record list
print()
print("1.4 List follow-up records")
resp, code = api_raw("GET", "/health/follow-up-records", token=TOKEN)
record("follow_up", "List follow-up records", resp, code)
# 1.5 Auth check - no token
print()
print("1.5 Auth check - no token (expect 401)")
resp, code = api_raw("GET", "/health/follow-up-tasks")
if code == 401:
print(f" [PASS] No token -> 401: HTTP {code}")
results["follow_up"].append({
"name": "Auth check (no token)", "status_code": code, "success": True,
"response_success": resp.get("success"), "error": "", "detail": "401 Unauthorized",
})
else:
print(f" [FAIL] Expected 401, got: HTTP {code}, body={json.dumps(resp)[:100]}")
results["follow_up"].append({
"name": "Auth check (no token)", "status_code": code, "success": False,
"response_success": resp.get("success"), "error": f"Expected 401, got {code}", "detail": "",
})
# 1.6 Validation check - missing required fields
print()
print("1.6 Validation - missing required fields (expect 400/422)")
body_invalid = {"follow_up_type": "phone"}
resp, code = api_raw("POST", "/health/follow-up-tasks", body_invalid, token=TOKEN)
is_validation_error = code in (400, 422)
if is_validation_error:
print(f" [PASS] Validation error: HTTP {code}, msg={resp.get('error', resp.get('message', ''))[:80]}")
else:
print(f" [FAIL] Expected 400/422, got: HTTP {code}, body={json.dumps(resp)[:100]}")
results["follow_up"].append({
"name": "Validation (missing fields)", "status_code": code, "success": is_validation_error,
"response_success": resp.get("success"),
"error": resp.get("error", resp.get("message", ""))[:80],
"detail": f"Expected 400/422, got {code}",
})
# ============================================================
# Part 2: Consultation Management
# ============================================================
print()
print("=" * 70)
print("Part 2: Consultation Management")
print("=" * 70)
# 2.1 Session list
print()
print("2.1 List consultation sessions")
resp, code = api_raw("GET", "/health/consultation-sessions", token=TOKEN)
record("consultation", "List consultation sessions", resp, code)
# 2.2 Create session
print()
print("2.2 Create consultation session")
if PATIENT_ID and DOCTOR_ID:
time.sleep(1)
body = {
"patient_id": PATIENT_ID,
"doctor_id": DOCTOR_ID,
"subject": "E2E test consultation",
}
resp, code = api_raw("POST", "/health/consultation-sessions", body, token=TOKEN)
consult_data = record("consultation", "Create consultation session", resp, code)
if resp.get("success") and isinstance(consult_data, dict) and "id" in consult_data:
consult_id = consult_data["id"]
print()
print("2.2a Get consultation detail")
resp, code = api_raw("GET", f"/health/consultation-sessions/{consult_id}", token=TOKEN)
record("consultation", "Get consultation detail", resp, code)
print()
print("2.2b Send consultation message")
msg_body = {
"session_id": consult_id,
"content": "E2E test message",
"message_type": "text",
}
resp, code = api_raw("POST", "/health/consultation-messages", msg_body, token=TOKEN)
record("consultation", "Send consultation message", resp, code)
print()
print("2.2c List consultation messages")
resp, code = api_raw("GET", f"/health/consultation-sessions/{consult_id}/messages", token=TOKEN)
record("consultation", "List consultation messages", resp, code)
print()
print("2.2d Close consultation session")
resp, code = api_raw("PUT", f"/health/consultation-sessions/{consult_id}/close", token=TOKEN)
record("consultation", "Close consultation session", resp, code)
print()
print("2.2e Create follow-up from consultation")
fu_body = {
"follow_up_type": "phone",
"planned_date": next_week if PATIENT_ID else "2026-06-01",
"content_template": "Follow-up from E2E consultation",
}
resp, code = api_raw("POST", f"/health/consultation-sessions/{consult_id}/follow-up", fu_body, token=TOKEN)
record("consultation", "Create follow-up from consultation", resp, code)
else:
print(f" [SKIP] Missing patient or doctor ID")
# 2.3 Auth check
print()
print("2.3 Auth check - no token (expect 401)")
resp, code = api_raw("GET", "/health/consultation-sessions")
if code == 401:
print(f" [PASS] No token -> 401: HTTP {code}")
results["consultation"].append({
"name": "Auth check (no token)", "status_code": code, "success": True,
"response_success": resp.get("success"), "error": "", "detail": "401 Unauthorized",
})
else:
print(f" [FAIL] Expected 401, got: HTTP {code}")
results["consultation"].append({
"name": "Auth check (no token)", "status_code": code, "success": False,
"response_success": resp.get("success"), "error": f"Expected 401, got {code}", "detail": "",
})
# 2.4 Doctor dashboard
print()
print("2.4 Doctor dashboard")
resp, code = api_raw("GET", "/health/doctor/dashboard", token=TOKEN)
record("consultation", "Doctor dashboard", resp, code)
# ============================================================
# Part 3: Points / Rewards
# ============================================================
print()
print("=" * 70)
print("Part 3: Points / Rewards System")
print("=" * 70)
# 3.1 Rules
print()
print("3.1 List points rules (admin)")
resp, code = api_raw("GET", "/health/admin/points/rules", token=TOKEN)
record("points", "List points rules", resp, code)
# 3.2 Products
print()
print("3.2 List points products (admin)")
resp, code = api_raw("GET", "/health/admin/points/products", token=TOKEN)
record("points", "List points products (admin)", resp, code)
# 3.3 Orders
print()
print("3.3 List points orders (admin)")
resp, code = api_raw("GET", "/health/admin/points/orders", token=TOKEN)
record("points", "List points orders (admin)", resp, code)
# 3.4 Account
print()
print("3.4 Get points account (patient)")
resp, code = api_raw("GET", "/health/points/account", token=TOKEN)
record("points", "Get points account", resp, code)
# 3.5 Checkin
print()
print("3.5 Daily checkin")
resp, code = api_raw("POST", "/health/points/checkin", token=TOKEN)
is_checkin_ok = resp.get("success") == True or "already" in str(resp.get("error", "")).lower() or "already" in str(resp.get("message", "")).lower()
print(f" [{'PASS' if is_checkin_ok else 'INFO'}] Checkin: HTTP {code}, success={resp.get('success')}, error={resp.get('error', resp.get('message', ''))}")
results["points"].append({
"name": "Daily checkin", "status_code": code, "success": is_checkin_ok,
"response_success": resp.get("success"),
"error": resp.get("error", resp.get("message", "")),
"detail": "Success or already checked in = PASS",
})
# 3.6 Checkin status
print()
print("3.6 Checkin status")
resp, code = api_raw("GET", "/health/points/checkin/status", token=TOKEN)
record("points", "Checkin status", resp, code)
# 3.7 Transactions
print()
print("3.7 List transactions")
resp, code = api_raw("GET", "/health/points/transactions", token=TOKEN)
record("points", "List transactions", resp, code)
# 3.8 Products (patient view)
print()
print("3.8 List products (patient)")
resp, code = api_raw("GET", "/health/points/products", token=TOKEN)
record("points", "List products (patient)", resp, code)
# 3.9 Statistics (admin)
print()
print("3.9 Points statistics (admin)")
resp, code = api_raw("GET", "/health/admin/points/statistics", token=TOKEN)
record("points", "Points statistics", resp, code)
# 3.10 Offline events (patient)
print()
print("3.10 List offline events (patient)")
resp, code = api_raw("GET", "/health/offline-events", token=TOKEN)
record("points", "List offline events", resp, code)
# 3.11 Auth check
print()
print("3.11 Auth check - no token (expect 401)")
resp, code = api_raw("GET", "/health/admin/points/rules")
if code == 401:
print(f" [PASS] No token -> 401: HTTP {code}")
results["points"].append({
"name": "Auth check (no token)", "status_code": code, "success": True,
"response_success": resp.get("success"), "error": "", "detail": "401 Unauthorized",
})
else:
print(f" [FAIL] Expected 401, got: HTTP {code}")
results["points"].append({
"name": "Auth check (no token)", "status_code": code, "success": False,
"response_success": resp.get("success"), "error": f"Expected 401, got {code}", "detail": "",
})
# ============================================================
# Summary Report
# ============================================================
print()
print("=" * 70)
print("SUMMARY REPORT")
print("=" * 70)
total_pass = 0
total_fail = 0
for section_name, section_key in [("Follow-up", "follow_up"), ("Consultation", "consultation"), ("Points/Rewards", "points")]:
items = results[section_key]
passed = sum(1 for i in items if i["success"])
failed = sum(1 for i in items if not i["success"])
total_pass += passed
total_fail += failed
print()
print(f"--- {section_name} ---")
print(f" PASS: {passed}, FAIL: {failed}, Total: {len(items)}")
for item in items:
icon = "PASS" if item["success"] else "FAIL"
detail = f" | {item['detail']}" if item["detail"] else ""
print(f" [{icon}] {item['name']}: HTTP {item['status_code']}{detail}")
total = total_pass + total_fail
rate = (total_pass / total * 100) if total > 0 else 0
print()
print("=" * 70)
print(f"TOTAL: {total} tests, PASS {total_pass} ({rate:.1f}%), FAIL {total_fail}")
print("=" * 70)
if total_fail > 0:
print()
print("FAILED ITEMS:")
for section_name, section_key in [("Follow-up", "follow_up"), ("Consultation", "consultation"), ("Points/Rewards", "points")]:
for item in results[section_key]:
if not item["success"]:
print(f" - [{section_name}] {item['name']}: HTTP {item['status_code']}, error={item['error']}")