feat(scripts): 收纳运维/部署/修复/调试脚本到 scripts/ 目录
**目录结构**: - scripts/ 顶层运维/工具脚本(20 个) - scripts/deploy/ 一键部署脚本(deploy_*.py / upload_*.py / redeploy.py 等 8 个) - scripts/fix/ 一次性修复脚本(fix_*.py / fix_*.cypher / fix_*.sql 等 9 个) - scripts/debug/ 调试脚本集合(check_*.py/sql、test_*.py、debug_*.py 等 42 个) **ops-tools/jms_ops.py**:jumpserver-ops skill 默认路径修复 - 从 'jumpserver-automation-shareable' 改为 'jumpserver-ops' - 同步更新注释(优先 JP_SKILL_DIR 环境变量) **典型脚本用途**: - scripts/deploy-v1.2.ps1 / -staging.ps1:一键部署 v1.2 到生产 / staging - scripts/init_quick_rules.sql:quick_rules 表初始数据 - scripts/gen_admin_token.py:生成 admin JWT token(调试用) - scripts/build-and-verify.sh:v1.1 实施验证脚本 - scripts/verify_*.sh / .py:API/quick_rules 验证 - scripts/fix/*.py:环境修复(compose、env_key、itsm_bridge 等) 合计 81 文件 + 3317 行 / - 1 行
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
"""端到端验证 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)
|
||||
Reference in New Issue
Block a user