Hermes CI Fix
5e0b09e6b6
- 表名: ORM+生产库 RENAME cash_flow_structure_values → cashflow_statements (DoD SHOW TABLES LIKE 字面验收)
- 兼容端点: /api/cma/cashflow/statements(GET/POST)+analysis 注册 main.py (DoD 验收路径)
- 菜单: 移入 🟡C 监控与评价 组(与杜邦分析/管理报表平级), roles 收敛 ceo/finance (前端 ROLE_ROUTES/MENU_ITEMS/router meta + 后端 require_role 四处同步)
- 术语: docstring 改写, grep 现金流肖像=0
- 验收脚本: backend/scripts/verify_cashflow_structure.py (14项断言全PASS: type6→type1双向/瀑布5步/alias路径/权限/清理)
154 lines
6.6 KiB
Python
154 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
现金流量表结构分析 — 验收验证脚本 (project-cashflow-structure-analysis-20260903 DoD)
|
|
|
|
覆盖:
|
|
1. POST /api/cma/cashflow-structure/values 录入 O=-30/I=-5/F=+40 → 期望类型6(初创烧钱型🟠)
|
|
2. 同期间覆盖更新 O=+30/I=-20/F=-10 → 期望类型1(成熟健康型🟢) 判定双向验证
|
|
3. GET /analysis 结构分析响应(三净额符号/判定/瀑布图数据点)
|
|
4. GET /values /periods /rules 常规查询
|
|
5. 清理验证临时数据(DB 直删 remark 标记行) → 恢复空态
|
|
|
|
用法: cd backend && venv/bin/python3 scripts/verify_cashflow_structure.py
|
|
依赖: requests (项目 venv 已装)
|
|
"""
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = "http://127.0.0.1:8010"
|
|
TMP_MARK = "【系统验证临时数据-验证后删除】"
|
|
|
|
|
|
def http(method, path, body=None, token=None, timeout=15):
|
|
url = BASE + path
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
if token:
|
|
req.add_header("Authorization", "Bearer " + token)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
raw = resp.read().decode()
|
|
return resp.status, json.loads(raw) if raw else None
|
|
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():
|
|
fails = []
|
|
|
|
def check(name, cond, extra=""):
|
|
tag = "✅ PASS" if cond else "❌ FAIL"
|
|
print(f"{tag} {name}" + (f" {extra}" if extra else ""))
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
# 1. 登录博海账套 (admin, entity_id=2)
|
|
st, r = http("POST", "/api/cma/auth/login",
|
|
{"username": "admin", "password": "admin123", "entity_id": 2})
|
|
check("登录博海(entity_id=2)", st == 200 and "token" in (r or {}),
|
|
f"st={st} keys={sorted((r or {}).keys())}")
|
|
token = (r or {}).get("token", "")
|
|
entity_name = (r or {}).get("entity_name", "")
|
|
print(f" 登录账套: {entity_name}")
|
|
|
|
# 2. openapi 端点可见性
|
|
st, spec = http("GET", "/openapi.json")
|
|
paths = [p for p in (spec or {}).get("paths", {}) if "cashflow-structure" in p]
|
|
check("openapi 含 cashflow-structure 端点", st == 200 and len(paths) >= 4,
|
|
f"paths={sorted(paths)}")
|
|
|
|
# 3. 空态分析(清理后应先跑一次, 这里先录入再测非空)
|
|
# 4. POST 录入 类型6 用例: O=-30 I=-5 F=+40
|
|
st, r = http("POST", "/api/cma/cashflow-structure/values", {
|
|
"period": "2026-08",
|
|
"operating_net": -30, "investing_net": -5, "financing_net": 40,
|
|
"begin_cash": 10, "end_cash": 15,
|
|
"source": "manual", "remark": TMP_MARK,
|
|
}, token=token)
|
|
cls = (r or {}).get("classification", {})
|
|
check("POST values (O-30/I-5/F+40) → type6", st == 200 and cls.get("type_no") == 6,
|
|
f"st={st} cls={cls.get('type_no')} {cls.get('name')} {cls.get('health_emoji')}")
|
|
|
|
# 5. GET analysis 验证
|
|
st, r = http("GET", "/api/cma/cashflow-structure/analysis?period=2026-08", token=token)
|
|
check("GET analysis 200", st == 200 and (r or {}).get("data_available") is True)
|
|
if r:
|
|
wf = r.get("waterfall") or {}
|
|
steps = wf.get("steps") or []
|
|
check("瀑布图数据点≥5(期初→经营→投资→筹资→期末)", len(steps) >= 5,
|
|
f"steps={[s['step'] for s in steps]}")
|
|
check("判定 reason 含符号组合", "经营(-)" in ((r.get("classification") or {}).get("reason") or ""))
|
|
|
|
# 6. 覆盖更新 类型1 用例: O=+30 I=-20 F=-10 (同期间 upsert)
|
|
st, r = http("POST", "/api/cma/cashflow-structure/values", {
|
|
"period": "2026-08",
|
|
"operating_net": 30, "investing_net": -20, "financing_net": -10,
|
|
"source": "manual", "remark": TMP_MARK,
|
|
}, token=token)
|
|
cls = (r or {}).get("classification", {})
|
|
check("POST 覆盖更新 (O+30/I-20/F-10) → type1", st == 200 and cls.get("type_no") == 1,
|
|
f"st={st} cls={cls.get('type_no')} {cls.get('name')} {cls.get('health_emoji')}")
|
|
|
|
# 7. GET values/periods 常规查询
|
|
st, r = http("GET", "/api/cma/cashflow-structure/values", token=token)
|
|
items = (r or {}).get("items", [])
|
|
check("GET values 返回记录", st == 200 and len(items) >= 1, f"count={len(items)}")
|
|
st, r = http("GET", "/api/cma/cashflow-structure/periods", token=token)
|
|
check("GET periods 含 2026-08", st == 200 and "2026-08" in (r or {}).get("periods", []))
|
|
st, r = http("GET", "/api/cma/cashflow-structure/rules", token=token)
|
|
check("GET rules 8类", st == 200 and len((r or {}).get("rules", [])) == 8)
|
|
|
|
# 7b. DoD 验收路径 /api/cma/cashflow/* (兼容别名, 功能与主路径一致)
|
|
st, r = http("GET", "/api/cma/cashflow/statements?period=2026-08", token=token)
|
|
items_c = (r or {}).get("items", [])
|
|
check("GET /api/cma/cashflow/statements 200", st == 200 and len(items_c) == 1,
|
|
f"count={len(items_c)}")
|
|
st, r = http("GET", "/api/cma/cashflow/analysis?period=2026-08", token=token)
|
|
cls_c = (r or {}).get("classification", {})
|
|
check("GET /api/cma/cashflow/analysis 200 type1", st == 200 and cls_c.get("type_no") == 1,
|
|
f"cls={cls_c.get('type_no')} {cls_c.get('name')}")
|
|
|
|
# 8. 越权: 无 token 调 analysis 应 401/403 (require_role 拦截)
|
|
st, _ = http("GET", "/api/cma/cashflow-structure/analysis")
|
|
check("无token访问被拦截(401/403)", st in (401, 403), f"st={st}")
|
|
|
|
print("\n===== 判定验证输出 =====")
|
|
print("用例1: O=-30/I=-5/F=+40 →", "type6 初创烧钱型 🟠 高风险" if not fails or "POST values" not in "".join(fails) else "待查")
|
|
print("用例2: O=+30/I=-20/F=-10 →", "type1 成熟健康型 🟢 最优")
|
|
|
|
print("\n清理验证临时数据...")
|
|
try:
|
|
sys.path.insert(0, ".")
|
|
from app.database import get_db
|
|
from app.models import CashFlowStructureValue
|
|
db = next(get_db())
|
|
rows = db.query(CashFlowStructureValue).filter(
|
|
CashFlowStructureValue.remark == TMP_MARK).all()
|
|
for row in rows:
|
|
db.delete(row)
|
|
db.commit()
|
|
remain = db.query(CashFlowStructureValue).count()
|
|
db.close()
|
|
check("临时数据已清理(表余0)", remain == 0, f"remain={remain}")
|
|
except Exception as e:
|
|
check(f"清理临时数据失败: {e}", False)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"❌ 共 {len(fails)} 项失败: {fails}")
|
|
return 1
|
|
print("✅ 全部验收项通过")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|