Hermes CI Fix
40d522cd34
- Login.vue: no prefilled default creds; no anonymous entity enumeration on mount - deploy.sh: readiness polling (0.5s/30s cap) replaces fixed sleep 3; login smoke reads same-source 600 secret file, missing file = explicit abort - 19 verify/check/gen_report scripts: shared _admin_creds() loader (CMA_ADMIN_ENV -> ops .secrets -> profile copy); no hardcoded creds - tests: single TEST_USER_PASSWORD fixture constant; docs: pointer to 600 secret file; removed 4 legacy .bak + pycache residue
81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
"""验证年度预算分解幂等 — R5 (2026-08-30)
|
||
|
||
调用 /api/cma/budget/auto-decompose 3 次，对比月度预算值是否不变。
|
||
用法: cd /root/cma-management/backend && ./venv/bin/python3 scripts/verify_decompose_idempotent.py
|
||
"""
|
||
import sys
|
||
import os
|
||
import json
|
||
import urllib.request
|
||
|
||
# ---- 管理员凭据同源加载（2026-09-14 cma-defaultcred-code-fix）----
|
||
# 生产凭据只存 600 密钥文件；禁止在脚本/文档/日志里硬编码口令。
|
||
# 顺序：$CMA_ADMIN_ENV → wecom-ops 主副本（ops 轮换落点，权威） → 本 profile 副本。
|
||
def _admin_creds():
|
||
"""返回 (user, password, 来源文件)；未找到 → 显式报错退出（不静默退回默认口令）"""
|
||
for p in (os.environ.get("CMA_ADMIN_ENV"),
|
||
"/root/.hermes/profiles/wecom-ops/.secrets/cma-admin.env",
|
||
"/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env"):
|
||
if p and os.path.isfile(p):
|
||
kv = {}
|
||
for line in open(p, encoding="utf-8"):
|
||
line = line.strip()
|
||
if "=" in line and not line.startswith("#"):
|
||
k, v = line.split("=", 1)
|
||
kv[k.strip()] = v.strip().strip('"').strip("'")
|
||
if kv.get("CMA_ADMIN_USER") and kv.get("CMA_ADMIN_PASS"):
|
||
return kv["CMA_ADMIN_USER"], kv["CMA_ADMIN_PASS"], p
|
||
raise SystemExit("[错误] 未找到 CMA 管理员凭据：请设 $CMA_ADMIN_ENV，或准备 600 权限密钥文件 "
|
||
".secrets/cma-admin.env（键名 CMA_ADMIN_USER / CMA_ADMIN_PASS）")
|
||
|
||
|
||
ADMIN_USER, ADMIN_PASS, CRED_SRC = _admin_creds()
|
||
print(f"[creds] 来源={CRED_SRC} 账号={ADMIN_USER}（口令不回显）")
|
||
|
||
BASE = os.getenv("CMA_BASE", "http://127.0.0.1:8010")
|
||
|
||
|
||
def post(path, body, token=None):
|
||
req = urllib.request.Request(
|
||
BASE + path,
|
||
data=json.dumps(body).encode("utf-8"),
|
||
headers={"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {token}" if token else ""},
|
||
)
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def main():
|
||
# 登录（账套模式必须 entity_id）
|
||
login = post("/api/cma/auth/login", {"username": ADMIN_USER, "password": ADMIN_PASS, "entity_id": 1})
|
||
token = login.get("token") or login.get("access_token")
|
||
if not token:
|
||
print("❌ 登录失败:", login)
|
||
sys.exit(1)
|
||
print("✅ 登录成功")
|
||
|
||
runs = []
|
||
for i in range(3):
|
||
r = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token)
|
||
print(f"第{i+1}次: {r.get('message', '')} created={r.get('created', 0)}")
|
||
# 提取 (kpi_id -> monthly tuple)
|
||
snap = {}
|
||
for res in r.get("results", []):
|
||
snap[res["kpi_id"]] = tuple(res.get("monthly") or [])
|
||
runs.append(snap)
|
||
|
||
# 对比三次结果
|
||
same = runs[0] == runs[1] == runs[2]
|
||
print(f"\n三次结果一致: {'✅ 是（幂等）' if same else '❌ 否（不幂等）'}")
|
||
if not same:
|
||
for i in range(1, 3):
|
||
for kid in runs[0]:
|
||
if runs[0].get(kid) != runs[i].get(kid):
|
||
print(f" KPI {kid} 第1次={runs[0].get(kid)} 第{i+1}次={runs[i].get(kid)}")
|
||
sys.exit(0 if same else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|