chore: CI流程加固+Schema检查门禁
- deploy.sh: 提交纪律检查(本地未提交修改→中止) + 冒烟测试(健康/登录/KPI/BOT/Schema) - .woodpecker: 加前端typecheck+后端pytest测试步骤, backend-deploy加提交纪律检查 - 新增schema_check.py: ORM与数据库表结构一致性检查 - 修复budget_plans表缺3列(source_kpi_id/source_type/calc_logic) - 附带入库: budget测试+文档
This commit is contained in:
Hermes CI Fix
@@ -4,6 +4,7 @@ from fastapi.testclient import TestClient
|
||||
from sqlalchemy.orm import Session
|
||||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||||
from app.models.budget_plan import BudgetPlan
|
||||
from app.models import KPIValue, SystemConfig, CashPlan
|
||||
|
||||
|
||||
class TestBudgetPlans:
|
||||
@@ -683,3 +684,532 @@ class TestBudgetContract20260825:
|
||||
BudgetPlan.kpi_id, BudgetPlan.period, BudgetPlan.version).having(
|
||||
func.count() > 1).all()
|
||||
assert len(rows) == 0, f"存在重复预算记录: {rows}"
|
||||
|
||||
|
||||
class TestBudgetPlansEdges:
|
||||
"""预算计划CRUD异常路径与列表过滤（补齐未覆盖端点）"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_create_plan_kpi_not_found(self, client: TestClient, db: Session):
|
||||
"""KPI不存在 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": 99999, "period": "2026-06", "budget_value": 100.0})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_update_plan_not_found(self, client: TestClient, db: Session):
|
||||
"""更新不存在的计划 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.put(f"{self.BASE}/plans/99999", headers=auth_header(token),
|
||||
json={"budget_value": 100.0})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_delete_plan_not_found(self, client: TestClient, db: Session):
|
||||
"""删除不存在的计划 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.delete(f"{self.BASE}/plans/99999", headers=auth_header(token))
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_list_plans_filters(self, client: TestClient, db: Session):
|
||||
"""列表按 year/period/version 过滤"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="PLAN_FILTER_KPI")
|
||||
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 100.0})
|
||||
|
||||
assert client.get(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
params={"year": 2026}).json()["total"] == 1
|
||||
assert client.get(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
params={"year": 2025}).json()["total"] == 0
|
||||
assert client.get(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
params={"period": "2026-06"}).json()["total"] == 1
|
||||
assert client.get(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
params={"version": "v1.0"}).json()["total"] == 1
|
||||
|
||||
|
||||
class TestBudgetAutoDecomposeEdges:
|
||||
"""自动分解异常路径与加权模式"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_auto_decompose_single_missing_budget(self, client: TestClient, db: Session):
|
||||
"""单KPI模式缺 annual_budget → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_DECOMP_MISSING_BUDGET")
|
||||
|
||||
resp = client.post(f"{self.BASE}/auto-decompose", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "year": 2026})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_auto_decompose_kpi_not_found(self, client: TestClient, db: Session):
|
||||
"""单KPI模式KPI不存在 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/auto-decompose", headers=auth_header(token),
|
||||
json={"kpi_id": 99999, "annual_budget": 120000, "year": 2026})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_auto_decompose_single_weighted(self, client: TestClient, db: Session):
|
||||
"""单KPI加权分解（去年各月实际值作为权重）"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_DECOMP_WEIGHTED")
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2025-01", actual_value=10.0))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2025-02", actual_value=20.0))
|
||||
db.commit()
|
||||
|
||||
resp = client.post(f"{self.BASE}/auto-decompose", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "annual_budget": 120000, "year": 2026,
|
||||
"method": "weighted"})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["method"] == "weighted"
|
||||
assert len(data["monthly_budgets"]) == 12
|
||||
|
||||
|
||||
class TestBudgetVersionSubmitDiff:
|
||||
"""版本提交 + 版本差异对比（补齐未覆盖端点）"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_submit_version(self, client: TestClient, db: Session):
|
||||
"""提交版本审批：active → submitted"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="VER_SUBMIT_KPI")
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 10000.0})
|
||||
|
||||
resp = client.post(f"{self.BASE}/versions/submit", headers=auth_header(token),
|
||||
json={"version": "v1.0"})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["success"] is True
|
||||
assert data["status"] == "submitted"
|
||||
assert data["count"] == 1
|
||||
|
||||
def test_submit_version_not_found(self, client: TestClient, db: Session):
|
||||
"""提交不存在的版本 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/versions/submit", headers=auth_header(token),
|
||||
json={"version": "v9.9"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_approve_version_not_found(self, client: TestClient, db: Session):
|
||||
"""审批不存在的版本 → 404"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/versions/approve", headers=auth_header(token),
|
||||
json={"version": "v9.9", "action": "approved"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
def test_diff_versions(self, client: TestClient, db: Session):
|
||||
"""版本差异对比：v1.0 vs v2.0 逐KPI差异"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="VER_DIFF_KPI")
|
||||
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 100.0,
|
||||
"version": "v1.0"})
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 150.0,
|
||||
"version": "v2.0"})
|
||||
|
||||
resp = client.post(f"{self.BASE}/versions/diff", headers=auth_header(token),
|
||||
json={"version_a": "v1.0", "version_b": "v2.0", "year": 2026})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["summary"]["changed_count"] == 1
|
||||
assert data["summary"]["total_a"] == 100.0
|
||||
assert data["summary"]["total_b"] == 150.0
|
||||
assert data["diffs"][0]["version_a"] == 100.0
|
||||
assert data["diffs"][0]["version_b"] == 150.0
|
||||
|
||||
def test_diff_versions_missing_params(self, client: TestClient, db: Session):
|
||||
"""缺少 version_a/version_b → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/versions/diff", headers=auth_header(token),
|
||||
json={"year": 2026})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_versions_year_filter(self, client: TestClient, db: Session):
|
||||
"""版本列表按年份过滤"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="VER_YEAR_KPI")
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 10000.0})
|
||||
|
||||
assert len(client.get(f"{self.BASE}/versions", headers=auth_header(token),
|
||||
params={"year": 2026}).json()) >= 1
|
||||
assert client.get(f"{self.BASE}/versions", headers=auth_header(token),
|
||||
params={"year": 2030}).json() == []
|
||||
|
||||
|
||||
class TestBudgetApplyMethod:
|
||||
"""预算方法落地 apply-method（补齐未覆盖端点）"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def _seed_core_kpis(self, db: Session):
|
||||
for code in ["F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_GROSS_MARGIN"]:
|
||||
create_test_kpi(db, kpi_code=code, kpi_name=code)
|
||||
|
||||
def test_apply_method_zero_based(self, client: TestClient, db: Session):
|
||||
"""零基预算方法落地到年度预算"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
self._seed_core_kpis(db)
|
||||
|
||||
resp = client.post(f"{self.BASE}/apply-method", headers=auth_header(token),
|
||||
json={"method": "zero_based", "year": 2026})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["method"] == "zero_based"
|
||||
assert data["total_budget"] == 71.4
|
||||
assert len(data["applied"]) == 4
|
||||
codes = {a["kpi_code"] for a in data["applied"]}
|
||||
assert codes == {"F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_GROSS_MARGIN"}
|
||||
revenue = [a for a in data["applied"] if a["kpi_code"] == "F_REVENUE"][0]
|
||||
assert revenue["budget_value"] == 71.4
|
||||
|
||||
def test_apply_method_unknown(self, client: TestClient, db: Session):
|
||||
"""未知预算方法 → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/apply-method", headers=auth_header(token),
|
||||
json={"method": "bogus"})
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_apply_method_no_kpis(self, client: TestClient, db: Session):
|
||||
"""无核心KPI可应用 → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
resp = client.post(f"{self.BASE}/apply-method", headers=auth_header(token),
|
||||
json={"method": "zero_based", "year": 2026})
|
||||
assert resp.status_code == 400
|
||||
assert "未找到" in resp.json()["detail"]
|
||||
|
||||
|
||||
class TestBudgetSyncCashPlans:
|
||||
"""预算→现金流联动（补齐未覆盖端点）"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_sync_cash_plans_create_and_update(self, client: TestClient, db: Session):
|
||||
"""按预算KPI生成/更新收付款计划（upsert幂等）"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi_rev = create_test_kpi(db, kpi_code="SYNC_REV_KPI", kpi_name="营业收入")
|
||||
kpi_pay = create_test_kpi(db, kpi_code="SYNC_PAY_KPI", kpi_name="费用总额")
|
||||
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi_rev.id, "period": "2026-06", "budget_value": 1000.0})
|
||||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||||
json={"kpi_id": kpi_pay.id, "period": "2026-06", "budget_value": 500.0})
|
||||
|
||||
resp = client.post(f"{self.BASE}/sync-cash-plans", headers=auth_header(token))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["created"] == 2
|
||||
assert data["updated"] == 0
|
||||
|
||||
# 再次执行 → 更新而非新增
|
||||
resp2 = client.post(f"{self.BASE}/sync-cash-plans", headers=auth_header(token))
|
||||
assert resp2.status_code == 200
|
||||
assert resp2.json()["created"] == 0
|
||||
assert resp2.json()["updated"] == 2
|
||||
|
||||
plans = db.query(CashPlan).all()
|
||||
types = {p.plan_type for p in plans}
|
||||
assert types == {"receive", "pay"}
|
||||
|
||||
|
||||
class TestBudgetGenerateCandidates:
|
||||
"""KPI→预算候选列表（补齐未覆盖端点）"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_kpi_budget_candidates(self, client: TestClient, db: Session):
|
||||
"""按类型分类返回预算建议（降本/增收/能力/系统）"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
|
||||
kpi_cost = create_test_kpi(db, kpi_code="CAND_COST", category="cost_control", target_value=50.0)
|
||||
db.add(KPIValue(kpi_id=kpi_cost.id, period="2026-06", actual_value=80.0))
|
||||
create_test_kpi(db, kpi_code="CAND_REV", category="revenue_growth", target_value=100.0)
|
||||
create_test_kpi(db, kpi_code="CAND_CAP", category="talent_pipeline", target_value=100.0)
|
||||
create_test_kpi(db, kpi_code="CAND_SYS", category="supply_chain", target_value=100.0)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"{self.BASE}/kpi-budget-candidates", headers=auth_header(token),
|
||||
params={"year": 2026})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()["data"]
|
||||
assert set(data.keys()) == {"cost_reduction", "revenue_growth", "capability", "system"}
|
||||
|
||||
assert len(data["cost_reduction"]) == 1
|
||||
assert data["cost_reduction"][0]["suggested_budget"] == 9.0 # (80-50)*0.3
|
||||
assert data["revenue_growth"][0]["suggested_budget"] == 20.0 # 100*0.2
|
||||
assert data["capability"][0]["suggested_budget"] == 20000.0 # 2000*10
|
||||
assert data["system"][0]["suggested_budget"] == 15.0 # 100*0.15
|
||||
|
||||
def test_generate_from_kpis(self, client: TestClient, db: Session):
|
||||
"""从选中KPI生成预算科目（修复source_type/source_kpi_id/calc_logic字段后完整流程）"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="GEN_KPI_01")
|
||||
|
||||
resp = client.post(
|
||||
f"{self.BASE}/generate-from-kpis",
|
||||
headers=auth_header(token),
|
||||
json={
|
||||
"year": 2026,
|
||||
"month": 7,
|
||||
"version": "v1.0",
|
||||
"items": [
|
||||
{"kpi_id": kpi.id, "budget_amount": 30000.0,
|
||||
"calc_logic": "目标值100000×0.3=30000", "calc_type": "增收类"}
|
||||
],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["total_amount"] == 30000.0
|
||||
assert body["items"][0]["kpi_code"] == "GEN_KPI_01"
|
||||
|
||||
# 验证落库字段
|
||||
plan = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).first()
|
||||
assert plan is not None
|
||||
assert plan.source_type == "kpi_generated"
|
||||
assert plan.source_kpi_id == kpi.id
|
||||
assert plan.calc_logic == "目标值100000×0.3=30000"
|
||||
assert "增收类" in plan.remark
|
||||
|
||||
def test_generate_from_kpis_empty_items(self, client: TestClient, db: Session):
|
||||
"""未选择KPI → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
resp = client.post(
|
||||
f"{self.BASE}/generate-from-kpis",
|
||||
headers=auth_header(token),
|
||||
json={"year": 2026, "month": 7, "items": []},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_generate_from_kpis_updates_existing(self, client: TestClient, db: Session):
|
||||
"""同一KPI+期间+版本已存在 → 更新预算值而非新增"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="GEN_KPI_02")
|
||||
|
||||
# 先创建一条预算
|
||||
resp1 = client.post(
|
||||
f"{self.BASE}/generate-from-kpis",
|
||||
headers=auth_header(token),
|
||||
json={"year": 2026, "month": 7, "version": "v1.0",
|
||||
"items": [{"kpi_id": kpi.id, "budget_amount": 10000.0}]},
|
||||
)
|
||||
assert resp1.status_code == 200
|
||||
first_id = resp1.json()["items"][0]["plan_id"]
|
||||
|
||||
# 再次生成 → 更新而非新增
|
||||
resp2 = client.post(
|
||||
f"{self.BASE}/generate-from-kpis",
|
||||
headers=auth_header(token),
|
||||
json={"year": 2026, "month": 7, "version": "v1.0",
|
||||
"items": [{"kpi_id": kpi.id, "budget_amount": 25000.0}]},
|
||||
)
|
||||
assert resp2.status_code == 200
|
||||
assert resp2.json()["items"][0]["plan_id"] == first_id
|
||||
|
||||
plans = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).all()
|
||||
assert len(plans) == 1
|
||||
assert plans[0].budget_value == 25000.0
|
||||
|
||||
|
||||
class TestBudgetDeviationReportEdges:
|
||||
"""偏差报告边界：under_budget / 维度过滤"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_deviation_report_under_budget(self, client: TestClient, db: Session):
|
||||
"""实际低于预算 → 结余统计"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_DEV_UNDER")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=100.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=80.0))
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"{self.BASE}/deviation-report?year=2026&month=6",
|
||||
headers=auth_header(token))
|
||||
assert resp.status_code == 200
|
||||
summary = resp.json()["summary"]
|
||||
assert summary["over_budget"] == 0
|
||||
assert summary["under_budget"] == 1
|
||||
|
||||
def test_deviation_report_dimension_filter(self, client: TestClient, db: Session):
|
||||
"""按维度过滤KPI"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_DEV_DIM", dimension="customer")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=100.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=120.0))
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"{self.BASE}/deviation-report?year=2026&month=6&dimension=customer",
|
||||
headers=auth_header(token))
|
||||
assert resp.json()["summary"]["total_kpis"] == 1
|
||||
|
||||
resp2 = client.get(f"{self.BASE}/deviation-report?year=2026&month=6&dimension=finance",
|
||||
headers=auth_header(token))
|
||||
assert resp2.json()["summary"]["total_kpis"] == 0
|
||||
|
||||
|
||||
class TestBudgetRollForwardEdges:
|
||||
"""滚动延展边界：固定模式 / 配置异常"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_roll_forward_fixed_mode(self, client: TestClient, db: Session):
|
||||
"""已配置固定预算模式 → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
client.post(f"{self.BASE}/config", headers=auth_header(token),
|
||||
json={"mode": "fixed"})
|
||||
|
||||
resp = client.post(f"{self.BASE}/roll-forward", headers=auth_header(token))
|
||||
assert resp.status_code == 400
|
||||
assert "固定预算模式" in resp.json()["detail"]
|
||||
|
||||
def test_roll_forward_bad_config(self, client: TestClient, db: Session):
|
||||
"""预算模式配置为非法JSON → 400"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
db.add(SystemConfig(config_key="budget_mode", config_value="not-json"))
|
||||
db.commit()
|
||||
|
||||
resp = client.post(f"{self.BASE}/roll-forward", headers=auth_header(token))
|
||||
assert resp.status_code == 400
|
||||
assert "配置异常" in resp.json()["detail"]
|
||||
|
||||
|
||||
class TestBudgetComparisonRolling:
|
||||
"""滚动预算下的实际vs预测对比"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_comparison_rolling_mode(self, client: TestClient, db: Session):
|
||||
"""滚动模式下只返回 rolling_months 个月份"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
client.post(f"{self.BASE}/config", headers=auth_header(token),
|
||||
json={"mode": "rolling", "rolling_months": 3})
|
||||
|
||||
resp = client.get(f"{self.BASE}/comparison", headers=auth_header(token))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["budget_mode"] == "rolling"
|
||||
assert len(data["periods"]) == 3
|
||||
|
||||
|
||||
class TestBudgetDeviationCheckEdges:
|
||||
"""偏差预警边界：critical / 幂等 / 零预算跳过"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_deviation_check_critical(self, client: TestClient, db: Session):
|
||||
"""偏差超过50% → critical"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_ALERT_CRIT")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=100.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=200.0)) # +100%
|
||||
db.commit()
|
||||
|
||||
resp = client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||||
json={"period": "2026-06", "threshold": 20})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["alerts_generated"] == 1
|
||||
assert resp.json()["alerts"][0]["alert_level"] == "critical"
|
||||
|
||||
def test_deviation_check_idempotent(self, client: TestClient, db: Session):
|
||||
"""重复检查不重复生成预警"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_ALERT_IDEM")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=100.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=150.0))
|
||||
db.commit()
|
||||
|
||||
resp1 = client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||||
json={"period": "2026-06", "threshold": 20})
|
||||
assert resp1.json()["alerts_generated"] == 1
|
||||
|
||||
resp2 = client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||||
json={"period": "2026-06", "threshold": 20})
|
||||
assert resp2.json()["alerts_generated"] == 0
|
||||
|
||||
def test_deviation_check_zero_budget_skip(self, client: TestClient, db: Session):
|
||||
"""预算为0 → 跳过不预警"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_ALERT_ZERO")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=0.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=50.0))
|
||||
db.commit()
|
||||
|
||||
resp = client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||||
json={"period": "2026-06", "threshold": 20})
|
||||
assert resp.json()["alerts_generated"] == 0
|
||||
|
||||
|
||||
class TestBudgetDeviationAlertsEdges:
|
||||
"""偏差预警列表按 kpi_id/period 过滤"""
|
||||
|
||||
BASE = "/api/cma/budget"
|
||||
|
||||
def test_list_alerts_kpi_period_filter(self, client: TestClient, db: Session):
|
||||
"""按KPI与期间过滤预警"""
|
||||
create_test_user(db)
|
||||
token = get_token_for_user(client)
|
||||
kpi = create_test_kpi(db, kpi_code="BUDGET_DEV_ALERT_FLT")
|
||||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=100.0,
|
||||
budget_year=2026, budget_month=6, status="active"))
|
||||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=200.0))
|
||||
db.commit()
|
||||
client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||||
json={"period": "2026-06"})
|
||||
|
||||
resp = client.get(f"{self.BASE}/deviation-alerts", headers=auth_header(token),
|
||||
params={"kpi_id": kpi.id, "period": "2026-06"})
|
||||
assert resp.json()["total"] == 1
|
||||
|
||||
resp2 = client.get(f"{self.BASE}/deviation-alerts", headers=auth_header(token),
|
||||
params={"kpi_id": kpi.id, "period": "2026-01"})
|
||||
assert resp2.json()["total"] == 0
|
||||
|
||||
Reference in New Issue
Block a user