Files
cma-management/backend/scripts/verify_decompose_dialog_review.py
T
Hermes CI Fix 40d522cd34 fix(security): retire default-credential surface (yanxue-20260914-cma-defaultcred-code-fix)
- 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
2026-09-14 17:39:53 +08:00

105 lines
4.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""budget-decompose-dialog-fix 独立复核:API 级实测 auto-decompose 全链路"""
import json
import sys
import urllib.request
import os # noqa: E402 (凭据加载器需要)
# ---- 管理员凭据同源加载(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 = "http://127.0.0.1:8010"
def post(path, body, token=None, method="POST"):
req = urllib.request.Request(
BASE + path,
data=json.dumps(body).encode("utf-8"),
method=method,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}" if token else "",
},
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.status, json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as e:
try:
return e.code, json.loads(e.read().decode("utf-8"))
except Exception:
return e.code, {"detail": e.read().decode("utf-8", "ignore")}
def main():
# 1. 登录(账套模式 entity_id=1)
status, 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 status != 200 or not token:
print("FAIL login:", status, login)
sys.exit(1)
print("PASS 登录成功, token 前缀:", token[:12], "...")
# 2. 调用 auto-decompose(equal 均分)
status, r = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token)
print("auto-decompose status:", status)
if status != 200:
print(" detail:", r.get("detail", r))
print("FAIL auto-decompose 非200(可能该年无年度预算数据)")
sys.exit(2)
print(" message:", r.get("message"))
results = r.get("results") or []
print(" created:", r.get("created"), " results数:", len(results))
for res in results[:5]:
print(" -", res.get("kpi_code"), res.get("kpi_name"),
"annual=", res.get("annual_budget"), "method=", res.get("method"),
"monthly_count=", len(res.get("monthly") or []))
if not results:
print("FAIL results 为空")
sys.exit(3)
# 3. 验证每条结果字段完整(前端表格依赖)
required = ["kpi_code", "kpi_name", "annual_budget", "method", "monthly"]
for res in results:
missing = [k for k in required if k not in res]
if missing:
print("FAIL 结果缺字段:", missing, res)
sys.exit(4)
if not res.get("monthly"):
print("FAIL monthly 为空:", res.get("kpi_code"))
sys.exit(5)
print("PASS 所有结果字段完整(kpi_code/kpi_name/annual_budget/method/monthly)")
# 4. 幂等抽查:再调一次,结果一致
status2, r2 = post("/api/cma/budget/auto-decompose", {"year": 2026, "method": "equal", "version": "v1.0"}, token)
snap1 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in results}
snap2 = {res["kpi_id"]: tuple(res.get("monthly") or []) for res in (r2.get("results") or [])}
print("PASS 二次调用幂等一致" if snap1 == snap2 else "WARN 二次调用结果不同(非幂等)")
print("\nRESULT: API 全链路通过")
if __name__ == "__main__":
main()