Hermes CI Fix
d20a270ba2
覆盖： - 进销项倒挂/应交税费异常/账面毛利异常/收入vs现金/收入vs存货/人员vs收入/成本vs收入 - 七流健康评分(绿100/黄60/红0) - 酣客场景: 触发>=3项红色(进销项倒挂/应交429万/毛利-0.43%)+评分<60 - 博海场景: 无红色+评分>=60(IT经销13%毛利合理) 27 passed — 为项目Bot开发七流自检API提供规则基线
244 lines
9.1 KiB
Python
244 lines
9.1 KiB
Python
"""七流一致自检规则测试 — 金税四期七流框架
|
||
|
||
覆盖 /root/cma_seven_flow_design_20260825.md 的7条自检规则：
|
||
1. 进销项倒挂
|
||
2. 应交税费异常
|
||
3. 账面毛利异常
|
||
4. 收入vs现金
|
||
5. 收入vs存货
|
||
6. 人员vs收入
|
||
7. 成本vs收入
|
||
|
||
规则是纯函数（不依赖DB），测试直接验证判定逻辑。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
|
||
# ── 七流自检规则引擎（与设计方案一致） ──
|
||
|
||
def rule_invoice_inverted(input_tax, output_tax):
|
||
"""规则1：进项税额 > 销项税额×1.2 → 倒挂风险"""
|
||
if output_tax <= 0:
|
||
return None
|
||
ratio = input_tax / output_tax
|
||
return {"passed": ratio <= 1.2, "value": round(ratio, 2),
|
||
"threshold": 1.2, "level": "red" if ratio > 1.2 else "green",
|
||
"message": f"进项/销项比 {ratio:.2f}（阈值1.2）"}
|
||
|
||
|
||
def rule_tax_payable_ratio(tax_payable, income):
|
||
"""规则2：应交税费/收入 > 20% → 异常"""
|
||
if income <= 0:
|
||
return None
|
||
ratio = tax_payable / income
|
||
return {"passed": ratio <= 0.2, "value": round(ratio * 100, 1),
|
||
"threshold": 20.0, "level": "red" if ratio > 0.2 else "green",
|
||
"message": f"应交税费/收入 {ratio*100:.1f}%（阈值20%）"}
|
||
|
||
|
||
def rule_margin_abnormal(margin_rate, industry_base=10.0):
|
||
"""规则3：毛利率 < 行业基准×0.5 → 异常（需解释链）"""
|
||
threshold = industry_base * 0.5
|
||
return {"passed": margin_rate >= threshold, "value": round(margin_rate, 2),
|
||
"threshold": threshold, "level": "red" if margin_rate < threshold else "green",
|
||
"message": f"毛利率 {margin_rate}%（行业基准{industry_base}%×0.5={threshold}%）"}
|
||
|
||
|
||
def rule_income_vs_cash(cashflow, revenue_growth):
|
||
"""规则4：经营现金流为负 且 收入负增长 → 预警"""
|
||
risky = (cashflow < 0) and (revenue_growth < 0)
|
||
return {"passed": not risky, "value": f"现金流{cashflow}, 增长{revenue_growth}%",
|
||
"level": "red" if risky else "green",
|
||
"message": "现金流负+收入负增长" if risky else "现金流/收入正常"}
|
||
|
||
|
||
def rule_income_vs_inventory(revenue_growth, inventory_change):
|
||
"""规则5：收入降>20% 但 存货不降 → 预警"""
|
||
risky = (revenue_growth < -20) and (inventory_change >= 0)
|
||
return {"passed": not risky, "value": f"收入{revenue_growth}%, 存货{inventory_change}%",
|
||
"level": "yellow" if risky else "green",
|
||
"message": "收入下滑但存货未降" if risky else "收入/存货匹配"}
|
||
|
||
|
||
def rule_personnel_vs_income(revenue_growth, employee_change):
|
||
"""规则6：收入降>20% 但 人员不变 → 预警"""
|
||
risky = (revenue_growth < -20) and (employee_change == 0)
|
||
return {"passed": not risky, "value": f"收入{revenue_growth}%, 人员{employee_change}%",
|
||
"level": "yellow" if risky else "green",
|
||
"message": "收入下滑但人员未变" if risky else "收入/人员匹配"}
|
||
|
||
|
||
def rule_cost_vs_income(cost_rate):
|
||
"""规则7：成本率 > 100% → 异常（需厂补/渠补解释）"""
|
||
return {"passed": cost_rate <= 100, "value": round(cost_rate, 1),
|
||
"threshold": 100.0, "level": "red" if cost_rate > 100 else "green",
|
||
"message": f"成本率 {cost_rate:.1f}%（阈值100%）"}
|
||
|
||
|
||
def seven_flow_score(checks):
|
||
"""七流健康评分：green=100, yellow=60, red=0，取平均"""
|
||
scores = {"green": 100, "yellow": 60, "red": 0}
|
||
if not checks:
|
||
return 0
|
||
return round(sum(scores.get(c["level"], 50) for c in checks) / len(checks))
|
||
|
||
|
||
# ── 测试类 ──
|
||
|
||
class TestRuleInvoiceInverted:
|
||
def test_normal(self):
|
||
r = rule_invoice_inverted(100, 120)
|
||
assert r["passed"] is True
|
||
assert r["level"] == "green"
|
||
|
||
def test_inverted(self):
|
||
r = rule_invoice_inverted(200, 120) # 1.67 > 1.2
|
||
assert r["passed"] is False
|
||
assert r["level"] == "red"
|
||
|
||
def test_boundary(self):
|
||
r = rule_invoice_inverted(120, 100) # 1.2 = 阈值
|
||
assert r["passed"] is True
|
||
|
||
def test_zero_output(self):
|
||
assert rule_invoice_inverted(100, 0) is None
|
||
|
||
|
||
class TestRuleTaxPayableRatio:
|
||
def test_normal(self):
|
||
r = rule_tax_payable_ratio(10, 100) # 10%
|
||
assert r["passed"] is True
|
||
|
||
def test_abnormal(self):
|
||
r = rule_tax_payable_ratio(429, 100) # 429% (酣客场景)
|
||
assert r["passed"] is False
|
||
assert r["level"] == "red"
|
||
|
||
def test_zero_income(self):
|
||
assert rule_tax_payable_ratio(10, 0) is None
|
||
|
||
|
||
class TestRuleMarginAbnormal:
|
||
def test_normal_margin(self):
|
||
r = rule_margin_abnormal(13.0, 10.0) # 博海IT经销13%
|
||
assert r["passed"] is True
|
||
|
||
def test_abnormal_margin(self):
|
||
r = rule_margin_abnormal(-0.43, 10.0) # 酣客账面-0.43%
|
||
assert r["passed"] is False
|
||
assert r["level"] == "red"
|
||
|
||
def test_threshold_boundary(self):
|
||
r = rule_margin_abnormal(5.0, 10.0) # = 5 = 10×0.5
|
||
assert r["passed"] is True
|
||
|
||
|
||
class TestRuleIncomeVsCash:
|
||
def test_both_negative(self):
|
||
r = rule_income_vs_cash(-13.89, -37.0) # 酣客：现金流负+收入负增长
|
||
assert r["passed"] is False
|
||
assert r["level"] == "red"
|
||
|
||
def test_positive_cash(self):
|
||
r = rule_income_vs_cash(33.86, -37.0)
|
||
assert r["passed"] is True
|
||
|
||
def test_negative_cash_positive_growth(self):
|
||
r = rule_income_vs_cash(-5, 10)
|
||
assert r["passed"] is True
|
||
|
||
|
||
class TestRuleIncomeVsInventory:
|
||
def test_income_down_inventory_up(self):
|
||
r = rule_income_vs_inventory(-37, 5) # 收入降37%存货增5%
|
||
assert r["passed"] is False
|
||
assert r["level"] == "yellow"
|
||
|
||
def test_income_down_inventory_down(self):
|
||
r = rule_income_vs_inventory(-37, -2) # 都降
|
||
assert r["passed"] is True
|
||
|
||
def test_slight_income_drop(self):
|
||
r = rule_income_vs_inventory(-10, 5) # 降幅<20%
|
||
assert r["passed"] is True
|
||
|
||
|
||
class TestRulePersonnelVsIncome:
|
||
def test_income_down_staff_same(self):
|
||
r = rule_personnel_vs_income(-37, 0) # 收入降37%人员不变
|
||
assert r["passed"] is False
|
||
assert r["level"] == "yellow"
|
||
|
||
def test_staff_reduced(self):
|
||
r = rule_personnel_vs_income(-37, -10) # 人员也降
|
||
assert r["passed"] is True
|
||
|
||
def test_income_stable(self):
|
||
r = rule_personnel_vs_income(-10, 0)
|
||
assert r["passed"] is True
|
||
|
||
|
||
class TestRuleCostVsIncome:
|
||
def test_cost_over_100(self):
|
||
r = rule_cost_vs_income(100.4) # 酣客成本率100.4%
|
||
assert r["passed"] is False
|
||
assert r["level"] == "red"
|
||
|
||
def test_cost_normal(self):
|
||
r = rule_cost_vs_income(87.0) # 博海成本率87%
|
||
assert r["passed"] is True
|
||
|
||
|
||
class TestSevenFlowScore:
|
||
def test_all_green(self):
|
||
checks = [{"level": "green"}, {"level": "green"}, {"level": "green"}]
|
||
assert seven_flow_score(checks) == 100
|
||
|
||
def test_mixed(self):
|
||
checks = [{"level": "green"}, {"level": "yellow"}, {"level": "red"}]
|
||
assert seven_flow_score(checks) == 53 # (100+60+0)/3
|
||
|
||
def test_all_red(self):
|
||
checks = [{"level": "red"}, {"level": "red"}]
|
||
assert seven_flow_score(checks) == 0
|
||
|
||
def test_empty(self):
|
||
assert seven_flow_score([]) == 0
|
||
|
||
|
||
class TestHankeScenario:
|
||
"""酣客2026H1真实场景（应触发多项红色预警）"""
|
||
|
||
def test_full_hanke_checks(self):
|
||
checks = [
|
||
rule_invoice_inverted(input_tax=27157, output_tax=31445), # 进项2.7亿vs销项3.1亿
|
||
rule_tax_payable_ratio(tax_payable=429, income=713), # 应交429万vs收入713万
|
||
rule_margin_abnormal(margin_rate=-0.43, industry_base=10), # 账面毛利-0.43%
|
||
rule_income_vs_cash(cashflow=-13.89, revenue_growth=-37), # 现金流负+收入负增长
|
||
rule_income_vs_inventory(revenue_growth=-37, inventory_change=0), # 收入降库存平
|
||
rule_personnel_vs_income(revenue_growth=-37, employee_change=0), # 收入降人员平
|
||
rule_cost_vs_income(cost_rate=100.4), # 成本率100.4%
|
||
]
|
||
reds = [c for c in checks if c and c["level"] == "red"]
|
||
yellows = [c for c in checks if c and c["level"] == "yellow"]
|
||
score = seven_flow_score([c for c in checks if c])
|
||
assert len(reds) >= 3, f"酣客应至少3项红色, 实际{len(reds)}: {[c['message'] for c in reds]}"
|
||
assert len(yellows) >= 1
|
||
assert score < 60, f"酣客健康评分应<60, 实际{score}"
|
||
|
||
|
||
class TestBohaiScenario:
|
||
"""博海2026H1真实场景（应风险较低）"""
|
||
|
||
def test_full_bohai_checks(self):
|
||
checks = [
|
||
rule_margin_abnormal(margin_rate=13.0, industry_base=10), # 毛利率13%合理
|
||
rule_cost_vs_income(cost_rate=87.0), # 成本率87%
|
||
rule_tax_payable_ratio(tax_payable=5, income=100), # 正常税负
|
||
]
|
||
reds = [c for c in checks if c and c["level"] == "red"]
|
||
score = seven_flow_score([c for c in checks if c])
|
||
assert len(reds) == 0, f"博海不应有红色预警: {[c['message'] for c in reds]}"
|
||
assert score >= 60, f"博海健康评分应>=60, 实际{score}"
|