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.8 KiB
Python
81 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""2026-09-07 预算版本管理UI修复 - 生产API验证"""
|
||
import json, urllib.request, ssl, sys
|
||
|
||
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 = "https://cma.sxbh.ltd/api/cma"
|
||
|
||
def http(url, data=None, headers=None, method=None):
|
||
req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
|
||
ctx = ssl.create_default_context()
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
with urllib.request.urlopen(req, timeout=30, context=ctx) as r:
|
||
return json.loads(r.read().decode())
|
||
|
||
def login(entity_id):
|
||
body = json.dumps({"username": ADMIN_USER, "password": ADMIN_PASS, "entity_id": entity_id}).encode()
|
||
r = http(BASE + "/auth/login", data=body, headers={"Content-Type": "application/json"}, method="POST")
|
||
return r["token"]
|
||
|
||
def versions(token, year=None):
|
||
url = BASE + "/budget/versions" + (f"?year={year}" if year else "")
|
||
r = http(url, headers={"Authorization": f"Bearer {token}"})
|
||
return r if isinstance(r, list) else r.get("data", r)
|
||
|
||
ok = True
|
||
for eid, ename in [(1, "酣客"), (2, "博海")]:
|
||
try:
|
||
tok = login(eid)
|
||
except Exception as e:
|
||
print(f"[{ename} entity{eid}] 登录失败: {e}")
|
||
ok = False
|
||
continue
|
||
try:
|
||
vs = versions(tok)
|
||
print(f"\n=== {ename} (entity={eid}) 共{len(vs)}个版本 ===")
|
||
for v in vs:
|
||
keys = sorted(v.keys())
|
||
print(f" {v['version']:>10} | status={v['status']:<10} | approval_status={v['approval_status']:<10} | plan_count={v['plan_count']} | kpi_count={v['kpi_count']} | total={v['total_budget']} | last_updated={v['last_updated']!r} | period_count={v['period_count']}")
|
||
# 断言新字段存在
|
||
for f in ["approval_status", "plan_count", "last_updated", "status", "kpi_count", "updated_at"]:
|
||
assert f in v, f"缺字段 {f}"
|
||
# 状态机映射断言
|
||
if v["status"] == "active":
|
||
assert v["approval_status"] == "draft", f"{v['version']} active 应映射 draft"
|
||
else:
|
||
assert v["approval_status"] == v["status"]
|
||
assert v["plan_count"] == v["kpi_count"]
|
||
assert v["last_updated"] == v["updated_at"]
|
||
except Exception as e:
|
||
print(f"[{ename}] 版本校验异常: {type(e).__name__} {e}")
|
||
ok = False
|
||
|
||
print("\n" + ("ALL_PASS" if ok else "HAS_FAIL"))
|
||
sys.exit(0 if ok else 1)
|