Simon
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)
|