179 lines
6.4 KiB
Python
179 lines
6.4 KiB
Python
"""工具层测试：cost_engine, predict_engine, cache 核心函数"""
|
||
import pytest
|
||
from app.utils.cost_engine import (
|
||
calc_variance,
|
||
calc_efficiency_variance,
|
||
)
|
||
from app.utils.predict_engine import (
|
||
cvp_analysis,
|
||
npv,
|
||
irr,
|
||
sensitivity_analysis,
|
||
scenario_analysis,
|
||
)
|
||
from app.utils.cache import _make_key, get, set, delete, clear_all
|
||
|
||
|
||
class TestCostVariance:
|
||
"""成本差异计算测试"""
|
||
|
||
def test_calc_variance_standard(self):
|
||
"""标准差异计算"""
|
||
result = calc_variance(10, 12, 5, 4.5)
|
||
# 量差 = (12-10)*5 = 10
|
||
# 价差 = (4.5-5)*12 = -6
|
||
# 总差异 = 10 + (-6) = 4
|
||
assert result["qty_variance"] == 10.0
|
||
assert result["price_variance"] == -6.0
|
||
assert result["total_variance"] == 4.0
|
||
assert result["standard_cost"] == 50.0
|
||
assert result["actual_cost"] == 54.0
|
||
|
||
def test_calc_variance_equal(self):
|
||
"""实际与标准完全一致"""
|
||
result = calc_variance(10, 10, 5, 5)
|
||
assert result["qty_variance"] == 0
|
||
assert result["price_variance"] == 0
|
||
assert result["total_variance"] == 0
|
||
|
||
def test_calc_variance_no_qty_diff(self):
|
||
"""只有价格差异"""
|
||
result = calc_variance(10, 10, 5, 6)
|
||
assert result["qty_variance"] == 0
|
||
assert result["price_variance"] == 10.0 # (6-5)*10
|
||
assert result["total_variance"] == 10.0
|
||
|
||
def test_calc_efficiency_variance(self):
|
||
"""效率差异计算"""
|
||
result = calc_efficiency_variance(100, 120, 50)
|
||
# (120-100)*50 = 1000
|
||
assert result["efficiency_variance"] == 1000.0
|
||
|
||
|
||
class TestCvpAnalysis:
|
||
"""CVP本量利分析测试"""
|
||
|
||
def test_basic_bep(self):
|
||
"""基础盈亏平衡点"""
|
||
result = cvp_analysis(unit_price=100, unit_variable_cost=60, fixed_cost=20000)
|
||
assert result["contribution_margin"] == 40.0
|
||
assert result["contribution_ratio"] == 40.0
|
||
assert result["bep_units"] == 500.0
|
||
assert result["bep_revenue"] == 50000.0
|
||
|
||
def test_with_actual_volume(self):
|
||
"""带实际销量的安全边际计算"""
|
||
result = cvp_analysis(unit_price=100, unit_variable_cost=60, fixed_cost=20000, actual_volume=800)
|
||
assert result["safety_margin_units"] == 300.0 # 800-500
|
||
assert result["safety_margin_ratio"] == 37.5 # 300/800*100
|
||
assert result["actual_profit"] == 12000.0 # (100-60)*800 - 20000
|
||
|
||
def test_with_target_profit(self):
|
||
"""带目标利润"""
|
||
result = cvp_analysis(unit_price=100, unit_variable_cost=60, fixed_cost=20000, target_profit=10000)
|
||
assert result["target_units"] == 750.0 # (20000+10000)/40
|
||
assert result["target_revenue"] == 75000.0
|
||
|
||
def test_price_less_than_variable_cost(self):
|
||
"""单价<=变动成本 → 错误"""
|
||
result = cvp_analysis(unit_price=50, unit_variable_cost=60, fixed_cost=10000)
|
||
assert "error" in result
|
||
|
||
|
||
class TestNpvIrr:
|
||
"""NPV/IRR投资决策测试"""
|
||
|
||
def test_npv_positive(self):
|
||
"""正NPV → 可行"""
|
||
result = npv(1000, [400, 500, 500], 10)
|
||
assert result["is_viable"] is True
|
||
assert result["npv"] > 0
|
||
assert result["profitability_index"] > 1.0
|
||
|
||
def test_npv_negative(self):
|
||
"""负NPV → 不可行"""
|
||
result = npv(10000, [100, 100, 100], 10)
|
||
assert result["is_viable"] is False
|
||
assert result["npv"] < 0
|
||
|
||
def test_npv_empty_cashflows(self):
|
||
"""空现金流 → 错误"""
|
||
result = npv(1000, [], 10)
|
||
assert "error" in result
|
||
|
||
def test_irr_basic(self):
|
||
"""基本IRR计算"""
|
||
result = irr(1000, [300, 400, 500, 200])
|
||
assert result["payback_period"] is not None
|
||
assert result["is_viable"] is True # IRR > 0
|
||
|
||
def test_irr_negative_case(self):
|
||
"""现金流总和小于投资→可能负IRR"""
|
||
result = irr(10000, [1000, 1000, 1000])
|
||
assert result["payback_period"] is None # 无法回收
|
||
|
||
|
||
class TestSensitivityAnalysis:
|
||
"""敏感性分析测试"""
|
||
|
||
def test_basic_sensitivity(self):
|
||
"""基础敏感性分析"""
|
||
result = sensitivity_analysis(base_revenue=1000, base_cost=800, step=10, max_step=20)
|
||
assert result["base_profit"] == 200.0
|
||
assert len(result["factors"]) > 0
|
||
# 0%变动时利润不变
|
||
zero = [f for f in result["factors"] if f["change_pct"] == 0][0]
|
||
assert zero["revenue_change_profit"] == 200.0
|
||
assert zero["revenue_sensitivity"] == 0.0
|
||
|
||
|
||
class TestScenarioAnalysis:
|
||
"""情景模拟测试"""
|
||
|
||
def test_three_scenarios(self):
|
||
"""三情景分析"""
|
||
result = scenario_analysis(
|
||
optimistic={"revenue": 130, "cost": 90},
|
||
pessimistic={"revenue": 80, "cost": 110},
|
||
base={"revenue": 100, "cost": 100},
|
||
)
|
||
assert len(result["scenarios"]) == 3
|
||
assert result["scenarios"][0]["scenario"] == "乐观"
|
||
assert result["scenarios"][1]["scenario"] == "中性"
|
||
assert result["scenarios"][2]["scenario"] == "悲观"
|
||
assert result["expected_profit"] == round((40 + 0 + (-30)) / 3, 2)
|
||
|
||
|
||
class TestCacheUtils:
|
||
"""缓存工具测试（不依赖Redis，测试逻辑）"""
|
||
|
||
def test_make_key_format(self):
|
||
"""缓存key格式"""
|
||
key = _make_key("test", "hello")
|
||
assert key.startswith("cma:cache:test:")
|
||
assert len(key) > len("cma:cache:test:")
|
||
|
||
def test_get_when_unavailable(self):
|
||
"""Redis不可用 → get返回None"""
|
||
# cache.py 在模块级别会尝试连接Redis，如果不可用则 _available=False
|
||
# 此时 get 返回 None
|
||
result = get("test", "key")
|
||
# 可能返回 None（Redis不可用）或者正常获取
|
||
assert result is None or isinstance(result, object)
|
||
|
||
def test_set_when_unavailable(self):
|
||
"""Redis不可用 → set返回False"""
|
||
result = set("test", "key", {"data": 123})
|
||
# 可能返回 False（Redis不可用）或者 True
|
||
assert result is False or result is True
|
||
|
||
def test_delete_when_unavailable(self):
|
||
"""Redis不可用 → delete返回False"""
|
||
result = delete("test", "key")
|
||
assert result is False or result is True
|
||
|
||
def test_clear_all_when_unavailable(self):
|
||
"""Redis不可用 → clear_all返回False"""
|
||
result = clear_all()
|
||
assert result is False or result is True
|