236 lines
9.2 KiB
Python
236 lines
9.2 KiB
Python
"""工具层测试：calc_engine 和 deviation_engine 核心函数"""
|
||
import pytest
|
||
from app.utils.deviation_engine import (
|
||
calc_deviation,
|
||
calc_period_deviation,
|
||
calc_period_diff,
|
||
check_trend_anomaly,
|
||
run_deviation_check,
|
||
)
|
||
from app.utils.calc_engine import calculate_all, _get
|
||
from app.models import KPIDefinition, KPIValue, BudgetPlan, KPIAlert
|
||
|
||
|
||
class TestDeviationCalc:
|
||
"""偏差计算核心函数测试"""
|
||
|
||
def test_calc_deviation_positive(self):
|
||
"""实际 > 预算 → 正差异"""
|
||
result = calc_deviation(120, 100)
|
||
assert result["deviation_amount"] == 20
|
||
assert result["deviation_rate"] == 20.0
|
||
assert result["is_over_budget"] is True
|
||
|
||
def test_calc_deviation_negative(self):
|
||
"""实际 < 预算 → 负差异"""
|
||
result = calc_deviation(80, 100)
|
||
assert result["deviation_amount"] == -20
|
||
assert result["deviation_rate"] == -20.0
|
||
assert result["is_over_budget"] is False
|
||
|
||
def test_calc_deviation_zero_budget(self):
|
||
"""预算为0 → 差异为None"""
|
||
result = calc_deviation(100, 0)
|
||
assert result["deviation_amount"] is None
|
||
assert result["deviation_rate"] is None
|
||
assert result["is_over_budget"] is None
|
||
|
||
def test_calc_deviation_none_budget(self):
|
||
"""预算为None → 差异为None"""
|
||
result = calc_deviation(100, None)
|
||
assert result["deviation_amount"] is None
|
||
|
||
def test_calc_deviation_exact_match(self):
|
||
"""实际==预算 → 差异0"""
|
||
result = calc_deviation(100, 100)
|
||
assert result["deviation_amount"] == 0
|
||
assert result["deviation_rate"] == 0.0
|
||
assert result["is_over_budget"] is False
|
||
|
||
|
||
class TestPeriodDeviation:
|
||
"""单期差异计算（依赖数据库）"""
|
||
|
||
def test_calc_period_deviation_with_budget(self, db):
|
||
"""有预算值和实际值 → 正确计算差异"""
|
||
kpi = KPIDefinition(kpi_code="TEST_DEV_001", kpi_name="测试KPI", dimension="finance",
|
||
target_value=1000, status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
|
||
bp = BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=500.0,
|
||
budget_year=2026, budget_month=6, status="active", version="v1.0")
|
||
db.add(bp)
|
||
kv = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=600.0)
|
||
db.add(kv)
|
||
db.commit()
|
||
|
||
result = calc_period_deviation(db, kpi.id, "2026-06")
|
||
assert result["kpi_code"] == "TEST_DEV_001"
|
||
assert result["actual_value"] == 600.0
|
||
assert result["budget_value"] == 500.0
|
||
assert result["deviation_amount"] == 100.0
|
||
assert result["deviation_rate"] == 20.0
|
||
|
||
def test_calc_period_deviation_no_budget_fallback_target(self, db):
|
||
"""无预算值时回退到 target_value/12（monthly频率）"""
|
||
kpi = KPIDefinition(kpi_code="TEST_DEV_002", kpi_name="测试KPI", dimension="finance",
|
||
target_value=1200, frequency="monthly", status="active")
|
||
db.add(kpi)
|
||
db.flush()
|
||
kv = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=150.0)
|
||
db.add(kv)
|
||
db.commit()
|
||
|
||
result = calc_period_deviation(db, kpi.id, "2026-06")
|
||
# target=1200, monthly budget = 1200/12 = 100
|
||
assert result["budget_value"] == 100.0
|
||
assert result["deviation_amount"] == 50.0
|
||
|
||
|
||
class TestPeriodDiff:
|
||
"""同比/环比差异测试"""
|
||
|
||
def test_yoy_diff(self, db):
|
||
"""同比差异计算"""
|
||
kpi = KPIDefinition(kpi_code="TEST_YOY", kpi_name="同比测试", dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
# 上年同期值
|
||
kv1 = KPIValue(kpi_id=kpi.id, period="2025-06", actual_value=100.0)
|
||
# 本期值
|
||
kv2 = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=120.0)
|
||
db.add_all([kv1, kv2])
|
||
db.commit()
|
||
|
||
result = calc_period_diff(db, kpi.id, "2026-06", "yoy")
|
||
assert result["label"] == "同比"
|
||
assert result["current_value"] == 120.0
|
||
assert result["prev_value"] == 100.0
|
||
assert result["diff_amount"] == 20.0
|
||
assert result["diff_rate"] == 20.0
|
||
|
||
def test_mom_diff_january(self, db):
|
||
"""环比：跨年情况（1月→去年12月）"""
|
||
kpi = KPIDefinition(kpi_code="TEST_MOM", kpi_name="环比测试", dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
kv1 = KPIValue(kpi_id=kpi.id, period="2025-12", actual_value=200.0)
|
||
kv2 = KPIValue(kpi_id=kpi.id, period="2026-01", actual_value=180.0)
|
||
db.add_all([kv1, kv2])
|
||
db.commit()
|
||
|
||
result = calc_period_diff(db, kpi.id, "2026-01", "mom")
|
||
assert result["label"] == "环比"
|
||
assert result["prev_period"] == "2025-12"
|
||
assert result["diff_amount"] == -20.0
|
||
assert result["diff_rate"] == -10.0
|
||
|
||
def test_diff_unknown_type(self, db):
|
||
"""未知比较类型"""
|
||
kpi = KPIDefinition(kpi_code="TEST_UNKNOWN", kpi_name="未知", dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
result = calc_period_diff(db, kpi.id, "2026-06", "wow")
|
||
assert "error" in result
|
||
|
||
|
||
class TestTrendAnomaly:
|
||
"""趋势异常检测"""
|
||
|
||
def test_continuous_decline(self, db):
|
||
"""连续3期下滑 → 触发异常"""
|
||
kpi = KPIDefinition(kpi_code="TEST_TREND_DEC", kpi_name="下滑趋势",
|
||
dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
for m, v in [(4, 150), (5, 120), (6, 90)]:
|
||
db.add(KPIValue(kpi_id=kpi.id, period=f"2026-{m:02d}", actual_value=float(v)))
|
||
db.commit()
|
||
|
||
result = check_trend_anomaly(db, kpi.id, "2026-06", consecutive=3)
|
||
assert result["anomaly"] is True
|
||
assert result["type"] == "continuous_decline"
|
||
assert "下滑" in result["message"]
|
||
|
||
def test_continuous_rise(self, db):
|
||
"""连续3期上升 → 触发异常"""
|
||
kpi = KPIDefinition(kpi_code="TEST_TREND_RISE", kpi_name="上升趋势",
|
||
dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
for m, v in [(4, 50), (5, 80), (6, 110)]:
|
||
db.add(KPIValue(kpi_id=kpi.id, period=f"2026-{m:02d}", actual_value=float(v)))
|
||
db.commit()
|
||
|
||
result = check_trend_anomaly(db, kpi.id, "2026-06", consecutive=3)
|
||
assert result["anomaly"] is True
|
||
assert result["type"] == "continuous_rise"
|
||
|
||
def test_no_anomaly(self, db):
|
||
"""不满足连续条件 → 无异常"""
|
||
kpi = KPIDefinition(kpi_code="TEST_TREND_NORMAL", kpi_name="正常趋势",
|
||
dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
for m, v in [(4, 100), (5, 120), (6, 90)]:
|
||
db.add(KPIValue(kpi_id=kpi.id, period=f"2026-{m:02d}", actual_value=float(v)))
|
||
db.commit()
|
||
|
||
result = check_trend_anomaly(db, kpi.id, "2026-06", consecutive=3)
|
||
assert result["anomaly"] is False
|
||
|
||
def test_insufficient_data(self, db):
|
||
"""数据不足 → 无异常"""
|
||
kpi = KPIDefinition(kpi_code="TEST_TREND_SHORT", kpi_name="数据不足",
|
||
dimension="finance", status="active")
|
||
db.add(kpi)
|
||
db.flush()
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=100.0))
|
||
db.commit()
|
||
|
||
result = check_trend_anomaly(db, kpi.id, "2026-06", consecutive=3)
|
||
assert result["anomaly"] is False
|
||
assert "数据不足" in result.get("reason", "")
|
||
|
||
|
||
class TestRunDeviationCheck:
|
||
"""run_deviation_check 集成测试"""
|
||
|
||
def test_run_check_creates_alerts(self, db):
|
||
"""偏差超阈值 → 创建预警"""
|
||
kpi = KPIDefinition(kpi_code="SALES_TOTAL", kpi_name="销售总额",
|
||
dimension="finance", target_value=100000, status="active")
|
||
db.add(kpi)
|
||
db.commit()
|
||
|
||
# 预算50000 vs 实际40000 → 差异率-20%（higher_better，低于预算触发）
|
||
bp = BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=50000.0,
|
||
budget_year=2026, budget_month=6, status="active", version="v1.0")
|
||
kv = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=40000.0)
|
||
db.add_all([bp, kv])
|
||
db.commit()
|
||
|
||
count = run_deviation_check(db, period="2026-06")
|
||
assert count >= 1
|
||
|
||
alert = db.query(KPIAlert).filter(KPIAlert.kpi_id == kpi.id).first()
|
||
assert alert is not None
|
||
assert "差异预警" in alert.alert_message
|
||
|
||
def test_run_check_no_alert_when_on_budget(self, db):
|
||
"""实际≈预算 → 不创建预警"""
|
||
kpi = KPIDefinition(kpi_code="SALES_TOTAL", kpi_name="销售总额",
|
||
dimension="finance", target_value=100000, status="active")
|
||
db.add(kpi)
|
||
db.flush()
|
||
bp = BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=50000.0,
|
||
budget_year=2026, budget_month=6, status="active", version="v1.0")
|
||
kv = KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50500.0)
|
||
db.add_all([kpi, bp, kv])
|
||
db.commit()
|
||
|
||
count = run_deviation_check(db, period="2026-06")
|
||
# 差异率1% < 10%阈值，不触发
|
||
assert count == 0
|