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
223 lines
11 KiB
Python
223 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""生产E2E实证: 成本模型研发归集维度 (cma-rd-dimension-20260905)
|
||
|
||
链路验证:
|
||
A. 标准成本 POST project_type=platform_dev → research_flag 自动=1 (研发归集)
|
||
B. 标准成本 POST 不传归集维度 → 默认 client_delivery/research_flag=0 (录入默认)
|
||
C. 标准成本列表 ?project_type=platform_dev 过滤 → 只含研发行, 不含客户交付
|
||
D. 标准成本列表 ?research_flag=1 过滤 → 只含研发行
|
||
E. 实际成本 POST project_type=platform_dev → 落库 + 列表过滤命中
|
||
F. 存量迁移保守默认: 两表 internal 行数 > 0 (ALTER 后存量全 internal)
|
||
G. 回归: 既有列表接口(标准/实际)返回 200 且 data 为列表
|
||
H. ORM schema: inspect 两表含 research_flag/project_type 列 + 索引
|
||
|
||
数据纪律: 临时行 product_code 带 E2E-RD- 前缀 + remark 带标记,
|
||
验证完成后 DB 精确清理并回查 COUNT=0。
|
||
|
||
用法: cd /root/cma-management/backend && venv/bin/python3 scripts/verify_cost_research_dimension_e2e.py
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime
|
||
|
||
# ---- 管理员凭据同源加载（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"
|
||
MARK = "【系统验证临时数据-验证后删除】"
|
||
PREFIX = "E2E-RD-"
|
||
STAMP = datetime.now().strftime("%H%M%S")
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
os.environ.setdefault("CMA_DB_HOST", "127.0.0.1")
|
||
from app.database import get_db, get_engine # noqa: E402
|
||
from sqlalchemy import text # noqa: E402
|
||
from sqlalchemy import inspect as sa_inspect # noqa: E402
|
||
|
||
|
||
def http(method, path, body=None, headers=None):
|
||
data = json.dumps(body).encode() if body is not None else None
|
||
req = urllib.request.Request(BASE + path, data=data, method=method)
|
||
req.add_header("Content-Type", "application/json")
|
||
for k, v in (headers or {}).items():
|
||
req.add_header(k, v)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30) as r:
|
||
return r.status, json.loads(r.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
raw = e.read().decode()
|
||
try:
|
||
return e.code, json.loads(raw)
|
||
except Exception:
|
||
return e.code, {"detail": raw}
|
||
|
||
|
||
def main():
|
||
failures = []
|
||
|
||
def check(name, cond, detail=""):
|
||
tag = "PASS" if cond else "FAIL"
|
||
print(f" [{tag}] {name}" + (f" | {detail}" if detail else ""))
|
||
if not cond:
|
||
failures.append(name)
|
||
|
||
# ── 0. 登录(admin entity=1) ──
|
||
st, r = http("POST", "/api/cma/auth/login",
|
||
{"username": ADMIN_USER, "password": ADMIN_PASS, "entity_id": 1})
|
||
check("登录 admin entity=1", st == 200, f"http={st}")
|
||
token = r.get("token") if st == 200 else ""
|
||
auth = {"Authorization": f"Bearer {token}"}
|
||
|
||
rd_code = f"{PREFIX}RD{STAMP}"
|
||
cd_code = f"{PREFIX}CD{STAMP}"
|
||
ac_code = f"{PREFIX}AC{STAMP}"
|
||
|
||
try:
|
||
# ── A. 标准成本: 平台研发 ──
|
||
st, r = http("POST", "/api/cma/cost/standard-costs", {
|
||
"product_code": rd_code, "product_name": "CMA平台化开发E2E",
|
||
"cost_type": "labor", "item_name": "研发工时E2E",
|
||
"standard_quantity": 10.0, "standard_price": 80.0, "unit": "小时",
|
||
"project_type": "platform_dev", "remark": MARK,
|
||
}, auth)
|
||
check("POST standard platform_dev 200", st == 200, f"http={st} body={r}")
|
||
|
||
# ── B. 标准成本: 不传归集维度 → 默认 client_delivery ──
|
||
st, r = http("POST", "/api/cma/cost/standard-costs", {
|
||
"product_code": cd_code, "product_name": "客户交付E2E",
|
||
"cost_type": "labor", "item_name": "实施工时E2E",
|
||
"standard_quantity": 5.0, "standard_price": 60.0, "unit": "小时",
|
||
"remark": MARK,
|
||
}, auth)
|
||
check("POST standard 默认client_delivery 200", st == 200, f"http={st} body={r}")
|
||
|
||
# ── C. 列表过滤: project_type=platform_dev ──
|
||
st, r = http("GET", f"/api/cma/cost/standard-costs?project_type=platform_dev", None, auth)
|
||
codes = [i["product_code"] for i in r.get("data", [])]
|
||
check("GET standard filter=platform_dev 200", st == 200, f"http={st}")
|
||
check("研发过滤含研发行", rd_code in codes, f"{rd_code} in {codes}")
|
||
check("研发过滤不含客户交付行", cd_code not in codes, f"{cd_code} not in {codes}")
|
||
|
||
# ── D. 列表过滤: research_flag=1 ──
|
||
st, r = http("GET", "/api/cma/cost/standard-costs?research_flag=1", None, auth)
|
||
codes_rf = [i["product_code"] for i in r.get("data", [])]
|
||
check("GET standard filter=research_flag=1 含研发行", rd_code in codes_rf,
|
||
f"{rd_code} in {codes_rf}")
|
||
|
||
# ── E. 实际成本: platform_dev ──
|
||
st, r = http("POST", "/api/cma/cost/actual-costs", {
|
||
"period": "2026-09", "product_code": ac_code, "product_name": "CMA平台化开发E2E",
|
||
"cost_type": "labor", "item_name": "研发工时E2E",
|
||
"actual_quantity": 40.0, "actual_price": 100.0,
|
||
"project_type": "platform_dev", "remark": MARK,
|
||
}, auth)
|
||
check("POST actual platform_dev 200", st == 200, f"http={st} body={r}")
|
||
st, r = http("GET", "/api/cma/cost/actual-costs?project_type=platform_dev&period=2026-09", None, auth)
|
||
ac_codes = [i["product_code"] for i in r.get("data", [])]
|
||
check("GET actual filter=platform_dev 含研发行", ac_code in ac_codes, f"{ac_code} in {ac_codes}")
|
||
|
||
# ── DB 落库校验 ──
|
||
db = next(get_db())
|
||
try:
|
||
row_s = db.execute(text(
|
||
"SELECT project_type, research_flag, entity_id FROM standard_costs WHERE product_code = :c"
|
||
), {"c": rd_code}).fetchone()
|
||
check("DB standard 落库 project_type=platform_dev", row_s and row_s[0] == "platform_dev",
|
||
f"row={row_s}")
|
||
check("DB standard research_flag 联动=1", row_s and row_s[1] == 1, f"row={row_s}")
|
||
check("DB standard entity_id=1(token绑定)", row_s and row_s[2] == 1, f"row={row_s}")
|
||
|
||
row_c = db.execute(text(
|
||
"SELECT project_type, research_flag FROM standard_costs WHERE product_code = :c"
|
||
), {"c": cd_code}).fetchone()
|
||
check("DB standard 默认 project_type=client_delivery", row_c and row_c[0] == "client_delivery",
|
||
f"row={row_c}")
|
||
check("DB standard 默认 research_flag=0", row_c and row_c[1] == 0, f"row={row_c}")
|
||
|
||
row_a = db.execute(text(
|
||
"SELECT project_type, research_flag FROM actual_costs WHERE product_code = :c"
|
||
), {"c": ac_code}).fetchone()
|
||
check("DB actual 落库 platform_dev+flag=1", row_a and row_a[0] == "platform_dev" and row_a[1] == 1,
|
||
f"row={row_a}")
|
||
|
||
# ── F. 存量保守默认: internal 行仍在(ALTER后存量 internal, 未误归研发) ──
|
||
n_std_internal = db.execute(text(
|
||
"SELECT COUNT(*) FROM standard_costs WHERE project_type = 'internal' AND product_code NOT LIKE :p"
|
||
), {"p": f"{PREFIX}%"}).scalar()
|
||
n_act_internal = db.execute(text(
|
||
"SELECT COUNT(*) FROM actual_costs WHERE project_type = 'internal' AND product_code NOT LIKE :p"
|
||
), {"p": f"{PREFIX}%"}).scalar()
|
||
check("存量 standard internal 保留(29)", n_std_internal >= 29, f"count={n_std_internal}")
|
||
check("存量 actual internal 保留(134)", n_act_internal >= 134, f"count={n_act_internal}")
|
||
|
||
# ── H. ORM schema: 列+索引存在 ──
|
||
insp = sa_inspect(get_engine())
|
||
for t in ("standard_costs", "actual_costs"):
|
||
cols = {c["name"] for c in insp.get_columns(t)}
|
||
idxs = {ix["name"] for ix in insp.get_indexes(t)}
|
||
check(f"{t} 含 research_flag+project_type 列",
|
||
{"research_flag", "project_type"} <= cols, f"cols={sorted(cols)}")
|
||
check(f"{t} 含两索引",
|
||
{f"ix_{t}_research_flag", f"ix_{t}_project_type"} <= idxs, f"idx={sorted(idxs)}")
|
||
finally:
|
||
db.rollback()
|
||
db.close()
|
||
|
||
# ── G. 回归: 原列表接口正常 ──
|
||
st, r = http("GET", "/api/cma/cost/standard-costs", None, auth)
|
||
check("回归 GET standard-costs(全量) 200+list", st == 200 and isinstance(r.get("data"), list),
|
||
f"http={st} n={len(r.get('data', []))}")
|
||
st, r = http("GET", "/api/cma/cost/actual-costs?period=2026-09", None, auth)
|
||
check("回归 GET actual-costs 200+list", st == 200 and isinstance(r.get("data"), list),
|
||
f"http={st} n={len(r.get('data', []))}")
|
||
|
||
finally:
|
||
# ── 清理临时数据(DB直删标记行) + 回查 ──
|
||
db = next(get_db())
|
||
try:
|
||
for t in ("standard_costs", "actual_costs"):
|
||
db.execute(text(f"DELETE FROM {t} WHERE product_code LIKE :p"), {"p": f"{PREFIX}%"})
|
||
db.commit()
|
||
for t in ("standard_costs", "actual_costs"):
|
||
n = db.execute(text(f"SELECT COUNT(*) FROM {t} WHERE product_code LIKE :p"),
|
||
{"p": f"{PREFIX}%"}).scalar()
|
||
check(f"清理回查 {t} 标记行=0", n == 0, f"count={n}")
|
||
finally:
|
||
db.rollback()
|
||
db.close()
|
||
|
||
print()
|
||
if failures:
|
||
print(f"RESULT: FAIL ({len(failures)}): {failures}")
|
||
sys.exit(1)
|
||
print("RESULT: ALL PASS")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|