289 lines
12 KiB
Python
289 lines
12 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
|
||
|
||
鉴权（cma-api-auth-surface-audit-20260914）：/api/cma/predict/* 已收口为「用户端点必须
|
||
带会话 Bearer」。本文件聚焦业务逻辑，`_predict_auth` autouse fixture 统一注入账套1的
|
||
测试 token 作为 client 默认头；需要其他账套的用例自行用 `_token_for_entity` 传 headers。
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import CashForecast, ForecastAccuracy, Entity, User, UserEntity
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
|
||
|
||
BASE = "/api/cma/predict"
|
||
|
||
|
||
def _token_for_entity(client: TestClient, db: Session, eid: int) -> str:
|
||
"""为指定账套签发测试 token（自动补实体 + 授权），供跨账套/自定义实体用例使用"""
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
db.add(Entity(id=eid, name=f"企业{eid}", short_name=f"E{eid}", status="active"))
|
||
db.commit()
|
||
user = db.query(User).filter(User.username == "testadmin").first()
|
||
if user is None:
|
||
user = create_test_user(db)
|
||
db.commit()
|
||
rel = db.query(UserEntity).filter(
|
||
UserEntity.user_id == user.id, UserEntity.entity_id == eid).first()
|
||
if not rel:
|
||
db.add(UserEntity(user_id=user.id, entity_id=eid))
|
||
db.commit()
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": eid})
|
||
assert resp.status_code == 200, f"登录 entity{eid} 失败: {resp.text[:200]}"
|
||
return resp.json()["token"]
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _predict_auth(client: TestClient, db: Session):
|
||
"""统一注入账套1的测试 token（显式传 headers= 的用例仍可覆盖，httpx 请求级优先）"""
|
||
if db.query(User).filter(User.username == "testadmin").first() is None:
|
||
create_test_user(db)
|
||
db.commit()
|
||
client.headers["Authorization"] = f"Bearer {get_token_for_user(client)}"
|
||
yield
|
||
|
||
|
||
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):
|
||
"""历史为空"""
|
||
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):
|
||
"""准确率为空"""
|
||
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):
|
||
"""博海增长质量：企业名取库内 Entity.name，且不采信请求体指标（账套隔离 2026-09-21）
|
||
|
||
契约变更：原断言硬编码 ENTITY_DATA 的 "陕西博海网络科技"（该硬编码兜底已按
|
||
batch5 派单要求移除）→ 改为断言库内来源；无 kpi_values 时如实返回未采集。
|
||
"""
|
||
token2 = _token_for_entity(client, db, 2)
|
||
resp = client.post(f"{BASE}/growth-quality", json={"entity_id": 2, "rebateRate": 99},
|
||
headers=auth_header(token2))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["entity_id"] == 2
|
||
assert data["entity_name"] == "企业2", "企业名应取库内 Entity.name，而非硬编码"
|
||
assert data["collected"] is False
|
||
assert data["overall"] is None, "库内无数据不得编造评分"
|
||
|
||
def test_growth_quality_custom(self, client: TestClient, db: Session):
|
||
"""自定义企业数据（低分场景）→ 契约变更：请求体指标不再被采信
|
||
|
||
原用例靠请求体 rebateRate/mgmtRatio 等打分得 danger；该回退已按 batch5 派单
|
||
要求移除 → 改为断言请求体指标被忽略、无库内数据即未采集（不编造评分）。
|
||
"""
|
||
token99 = _token_for_entity(client, db, 99)
|
||
resp = client.post(f"{BASE}/growth-quality", json={
|
||
"entity_id": 99,
|
||
"name": "测试企业",
|
||
"rebateRate": 90, "trueGrossMargin": 5, "cashRatio": 3,
|
||
"expenseGrowthRate": 3, "revenueGrowthRate": 1, "mgmtRatio": 500,
|
||
}, headers=auth_header(token99))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["entity_name"] == "企业99", "企业名取库内，不取请求体 name"
|
||
assert data["collected"] is False
|
||
assert data["overall"] is None and data["level_type"] is None
|
||
assert data["dimensions"] == {} and data["diagnosis"] is None
|