Hermes CI Fix
e5dd6e380b
- 后端: cashflow_structure.py 模块 (GET rules/periods/values/analysis + POST values upsert) - 判定: O/I/F 三净额符号 → 8类企业定性 (CMA P2 现金流量表分析, 无'现金流肖像'字样) - 数据: cash_flow_structure_values 表 (entity_id+period 唯一), 官方现金流量表净额按期间注入 - 前端: CashFlowStructure.vue 判定横幅/瀑布图/规则表/入库记录 + MyDialog 录入弹窗 - 菜单: 财务报表分析 → 现金流量表分析 (路由 /cashflow-structure, 角色 ceo/finance/it) - 测试: tests/test_cashflow_structure.py 7 条通过 (规则8类/判定/覆盖/博海边界6-7/非法期间)
108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
"""现金流量表结构分析模块测试 — 八类企业定性判定 (P1 2026-09-03)
|
|
|
|
覆盖 cashflow_structure.py:
|
|
GET /api/cma/cashflow-structure/rules 八类规则表
|
|
GET /api/cma/cashflow-structure/periods 已注入期间
|
|
GET /api/cma/cashflow-structure/analysis 结构分析+类型判定+瀑布数据
|
|
POST /api/cma/cashflow-structure/values 三大活动净额注入(upsert)
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
|
|
|
BASE = "/api/cma/cashflow-structure"
|
|
|
|
|
|
class TestRules:
|
|
def test_rules_8_types(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = client.get(f"{BASE}/rules", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
rules = resp.json()["rules"]
|
|
assert len(rules) == 8
|
|
# 判定组合必须恰好 8 种且无重复
|
|
combos = [(r["o_sign"], r["i_sign"], r["f_sign"]) for r in rules]
|
|
assert len(set(combos)) == 8
|
|
# 规范命名：不含民间叫法
|
|
assert all("现金流肖像" not in r["name"] for r in rules)
|
|
|
|
def test_rules_unauthorized(self, client: TestClient):
|
|
resp = client.get(f"{BASE}/rules")
|
|
assert resp.status_code in (401, 403)
|
|
|
|
|
|
class TestStructureAnalysis:
|
|
def _post_value(self, client, token, **kw):
|
|
payload = {
|
|
"period": "2026-08",
|
|
"operating_net": -10.0,
|
|
"investing_net": 3.0,
|
|
"financing_net": 12.0, # - + + → 类型7 输血生存型
|
|
"source": "official",
|
|
"remark": "pytest 临时数据",
|
|
}
|
|
payload.update(kw)
|
|
return client.post(f"{BASE}/values", headers=auth_header(token), json=payload)
|
|
|
|
def test_empty_analysis(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = client.get(f"{BASE}/analysis", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["data_available"] is False
|
|
assert body["classification"] is None
|
|
|
|
def test_upsert_and_classify_type7(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = self._post_value(client, token)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["ok"] is True
|
|
assert body["classification"]["classified"] is True
|
|
assert body["classification"]["type_no"] == 7
|
|
assert body["classification"]["name"] == "输血生存型"
|
|
|
|
# 分析接口自动命中
|
|
r2 = client.get(f"{BASE}/analysis", headers=auth_header(token), params={"period": "2026-08"})
|
|
assert r2.status_code == 200
|
|
a = r2.json()
|
|
assert a["data_available"] is True
|
|
assert a["classification"]["type_no"] == 7
|
|
assert a["nets"]["operating"]["sign"] == "-"
|
|
assert a["nets"]["investing"]["sign"] == "+"
|
|
assert a["nets"]["financing"]["sign"] == "+"
|
|
|
|
# periods / values 可查
|
|
r3 = client.get(f"{BASE}/periods", headers=auth_header(token))
|
|
assert "2026-08" in r3.json()["periods"]
|
|
|
|
def test_upsert_overwrite_same_period(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
self._post_value(client, token)
|
|
resp2 = self._post_value(client, token, operating_net=50.0, investing_net=-20.0, financing_net=-10.0)
|
|
assert resp2.status_code == 200
|
|
assert resp2.json()["classification"]["type_no"] == 1 # + - - → 成熟健康型
|
|
rows = client.get(f"{BASE}/values", headers=auth_header(token)).json()["items"]
|
|
assert len(rows) == 1 # 同期间覆盖不重复插入
|
|
|
|
def test_type6_7_boundary_like_bohai(self, client: TestClient, db: Session):
|
|
"""博海结构判定回归：经营弱/负 + 筹资依赖 → 类型6(- - +) / 类型7(- + +)"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = self._post_value(client, token, operating_net=-5.0, investing_net=-2.0, financing_net=9.0)
|
|
assert resp.json()["classification"]["type_no"] == 6
|
|
resp = self._post_value(client, token, operating_net=-5.0, investing_net=2.0, financing_net=9.0)
|
|
assert resp.json()["classification"]["type_no"] == 7
|
|
|
|
def test_invalid_period(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = self._post_value(client, token, period="2026/08")
|
|
assert resp.status_code == 422
|