test(bot-tier): M4 生产E2E实证脚本 verify_bot_write_tiering_e2e.py (P0-②)
全PASS: Bot建议→unapplied→人apply→目标生效+apply_detail before/after→低危kpi-result source_batch落库→audit L2留痕→清理回查=0
This commit is contained in:
Hermes CI Fix
@@ -0,0 +1,219 @@
|
||||
#!/usr/bin/env python3
|
||||
"""生产E2E实证: Bot写操作分级+分层授权 (P0-② 2026-09-03)
|
||||
|
||||
链路验证:
|
||||
A. Bot通道(X-BOT-KEY)创建 kpi_target 高危建议 → status=unapplied
|
||||
B. 人token apply → KPI目标值变更 + apply_detail(before→after) + applied_by/applied_at
|
||||
C. 低危直写 bot-bridge /kpi-result (X-BRIDGE-TOKEN) → KPIValue.source_batch 落库
|
||||
D. bot_audit.log 留痕(两条: 建议创建L2 + kpi-result L2)
|
||||
|
||||
数据纪律: 临时KPI/建议/值 全部带【系统验证临时数据-验证后删除】标记,
|
||||
验证完成后精确清理并回查 COUNT=0。审计日志文件行保留(留痕证据)。
|
||||
|
||||
用法: cd /root/cma-management/backend && venv/bin/python3 scripts/verify_bot_write_tiering_e2e.py
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
|
||||
BASE = "http://127.0.0.1:8010"
|
||||
MARK = "【系统验证临时数据-验证后删除】"
|
||||
KPI_CODE = "E2E_WT_TIER"
|
||||
SUG_PREFIX = "E2E-验证"
|
||||
CUR_PERIOD = "2026-09"
|
||||
BASELINE_TARGET = 100.0
|
||||
NEW_TARGET = 888.88
|
||||
LOW_VALUE = 777.0
|
||||
SOURCE = "e2e-bot-tier-20260903"
|
||||
|
||||
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 # noqa: E402
|
||||
from app.models import ( # noqa: E402
|
||||
BotBridgeConfig, KPIDefinition, KPIValue, AISuggestion, OperationLog, Entity,
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
return e.code, json.loads(e.read().decode())
|
||||
|
||||
|
||||
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. 取Bot密钥与bridge token ──
|
||||
with open(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env")) as f:
|
||||
env = {}
|
||||
for line in f:
|
||||
if "=" in line and not line.strip().startswith("#"):
|
||||
k, _, v = line.strip().partition("=")
|
||||
env[k] = v.strip().strip('"').strip("'")
|
||||
keys = json.loads(env["CMA_BOT_API_KEYS"])
|
||||
fin_key = next(k for k, v in keys.items() if v.get("name") == "财务BOT")
|
||||
|
||||
db = next(get_db())
|
||||
bridge = db.query(BotBridgeConfig).filter_by(is_active=True).first()
|
||||
bridge_token = bridge.token if bridge else None
|
||||
db.close()
|
||||
|
||||
# ── 1. 登录(人token, entity=1) ──
|
||||
st, r = http("POST", "/api/cma/auth/login",
|
||||
{"username": "admin", "password": "admin123", "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}"}
|
||||
|
||||
# ── 2. 临时KPI准备(存在则复用, 目标值重置基线) ──
|
||||
db = next(get_db())
|
||||
db.rollback()
|
||||
kpi = db.query(KPIDefinition).filter(
|
||||
KPIDefinition.kpi_code == KPI_CODE,
|
||||
KPIDefinition.entity_id == 1,
|
||||
).first()
|
||||
if not kpi:
|
||||
kpi = KPIDefinition(
|
||||
entity_id=1, kpi_code=KPI_CODE,
|
||||
kpi_name=f"{MARK}写分层E2E临时KPI",
|
||||
dimension="finance", category="financial_report", formula="-",
|
||||
data_source_type="manual", data_source=f"{MARK}E2E", data_owner="系统验证",
|
||||
frequency="monthly", unit="万元", target_value=BASELINE_TARGET,
|
||||
kpi_level="operational", status="active",
|
||||
)
|
||||
db.add(kpi)
|
||||
db.commit()
|
||||
db.refresh(kpi)
|
||||
kpi_id = kpi.id
|
||||
db.close()
|
||||
else:
|
||||
kpi_id = kpi.id
|
||||
kpi.target_value = BASELINE_TARGET
|
||||
db.commit()
|
||||
db.refresh(kpi)
|
||||
db.close()
|
||||
print(f" 临时KPI id={kpi_id} code={KPI_CODE} baseline={BASELINE_TARGET}")
|
||||
|
||||
# ── 3. Bot创建高危建议(kpi_target) ──
|
||||
sug_title = f"{SUG_PREFIX} 上调临时KPI目标→{NEW_TARGET}"
|
||||
st, r = http("POST", "/api/cma/bot/ai/suggestions", {
|
||||
"suggestion_type": "kpi_target", "title": sug_title, "content": "分层授权E2E实证",
|
||||
"source": "bot", "entity_id": 1, "target_id": kpi_id,
|
||||
"suggestion_data": {"kpi_id": kpi_id, "target_value": NEW_TARGET},
|
||||
}, {"X-BOT-KEY": fin_key})
|
||||
check("A1 Bot创建建议 200", st == 200, f"http={st} body={str(r)[:200]}")
|
||||
sug_id = r.get("data", {}).get("id") if st == 200 else None
|
||||
check("A2 status=unapplied", st == 200 and r["data"]["status"] == "unapplied",
|
||||
f"id={sug_id} source={r.get('data', {}).get('source')}")
|
||||
|
||||
# ── 4. 人apply ──
|
||||
st, r = http("POST", f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||||
{"action": "kpi_target", "target_value": NEW_TARGET}, auth)
|
||||
check("B1 apply 200", st == 200, f"http={st} body={str(r)[:200]}")
|
||||
ad = r.get("data", {}) if st == 200 else {}
|
||||
detail = (ad.get("apply_detail") or [{}])[0]
|
||||
check("B2 apply_detail before→after", detail.get("before") == BASELINE_TARGET
|
||||
and detail.get("after") == NEW_TARGET, str(detail)[:160])
|
||||
check("B3 applied_by/applied_at", bool(ad.get("applied_by")) and bool(ad.get("applied_at")),
|
||||
f"by={ad.get('applied_by')} at={ad.get('applied_at')}")
|
||||
|
||||
db = next(get_db())
|
||||
db.rollback()
|
||||
kpi2 = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
|
||||
db_target = getattr(kpi2, "target_value", None) if kpi2 else None
|
||||
check("B4 KPI目标值生效", db_target == NEW_TARGET,
|
||||
f"db_target={db_target}")
|
||||
db.close()
|
||||
|
||||
# ── 5. 低危直写 /kpi-result ──
|
||||
st, r = http("POST", "/api/cma/bot-bridge/kpi-result", {
|
||||
"entity_id": 1, "kpi_code": KPI_CODE, "period": CUR_PERIOD,
|
||||
"value": LOW_VALUE, "source": SOURCE, "remark": MARK,
|
||||
}, {"X-BRIDGE-TOKEN": bridge_token})
|
||||
check("C1 kpi-result 200", st == 200, f"http={st} body={str(r)[:200]}")
|
||||
|
||||
db = next(get_db())
|
||||
db.rollback()
|
||||
val = db.query(KPIValue).filter(
|
||||
KPIValue.kpi_id == kpi_id, KPIValue.period == CUR_PERIOD,
|
||||
).order_by(KPIValue.id.desc()).first()
|
||||
check("C2 source_batch落库", val is not None and val.source_batch == SOURCE,
|
||||
f"source_batch={val.source_batch if val else None} actual={val.actual_value if val else None}")
|
||||
db.close()
|
||||
|
||||
# ── 6. audit留痕 ──
|
||||
audit_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"logs", "bot_audit.log")
|
||||
lines = []
|
||||
if os.path.exists(audit_path):
|
||||
with open(audit_path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
audit_str = "".join(lines[-60:])
|
||||
check("D1 audit含建议创建记录",
|
||||
"/api/cma/bot/ai/suggestions" in audit_str and '"status": 200' in audit_str,
|
||||
f"sug_id={sug_id}")
|
||||
check("D2 audit含kpi-result记录",
|
||||
"/api/cma/bot-bridge/kpi-result" in audit_str and '"status": 200' in audit_str)
|
||||
# 取最近一条(倒序)建议创建成功记录验证字段；早于重启的401是pytest进程写文件的噪音
|
||||
rec = None
|
||||
for l in reversed(lines[-60:]):
|
||||
try:
|
||||
r = json.loads(l)
|
||||
except Exception:
|
||||
continue
|
||||
if r.get("endpoint") == "/api/cma/bot/ai/suggestions" and r.get("status") == 200:
|
||||
rec = r
|
||||
break
|
||||
if rec:
|
||||
check("D3 建议创建audit字段(bot_name/L2)", rec.get("bot_name") == "财务BOT"
|
||||
and rec.get("risk_level") == "L2", str(rec)[:200])
|
||||
|
||||
# ── 7. 清理(标记精确删除) + 回查 ──
|
||||
db = next(get_db())
|
||||
db.rollback()
|
||||
del_sug = db.query(AISuggestion).filter(
|
||||
AISuggestion.title.like(f"{SUG_PREFIX}%"), AISuggestion.title.like("%上调临时KPI目标%"),
|
||||
).delete(synchronize_session=False)
|
||||
del_val = db.query(KPIValue).filter(
|
||||
KPIValue.kpi_id == kpi_id, KPIValue.source_batch == SOURCE,
|
||||
).delete(synchronize_session=False)
|
||||
db.query(OperationLog).filter(OperationLog.action == "ai_suggestion_apply",
|
||||
OperationLog.target_id == kpi_id).delete(synchronize_session=False)
|
||||
del_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).delete(synchronize_session=False)
|
||||
db.commit()
|
||||
db.rollback()
|
||||
left_sug = db.query(AISuggestion).filter(
|
||||
AISuggestion.title.like(f"{SUG_PREFIX}%"), AISuggestion.title.like("%上调临时KPI目标%"),
|
||||
).count()
|
||||
left_val = db.query(KPIValue).filter(
|
||||
KPIValue.kpi_id == kpi_id, KPIValue.source_batch == SOURCE,
|
||||
).count()
|
||||
left_kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).count()
|
||||
check("E1 清理回查=0", left_sug == 0 and left_val == 0 and left_kpi == 0,
|
||||
f"sug={left_sug} val={left_val} kpi={left_kpi}")
|
||||
db.close()
|
||||
|
||||
print()
|
||||
if failures:
|
||||
print(f"E2E RESULT: FAIL ({len(failures)}): {failures}")
|
||||
sys.exit(1)
|
||||
print(f"E2E RESULT: PASS | 建议id={sug_id}(已清理) kpi_id={kpi_id}(已清理) 审计行保留于 logs/bot_audit.log")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -67,4 +67,8 @@
|
||||
|
||||
## 七、验收记录（E2E/回归）
|
||||
|
||||
见任务完成回报与 backend/tests/test_bot_suggestion_channel.py（新增 pytest 回归覆盖）。
|
||||
- 回归: backend/tests/test_bot_suggestion_channel.py（12条：鉴权/语义/多租户/幂等/审批闭环/审计/收口）；test_risk_levels.py 计数更新 L2=6。
|
||||
- 生产E2E: backend/scripts/verify_bot_write_tiering_e2e.py（可复跑，标记+清理+回查）2026-09-03 全 PASS：
|
||||
A Bot创建kpi_target建议→unapplied → B 人apply→KPI目标100→888.88 + apply_detail(before/after) + applied_by=管理员/applied_at → C /kpi-result 直写 source_batch=e2e-bot-tier-20260903 落库 → D bot_audit.log 两条 L2 记录（财务BOT / bot-bridge）→ E 清理回查=0。
|
||||
- pytest 全量: 726 passed / 4 failed(日历漂移预存：budget roll_forward + ai_suggestions budget_overrun + roadmap budget_headroom×2，非本次回归) / 1 xfailed。
|
||||
- 后端重启: systemctl restart cma-backend（ActiveEnter 晚于 commit）; /health ok。
|
||||
|
||||
Reference in New Issue
Block a user