281 lines
11 KiB
Python
281 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
UniAI Health (原愈极刻) 定制设备治疗参数功能 - 端到端全链路验证工具
|
|
1. 校验 RESTful API (管理员登录鉴权、规则管理、标签打标)
|
|
2. 模拟 IoT 设备与后台 UDP 服务 (端口 61100) 双向报文交互
|
|
3. 验证四层参数决策引擎：标签定制、单机定制、安全截断兜底、基准兜底
|
|
"""
|
|
|
|
import socket
|
|
import sys
|
|
import time
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
GREEN = "\033[92m"
|
|
RED = "\033[91m"
|
|
YELLOW = "\033[93m"
|
|
BLUE = "\033[94m"
|
|
RESET = "\033[0m"
|
|
|
|
UDP_HOST = "127.0.0.1"
|
|
UDP_PORT = 61100
|
|
HTTP_BASE = "http://127.0.0.1:8082"
|
|
|
|
def calc_crc(content: str) -> str:
|
|
"""计算异或校验码 (XOR Checksum)"""
|
|
if "$" in content and "*" in content:
|
|
raw = content[content.index("$") + 1 : content.rindex("*")]
|
|
else:
|
|
raw = content
|
|
xor_val = 0
|
|
for ch in raw.encode("ascii"):
|
|
xor_val ^= ch
|
|
return f"{xor_val:02X}"
|
|
|
|
def build_packet(inner_body: str) -> bytes:
|
|
"""构建带 $ 与 *{CRC}\\r\\n 的完整报文"""
|
|
raw = f"${inner_body}*"
|
|
crc = calc_crc(raw)
|
|
packet_str = f"${inner_body}*{crc}\r\n"
|
|
return packet_str.encode("utf-8")
|
|
|
|
def parse_kv(packet_str: str) -> dict:
|
|
"""解析以管道符分隔的报文内容"""
|
|
packet_str = packet_str.strip().strip("\r\n")
|
|
if packet_str.startswith("$"):
|
|
packet_str = packet_str[1:]
|
|
if "*" in packet_str:
|
|
packet_str = packet_str[:packet_str.rindex("*")]
|
|
|
|
parts = packet_str.split("|")
|
|
res = {}
|
|
if parts:
|
|
res["CMD"] = parts[0]
|
|
for p in parts[1:]:
|
|
if len(p) >= 2:
|
|
res[p[:2]] = p[2:]
|
|
return res
|
|
|
|
def test_login() -> str:
|
|
"""登录系统获取管理员 JWT Token"""
|
|
url = f"{HTTP_BASE}/pc/loginByUsername"
|
|
payload = json.dumps({"username": "admin", "password": "admin123"}).encode("utf-8")
|
|
req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=3) as resp:
|
|
raw_data = json.loads(resp.read().decode("utf-8"))
|
|
token = raw_data.get("token")
|
|
if not token and isinstance(raw_data.get("data"), dict):
|
|
token = raw_data["data"].get("token")
|
|
if token:
|
|
print(f" {GREEN}✓ 管理员身份认证成功 (username: admin){RESET}")
|
|
return token
|
|
else:
|
|
print(f" {RED}✗ 登录成功但未包含 Token: {raw_data}{RESET}")
|
|
return ""
|
|
except Exception as e:
|
|
print(f" {RED}✗ 管理员登录异常: {e}{RESET}")
|
|
return ""
|
|
|
|
def test_rest_api(token: str):
|
|
"""测试 REST 接口：规则分页与批量标签"""
|
|
print(f"\n{BLUE}▶ 测试用例: [REST API] 验证定制规则管理与标签接口{RESET}")
|
|
if not token:
|
|
print(f" {RED}✗ 缺少认证 Token，跳过鉴权接口测试{RESET}")
|
|
return False
|
|
|
|
headers = {
|
|
"jkzy-auth": token,
|
|
"User-Agent": "VerifyScript",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# 1. 查询规则列表
|
|
url_rules = f"{HTTP_BASE}/pc/admin/deviceTreatRule/page?pageNo=1&pageSize=10"
|
|
try:
|
|
req = urllib.request.Request(url_rules, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=3) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
if data.get("code") == 200:
|
|
records = data.get("data", {}).get("records", [])
|
|
print(f" {GREEN}✓ GET /pc/admin/deviceTreatRule/page 响应成功，已加载 {len(records)} 条定制规则{RESET}")
|
|
for r in records[:3]:
|
|
print(f" - [{r.get('ruleName')}] 类型:{r.get('targetType')} 目标:{r.get('targetValue')} FR:{r.get('fr')} UL:{r.get('ul')}")
|
|
else:
|
|
print(f" {RED}✗ 规则接口返回错误: {data}{RESET}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" {RED}✗ 规则列表接口访问失败: {e}{RESET}")
|
|
return False
|
|
|
|
# 2. 批量打标测试
|
|
url_tags = f"{HTTP_BASE}/pc/admin/device/setTags"
|
|
try:
|
|
payload = json.dumps({
|
|
"deviceIds": ["DEV_DEF_04"],
|
|
"tags": "temp_test_tag,wsl_verify"
|
|
}).encode("utf-8")
|
|
req = urllib.request.Request(url_tags, data=payload, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=3) as resp:
|
|
data = json.loads(resp.read().decode("utf-8"))
|
|
if data.get("code") == 200:
|
|
print(f" {GREEN}✓ POST /pc/admin/device/setTags 批量设备标签更新成功{RESET}")
|
|
else:
|
|
print(f" {YELLOW}⚠ 打标接口响应: {data.get('message')}{RESET}")
|
|
except Exception as e:
|
|
print(f" {YELLOW}⚠ 打标接口调用异常: {e}{RESET}")
|
|
|
|
return True
|
|
|
|
def test_device_treatment_params(device_id: str, case_name: str, expected_params: dict):
|
|
"""通过 UDP 模拟设备请求并断言下发的参数"""
|
|
print(f"\n{BLUE}▶ 测试用例: [{device_id}] - {case_name}{RESET}")
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.settimeout(3.0)
|
|
|
|
try:
|
|
# 第一步：设备上报状态 03（就绪等待处方）
|
|
req1 = build_packet(f"03|MU{device_id}|CI00|CN00|")
|
|
# print(f" [TX -> 61100] {req1.decode().strip()}")
|
|
sock.sendto(req1, (UDP_HOST, UDP_PORT))
|
|
|
|
# 接收服务端的 03 回复（包含生成的处方流水号 CI）
|
|
data1, _ = sock.recvfrom(2048)
|
|
resp1_str = data1.decode("utf-8", errors="ignore")
|
|
# print(f" [RX <- 61100] {resp1_str.strip()}")
|
|
kv1 = parse_kv(resp1_str)
|
|
|
|
ci = kv1.get("CI")
|
|
if not ci or ci == "00":
|
|
print(f" {RED}✗ 失败: 服务端未返回有效处方流水号 (CI={ci}){RESET}")
|
|
return False
|
|
|
|
# 第二步：设备使用状态 30 请求处方参数分段 TN1
|
|
req2 = build_packet(f"30|MU{device_id}|CI{ci}|TN1|")
|
|
# print(f" [TX -> 61100] {req2.decode().strip()}")
|
|
sock.sendto(req2, (UDP_HOST, UDP_PORT))
|
|
|
|
# 接收服务端下发的 30 处方参数报文
|
|
data2, _ = sock.recvfrom(2048)
|
|
resp2_str = data2.decode("utf-8", errors="ignore")
|
|
# print(f" [RX <- 61100] {resp2_str.strip()}")
|
|
kv2 = parse_kv(resp2_str)
|
|
|
|
# 断言校验
|
|
all_passed = True
|
|
actual_details = []
|
|
for param, exp_val in expected_params.items():
|
|
act_val = kv2.get(param)
|
|
# 兼容前导 0
|
|
match = (act_val == exp_val) or (act_val and act_val.lstrip("0") == exp_val.lstrip("0"))
|
|
if match:
|
|
actual_details.append(f"{param}={act_val} {GREEN}✓{RESET}")
|
|
else:
|
|
actual_details.append(f"{param}={act_val}(预期:{exp_val}) {RED}✗{RESET}")
|
|
all_passed = False
|
|
|
|
print(" 参数下发结果: " + ", ".join(actual_details))
|
|
if all_passed:
|
|
print(f" {GREEN}✓ PASS: 该设备下发参数完全符合定制/安全/基准规则预期!{RESET}")
|
|
return True
|
|
else:
|
|
print(f" {RED}✗ FAIL: 参数与预期不符!{RESET}")
|
|
return False
|
|
|
|
except socket.timeout:
|
|
print(f" {RED}✗ 接收超时: 服务端未及时回复 UDP 报文{RESET}")
|
|
return False
|
|
except Exception as e:
|
|
print(f" {RED}✗ 异常: {e}{RESET}")
|
|
return False
|
|
finally:
|
|
sock.close()
|
|
|
|
def main():
|
|
print(f"{BLUE}=================================================================={RESET}")
|
|
print(f"{BLUE} UniAI Health 定制治疗参数功能 WSL 本地全链路验证工具 {RESET}")
|
|
print(f"{BLUE}=================================================================={RESET}")
|
|
|
|
# 检查 HTTP 服务就绪
|
|
print(f"\n正在检查后台 HTTP (8082) 与 UDP (61100) 端口可用性...")
|
|
http_ready = False
|
|
try:
|
|
with urllib.request.urlopen(f"{HTTP_BASE}/pc/loginByUsername", timeout=2) as resp:
|
|
http_ready = True
|
|
except urllib.error.HTTPError as e:
|
|
# POST endpoint called with GET will return 405 Method Not Allowed, which proves server is up
|
|
http_ready = (e.code in (400, 405))
|
|
except Exception:
|
|
http_ready = False
|
|
|
|
if not http_ready:
|
|
print(f"{RED}✗ 后台服务尚未就绪或 8082 端口未启动。{RESET}")
|
|
print(f"{YELLOW}请先在 WSL-Ubuntu 终端中运行以下命令启动服务：{RESET}")
|
|
print(f" cd /mnt/d/Workspace/uniaihealth/server")
|
|
print(f" mvn spring-boot:run -pl admin -Dspring-boot.run.profiles=local\n")
|
|
sys.exit(1)
|
|
|
|
print(f"{GREEN}✓ 后台服务连接成功！开始执行测试用例...{RESET}")
|
|
|
|
# 登录获取 Token
|
|
token = test_login()
|
|
|
|
results = []
|
|
|
|
# 1. REST API
|
|
results.append(("REST API 测试", test_rest_api(token)))
|
|
|
|
# 2. 标签匹配覆盖 (DEV_GOLF_01 打了 golf_vip 标签)
|
|
results.append(("标签定制覆盖 (golf_vip)", test_device_treatment_params(
|
|
device_id="DEV_GOLF_01",
|
|
case_name="标签定制匹配 (golf_vip 覆盖基准)",
|
|
expected_params={"FR": "08", "PW": "55", "PL": "12", "UL": "05", "TL": "025"}
|
|
)))
|
|
|
|
# 3. 单机精确匹配覆盖 (DEV_SPEC_02 配置了单机规则)
|
|
results.append(("单机专属定制 (DEV_SPEC_02)", test_device_treatment_params(
|
|
device_id="DEV_SPEC_02",
|
|
case_name="单机专属规则覆盖 (DEV_SPEC_02 覆盖基准与标签)",
|
|
expected_params={"FR": "02", "PW": "40", "PL": "04", "UL": "02", "TL": "010"}
|
|
)))
|
|
|
|
# 4. 非医疗民用版安全极限截断 (DEV_CIVIL_03, cf=2, 规则配置超标 UL=18)
|
|
results.append(("民用机安全截断 (DEV_CIVIL_03)", test_device_treatment_params(
|
|
device_id="DEV_CIVIL_03",
|
|
case_name="硬件安全底线防护 (cf=2 非医疗版强制限幅 UL <= 10)",
|
|
expected_params={"UL": "10"}
|
|
)))
|
|
|
|
# 5. 普通设备基准默认参数兜底 (DEV_DEF_04, 无标签无单机规则)
|
|
results.append(("基准参数兜底 (DEV_DEF_04)", test_device_treatment_params(
|
|
device_id="DEV_DEF_04",
|
|
case_name="全局病症基准参数正常下发兜底",
|
|
expected_params={"FR": "03", "PW": "50", "PL": "08", "UL": "01", "TL": "017"}
|
|
)))
|
|
|
|
# 汇总
|
|
print(f"\n{BLUE}=================================================================={RESET}")
|
|
print(f"{BLUE} 测试结果汇总报告 {RESET}")
|
|
print(f"{BLUE}=================================================================={RESET}")
|
|
total = len(results)
|
|
passed = sum(1 for _, ok in results if ok)
|
|
for name, ok in results:
|
|
status_text = f"{GREEN}PASS ✓{RESET}" if ok else f"{RED}FAIL ✗{RESET}"
|
|
print(f" {name: <35} : {status_text}")
|
|
print(f"{BLUE}------------------------------------------------------------------{RESET}")
|
|
print(f"总计: {total} 项，成功: {GREEN}{passed}{RESET} 项，失败: {RED}{total - passed}{RESET} 项")
|
|
|
|
if passed == total:
|
|
print(f"{GREEN}★ 所有定制治疗参数与调度验证均 100% 成功通过！★{RESET}\n")
|
|
sys.exit(0)
|
|
else:
|
|
print(f"{RED}⚠ 存在未通过的测试项，请检查日志。{RESET}\n")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|