- 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
105 lines
4.5 KiB
Python
105 lines
4.5 KiB
Python
"""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()
|