90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
|
"""端到端验证 priority 修复 - 正确路径"""
|
|||
|
|
import json
|
|||
|
|
import secrets
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.parse
|
|||
|
|
import urllib.request
|
|||
|
|
|
|||
|
|
REDIS_PWD = "WKzl7jgKTkeWxFWGIBXfzokm59Gvfr76"
|
|||
|
|
|
|||
|
|
import redis
|
|||
|
|
|
|||
|
|
r = redis.Redis(host="redis", port=6379, password=REDIS_PWD, db=0)
|
|||
|
|
|
|||
|
|
# 写 token
|
|||
|
|
token = secrets.token_urlsafe(32)
|
|||
|
|
user_info = {
|
|||
|
|
"employee_id": "sxn",
|
|||
|
|
"username": "admin",
|
|||
|
|
"name": "宋献",
|
|||
|
|
"role": "admin",
|
|||
|
|
"department": "IT支持组",
|
|||
|
|
"login_source": "test",
|
|||
|
|
"login_method": "test_script",
|
|||
|
|
"last_active": "2026-07-28T02:00:00",
|
|||
|
|
}
|
|||
|
|
r.setex(f"user:token:{token}", 8 * 3600, json.dumps(user_info, ensure_ascii=False))
|
|||
|
|
print(f"TOKEN: {token}", flush=True)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def call_api(url, params):
|
|||
|
|
# 正确路径：/admin/quick-rules （nginx 会加 /api/ 前缀）
|
|||
|
|
full_url = f"http://127.0.0.1:8000{url}?{urllib.parse.urlencode(params)}"
|
|||
|
|
req = urllib.request.Request(
|
|||
|
|
full_url,
|
|||
|
|
headers={
|
|||
|
|
"Authorization": f"Bearer {token}",
|
|||
|
|
"X-Forwarded-For": "10.240.1.100",
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
try:
|
|||
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|||
|
|
raw = resp.read().decode("utf-8")
|
|||
|
|
return json.loads(raw)
|
|||
|
|
except urllib.error.HTTPError as e:
|
|||
|
|
try:
|
|||
|
|
body = e.read().decode("utf-8")
|
|||
|
|
except Exception:
|
|||
|
|
body = ""
|
|||
|
|
try:
|
|||
|
|
return json.loads(body)
|
|||
|
|
except Exception:
|
|||
|
|
return {"error": str(e), "body": body}
|
|||
|
|
except Exception as e:
|
|||
|
|
return {"error": str(e)}
|
|||
|
|
|
|||
|
|
|
|||
|
|
tests = [
|
|||
|
|
("rule_type=routing_target", {"rule_type": "routing_target", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=行政", {"rule_type": "routing_target", "category": "行政", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=人力资源", {"rule_type": "routing_target", "category": "人力资源", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=财务", {"rule_type": "routing_target", "category": "财务", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=法务", {"rule_type": "routing_target", "category": "法务", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=行政-物业", {"rule_type": "routing_target", "category": "行政-物业", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=IT服务", {"rule_type": "routing_target", "category": "IT服务", "page": 1, "size": 20}),
|
|||
|
|
("rule_type=routing_target&category=不存在的分类", {"rule_type": "routing_target", "category": "不存在的分类", "page": 1, "size": 20}),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
results = []
|
|||
|
|
for name, params in tests:
|
|||
|
|
print(f"\n===== Test: {name} =====", flush=True)
|
|||
|
|
result = call_api("/admin/quick-rules", params)
|
|||
|
|
code = result.get("code")
|
|||
|
|
if code == 0:
|
|||
|
|
data = result.get("data", {})
|
|||
|
|
items = data.get("items", [])
|
|||
|
|
print(f"✅ PASS: total={data.get('total')}, items={len(items)}", flush=True)
|
|||
|
|
if items:
|
|||
|
|
first = items[0]
|
|||
|
|
print(f" first: priority={first.get('priority')}, category={first.get('category')}, keyword={first.get('keyword')}", flush=True)
|
|||
|
|
results.append("PASS")
|
|||
|
|
else:
|
|||
|
|
print(f"❌ FAIL: code={code} msg={result.get('message') or result.get('body')}", flush=True)
|
|||
|
|
results.append("FAIL")
|
|||
|
|
|
|||
|
|
print(f"\n===== 汇总: {results.count('PASS')}/{len(results)} 通过 =====", flush=True)
|
|||
|
|
|
|||
|
|
# 清理
|
|||
|
|
r.delete(f"user:token:{token}")
|
|||
|
|
print("===== Cleanup OK =====", flush=True)
|