Hermes CI Fix
a13a080381
- 新增8个测试文件(bot_bridge/kpi_causality/cash/predict/reports/tax_compliance/expenses/probe_cost) - 增强 budget/auth/users + conftest账套模式适配 - 测试驱动修复: bot_bridge导入batch_id→source_batch; cash_forecast extra空dict - 全量: 451 passed, 1 xfailed; 报告 docs/cma-test-coverage-report.md
243 lines
9.7 KiB
Python
243 lines
9.7 KiB
Python
"""预测模拟模块测试 — CVP/投资决策/敏感性/情景/现金流/实物期权/增长质量
|
|
|
|
覆盖 predict.py 主要端点（纯计算 + 少量DB）：
|
|
cvp / investment / sensitivity / scenario / cvp-detailed / cash-forecast /
|
|
cash-forecast/history / accuracy / scenario-suggestions /
|
|
scenario-suggestion/generate / real-option / growth-quality
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import CashForecast, ForecastAccuracy
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
|
|
|
BASE = "/api/cma/predict"
|
|
|
|
|
|
class TestCvp:
|
|
def test_cvp_basic(self, client: TestClient, db: Session):
|
|
"""本量利分析"""
|
|
resp = client.post(f"{BASE}/cvp", json={
|
|
"unit_price": 100, "unit_variable_cost": 60, "fixed_cost": 100000,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["contribution_margin"] == 40
|
|
assert data["bep_units"] == 2500 # 100000/40
|
|
assert data["bep_revenue"] == 250000.0
|
|
|
|
def test_cvp_with_target_profit(self, client: TestClient, db: Session):
|
|
"""含目标利润"""
|
|
resp = client.post(f"{BASE}/cvp", json={
|
|
"unit_price": 100, "unit_variable_cost": 60, "fixed_cost": 100000,
|
|
"target_profit": 20000,
|
|
})
|
|
assert resp.status_code == 200
|
|
# 目标销量 = (100000+20000)/40 = 3000
|
|
assert resp.json()["target_units"] == 3000
|
|
|
|
def test_cvp_bad_input(self, client: TestClient, db: Session):
|
|
"""非法输入 → 400"""
|
|
resp = client.post(f"{BASE}/cvp", json={"unit_price": "abc"})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestInvestment:
|
|
def test_investment_npv_irr(self, client: TestClient, db: Session):
|
|
"""NPV/IRR计算"""
|
|
resp = client.post(f"{BASE}/investment", json={
|
|
"initial_investment": 10000,
|
|
"discount_rate": 10,
|
|
"cash_flows": [3000, 3000, 3000, 3000, 3000],
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "npv_analysis" in data
|
|
assert "irr_analysis" in data
|
|
|
|
def test_investment_empty_cashflows(self, client: TestClient, db: Session):
|
|
"""空现金流 → 400"""
|
|
resp = client.post(f"{BASE}/investment", json={"initial_investment": 100})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestSensitivityScenario:
|
|
def test_sensitivity(self, client: TestClient, db: Session):
|
|
"""敏感性分析"""
|
|
resp = client.post(f"{BASE}/sensitivity", json={
|
|
"base_revenue": 1000, "base_cost": 700,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["base_profit"] == 300.0
|
|
assert len(data["factors"]) > 0
|
|
assert "revenue_sensitivity" in data["factors"][0]
|
|
|
|
def test_scenario(self, client: TestClient, db: Session):
|
|
"""情景模拟"""
|
|
resp = client.post(f"{BASE}/scenario", json={
|
|
"optimistic": {"revenue": 120, "cost": 60},
|
|
"pessimistic": {"revenue": 80, "cost": 70},
|
|
"base": {"revenue": 100, "cost": 65},
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
scenarios = {s["scenario"] for s in data["scenarios"]}
|
|
assert scenarios == {"乐观", "中性", "悲观"}
|
|
|
|
def test_scenario_missing(self, client: TestClient, db: Session):
|
|
"""缺情景 → 400"""
|
|
resp = client.post(f"{BASE}/scenario", json={"base": {"revenue": 100}})
|
|
assert resp.status_code == 400
|
|
|
|
def test_cvp_detailed(self, client: TestClient, db: Session):
|
|
"""CVP详细分析（含保本图/方案推演）"""
|
|
resp = client.post(f"{BASE}/cvp-detailed", json={
|
|
"fixed_cost": 617, "variable_cost_rate": 0.4862,
|
|
"unit_price": 228, "current_volume": 5300,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["contribution_margin_rate"] == 51.38
|
|
assert len(data["scenarios"]) == 3
|
|
assert len(data["chart_data"]) > 0
|
|
assert data["breakeven_units"] > 0
|
|
|
|
|
|
class TestCashForecast:
|
|
def test_cash_forecast_no_data(self, client: TestClient, db: Session):
|
|
"""无历史KPI时仍返回预测（退化处理）"""
|
|
resp = client.post(f"{BASE}/cash-forecast", json={
|
|
"entity_id": 1, "days": 10, "current_cash": 50.0,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "daily" in data or "forecast" in data or "days" in data
|
|
|
|
def test_cash_forecast_history_empty(self, client: TestClient, db: Session):
|
|
"""历史为空"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = client.get(f"{BASE}/cash-forecast/history", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"] == []
|
|
|
|
def test_accuracy_empty(self, client: TestClient, db: Session):
|
|
"""准确率为空"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
resp = client.get(f"{BASE}/accuracy", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["data"] == []
|
|
assert data["summary"]["total_periods"] == 0
|
|
|
|
|
|
class TestScenarioSuggestions:
|
|
def test_suggestions_all(self, client: TestClient, db: Session):
|
|
"""获取全部情景建议"""
|
|
resp = client.get(f"{BASE}/scenario-suggestions")
|
|
assert resp.status_code == 200
|
|
types = {s["alert_type"] for s in resp.json()["data"]}
|
|
assert types == {"cash_low", "cash_critical", "cost_high", "revenue_drop"}
|
|
|
|
def test_suggestions_filter(self, client: TestClient, db: Session):
|
|
"""按类型过滤"""
|
|
resp = client.get(f"{BASE}/scenario-suggestions?alert_type=cash_low")
|
|
assert resp.status_code == 200
|
|
data = resp.json()["data"]
|
|
assert len(data) == 1
|
|
assert data[0]["alert_type"] == "cash_low"
|
|
|
|
def test_generate_suggestion(self, client: TestClient, db: Session):
|
|
"""动态生成建议"""
|
|
resp = client.post(f"{BASE}/scenario-suggestion/generate", json={
|
|
"alert_type": "cost_high", "kpi_name": "销售费用率",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert "title" in resp.json() or "description" in resp.json()
|
|
|
|
|
|
class TestRealOption:
|
|
def test_bs_call_expansion(self, client: TestClient, db: Session):
|
|
"""BSM看涨（扩张期权）"""
|
|
resp = client.post(f"{BASE}/real-option", json={
|
|
"opt_type": "expansion", "model": "bs",
|
|
"S0": 100, "X": 80, "t": 3, "r": 0.0174, "sigma": 0.30,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["option_value"] > 0
|
|
assert "intermediate" in data
|
|
assert "sensitivity" in data
|
|
|
|
def test_bs_put_abandon(self, client: TestClient, db: Session):
|
|
"""BSM看跌（放弃期权）"""
|
|
resp = client.post(f"{BASE}/real-option", json={
|
|
"opt_type": "abandon", "model": "bs",
|
|
"S0": 100, "X": 120, "t": 2, "r": 0.05, "sigma": 0.40,
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["option_value"] >= 0
|
|
|
|
def test_binomial_delay(self, client: TestClient, db: Session):
|
|
"""二叉树（延迟期权）"""
|
|
resp = client.post(f"{BASE}/real-option", json={
|
|
"opt_type": "delay", "model": "binomial",
|
|
"S0": 100, "X": 80, "t": 3, "r": 0.05, "sigma": 0.30, "n_steps": 50,
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["option_value"] > 0
|
|
|
|
def test_binomial_american_put(self, client: TestClient, db: Session):
|
|
"""二叉树美式看跌（放弃期权）"""
|
|
resp = client.post(f"{BASE}/real-option", json={
|
|
"opt_type": "abandon", "model": "binomial",
|
|
"S0": 100, "X": 120, "t": 2, "r": 0.05, "sigma": 0.40, "n_steps": 30,
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["option_value"] >= 0
|
|
|
|
def test_real_option_invalid_params(self, client: TestClient, db: Session):
|
|
"""非法参数 → 400"""
|
|
resp = client.post(f"{BASE}/real-option", json={"S0": -5})
|
|
assert resp.status_code == 400
|
|
|
|
def test_real_option_unsupported_comb(self, client: TestClient, db: Session):
|
|
"""不支持的组合 → 400"""
|
|
resp = client.post(f"{BASE}/real-option", json={"opt_type": "weird", "model": "bs"})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestGrowthQuality:
|
|
def test_growth_quality_entity1(self, client: TestClient, db: Session):
|
|
"""酣客增长质量"""
|
|
resp = client.post(f"{BASE}/growth-quality", json={"entity_id": 1})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "overall" in data or "level" in data or "score" in data
|
|
|
|
def test_growth_quality_entity2(self, client: TestClient, db: Session):
|
|
"""博海增长质量"""
|
|
resp = client.post(f"{BASE}/growth-quality", json={"entity_id": 2})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["entity_name"] == "陕西博海网络科技"
|
|
|
|
def test_growth_quality_custom(self, client: TestClient, db: Session):
|
|
"""自定义企业数据（低分场景）"""
|
|
resp = client.post(f"{BASE}/growth-quality", json={
|
|
"entity_id": 99,
|
|
"name": "测试企业",
|
|
"rebateRate": 90, "trueGrossMargin": 5, "cashRatio": 3,
|
|
"expenseGrowthRate": 3, "revenueGrowthRate": 1, "mgmtRatio": 500,
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["entity_name"] == "测试企业"
|
|
assert data["overall"] < 2
|
|
assert data["level_type"] == "danger"
|
|
# 低分诊断
|
|
assert "越增长越重" in data["diagnosis"]
|