""" 路线图R2/R5 测试（2026-08-30） R2: 机会检测（KPI向好/预算余量/预测上行） R5: 预算↔现金流↔行动 闭环自检 """ import pytest from datetime import datetime from sqlalchemy.orm import Session from tests.conftest import create_test_kpi from app.models import KPIDefinition, KPIValue, BudgetPlan, CashPlan, ActionPlan, KpiForecastLog from scripts.opportunity_detector import ( detect_kpi_improving, detect_budget_headroom, detect_rolling_up, detect_all, flatten, ) from scripts.closed_loop_check import check_entity, build_report def _kpi(db, code, target=100.0, **kw): return create_test_kpi(db, kpi_code=code, target_value=target, **kw) def _value(db, kpi_id, period, actual, entity_id=1): v = KPIValue(kpi_id=kpi_id, period=period, actual_value=actual, entity_id=entity_id) db.add(v) return v def _budget(db, kpi_id, period, value, year=None, month=None, entity_id=1): if year is None: year = int(period.split("-")[0]) month = int(period.split("-")[1]) b = BudgetPlan(entity_id=entity_id, kpi_id=kpi_id, period=period, budget_value=value, budget_year=year, budget_month=month, version="v1.0", status="active") db.add(b) return b class TestOpportunityR2: def test_kpi_improving(self, db): """连续3期执行率>110% → KPI向好机会""" kpi = _kpi(db, "OPP_01", target=100.0) _value(db, kpi.id, "2026-04", 120.0) _value(db, kpi.id, "2026-05", 130.0) _value(db, kpi.id, "2026-06", 140.0) db.commit() out = detect_kpi_improving(db, 1) assert len(out) == 1 assert out[0]["type"] == "kpi_improving" assert out[0]["kpi_id"] == kpi.id def test_kpi_improving_not_enough_data(self, db): """不足3期不判定""" kpi = _kpi(db, "OPP_02", target=100.0) _value(db, kpi.id, "2026-05", 130.0) _value(db, kpi.id, "2026-06", 140.0) db.commit() assert detect_kpi_improving(db, 1) == [] def test_kpi_improving_low_ratio_skip(self, db): """执行率未超110%不判定""" kpi = _kpi(db, "OPP_03", target=100.0) _value(db, kpi.id, "2026-04", 90.0) _value(db, kpi.id, "2026-05", 95.0) _value(db, kpi.id, "2026-06", 100.0) db.commit() assert detect_kpi_improving(db, 1) == [] def test_budget_headroom(self, db): """当月预算执行率<70% → 预算余量机会""" kpi = _kpi(db, "OPP_04", target=1000.0) _value(db, kpi.id, "2026-08", 300.0) _budget(db, kpi.id, "2026-08", 1000.0) db.commit() out = detect_budget_headroom(db, 1) assert len(out) == 1 assert out[0]["type"] == "budget_headroom" def test_budget_headroom_negative_skip(self, db): """实际值为负（现金流异常）不误判为余量""" kpi = _kpi(db, "OPP_05", target=1000.0) _value(db, kpi.id, "2026-08", -500.0) _budget(db, kpi.id, "2026-08", 1000.0) db.commit() assert detect_budget_headroom(db, 1) == [] def test_budget_headroom_dedup(self, db): """同KPI同期间多版本预算只取一条""" kpi = _kpi(db, "OPP_06", target=1000.0) _value(db, kpi.id, "2026-08", 300.0) _budget(db, kpi.id, "2026-08", 1000.0) b2 = _budget(db, kpi.id, "2026-08", 2000.0) b2.version = "v2.0" db.commit() assert len(detect_budget_headroom(db, 1)) == 1 def test_rolling_up(self, db): """预测值上升 → 滚动机会""" kpi = _kpi(db, "OPP_07", target=100.0) now = datetime.now() db.add(KpiForecastLog(entity_id=1, kpi_id=kpi.id, kpi_code=kpi.kpi_code, period="2026-07", forecast_value=100.0, model="linear", created_at=now)) db.add(KpiForecastLog(entity_id=1, kpi_id=kpi.id, kpi_code=kpi.kpi_code, period="2026-08", forecast_value=130.0, model="linear", created_at=now)) db.commit() out = detect_rolling_up(db, 1) assert len(out) == 1 assert out[0]["type"] == "rolling_up" def test_rolling_down_skip(self, db): """预测下降不判定为机会""" kpi = _kpi(db, "OPP_08", target=100.0) now = datetime.now() db.add(KpiForecastLog(entity_id=1, kpi_id=kpi.id, kpi_code=kpi.kpi_code, period="2026-07", forecast_value=130.0, model="linear", created_at=now)) db.add(KpiForecastLog(entity_id=1, kpi_id=kpi.id, kpi_code=kpi.kpi_code, period="2026-08", forecast_value=100.0, model="linear", created_at=now)) db.commit() assert detect_rolling_up(db, 1) == [] def test_flatten(self): d = {"kpi_improving": [1], "budget_headroom": [2, 3], "rolling_up": []} assert flatten(d) == [1, 2, 3] class TestClosedLoopR5: def test_overrun_missing_both(self, db): """超预算且缺现金流/行动 → 提示同步""" kpi = _kpi(db, "CL_01", target=100.0) _value(db, kpi.id, "2026-08", 200.0) _budget(db, kpi.id, "2026-08", 100.0) db.commit() r = check_entity(db, 1, "2026-08") assert len(r["issues"]) == 1 it = r["issues"][0] assert it["abnormal_type"] == "超预算" assert "现金流" in it["missing"] assert "行动方案" in it["missing"] def test_overrun_has_cash_and_action(self, db): """超预算但有现金流+行动 → 三闭环同步""" kpi = _kpi(db, "CL_02", target=100.0) _value(db, kpi.id, "2026-08", 200.0) b = _budget(db, kpi.id, "2026-08", 100.0) db.add(CashPlan(entity_id=1, plan_type="receive", related_kpi_id=kpi.id, budget_plan_id=b.id, amount=200.0, plan_date=datetime(2026, 8, 15), status="pending")) db.add(ActionPlan(kpi_id=kpi.id, title="改善计划", status="in_progress")) db.commit() r = check_entity(db, 1, "2026-08") assert len(r["issues"]) == 1 assert r["issues"][0]["missing"] == [] def test_normal_no_issue(self, db): """执行率正常 → 无异常""" kpi = _kpi(db, "CL_03", target=100.0) _value(db, kpi.id, "2026-08", 100.0) _budget(db, kpi.id, "2026-08", 100.0) db.commit() r = check_entity(db, 1, "2026-08") assert r["issues"] == [] def test_low_execution(self, db): """低执行率 → 异常（warning）""" kpi = _kpi(db, "CL_04", target=100.0) _value(db, kpi.id, "2026-08", 50.0) _budget(db, kpi.id, "2026-08", 100.0) db.commit() r = check_entity(db, 1, "2026-08") assert len(r["issues"]) == 1 assert r["issues"][0]["abnormal_type"] == "低执行" assert r["issues"][0]["level"] == "warning" def test_build_report(self): result = {"entity_id": 1, "period": "2026-08", "issues": [ {"kpi_id": 1, "kpi_name": "营收", "period": "2026-08", "budget_value": 100.0, "actual_value": 200.0, "exec_ratio": 200.0, "abnormal_type": "超预算", "level": "critical", "cash_plan_count": 0, "action_plan_count": 0, "missing": ["现金流", "行动方案"], "suggestion": "请同步现金流、行动方案"} ]} report = build_report([result], "2026-08-30 12:00:00") assert "闭环自检" in report assert "营收" in report assert "共发现异常 1 项" in report