Hermes CI Fix
9822fbd9ba
- alert_rules: 新增 rule_type=customer_concentration（单一/前N客户收入占比超阈值触发，黄=超阈值/红=阈值x倍数） 数据源 zxt_masterbill 客户(往来单位)收入聚合，unit_id<=0 占位剔除；无客户数据可回退 KPI 口径 复用现有 KPIAlert 生成链路：check-all 内联分支 + 专用端点 POST /run-customer-concentration - alert_generator(每日9:00): 新增规则引擎步骤，复用上述检查函数跑 forecast_deviation/customer_concentration - AlertList.vue: 规则类型筛选/编辑项 + 参数表单 + 客户集中度检查按钮 - 测试: tests/test_customer_concentration_rule.py 17 条（规则注册/聚合/双数据源/黄红灯/去重/check-all） - 验证脚本: scripts/verify_forecast_deviation_rules_e2e.py / verify_customer_concentration_rule_e2e.py
272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""
|
||
客户集中度预警规则测试（P2 2026-09-10）
|
||
|
||
覆盖：
|
||
1. 规则类型 customer_concentration 创建/校验（含非法类型拒绝）
|
||
2. _customer_revenue_shares 客户收入聚合（zxt_masterbill 数据源）
|
||
3. _check_customer_concentration 双数据源（客户收入 / KPI 口径）+ 黄红灯分级
|
||
4. check-all 与 /run-customer-concentration 触发链路 + 去重
|
||
"""
|
||
import pytest
|
||
from sqlalchemy import text as sa_text
|
||
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 import KPIAlert, KPIValue
|
||
from app.api.alert_rules import _check_customer_concentration, _customer_revenue_shares
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate_zxt_masterbill():
|
||
"""zxt_masterbill 是原生表（不在 Base.metadata 里），conftest 的 drop_all 清不掉它，
|
||
这里在每个用例前后显式丢弃，保证用例间数据隔离。"""
|
||
from app.database import get_engine
|
||
|
||
def _drop():
|
||
try:
|
||
with get_engine().connect() as conn:
|
||
conn.execute(sa_text("DROP TABLE IF EXISTS zxt_masterbill"))
|
||
conn.commit()
|
||
except Exception:
|
||
pass
|
||
|
||
_drop()
|
||
yield
|
||
_drop()
|
||
|
||
|
||
def _seed_masterbill(db: Session, rows, entity_id: int = 1):
|
||
"""在测试库建 zxt_masterbill 并写入 (unit_id, amount, bill_date) 测试数据"""
|
||
db.execute(sa_text("""
|
||
CREATE TABLE IF NOT EXISTS zxt_masterbill (
|
||
id INTEGER PRIMARY KEY,
|
||
entity_id INTEGER,
|
||
unit_id INTEGER,
|
||
sum_money FLOAT,
|
||
bill_date DATE
|
||
)
|
||
"""))
|
||
for i, (unit_id, amount, bill_date) in enumerate(rows, start=1):
|
||
db.execute(sa_text(
|
||
"INSERT INTO zxt_masterbill (id, entity_id, unit_id, sum_money, bill_date) "
|
||
"VALUES (:id, :eid, :uid, :amt, :bd)"
|
||
), {"id": i, "eid": entity_id, "uid": unit_id, "amt": amount, "bd": bill_date})
|
||
db.commit()
|
||
|
||
|
||
class TestRuleTypeRegistration:
|
||
"""规则类型注册与校验"""
|
||
|
||
def test_create_customer_concentration_rule(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_T1", kpi_name="客户集中度", dimension="customer")
|
||
|
||
resp = client.post("/api/cma/alert-rules", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id,
|
||
"rule_type": "customer_concentration",
|
||
"trigger_on": "actual",
|
||
"params": {"threshold_pct": 50},
|
||
})
|
||
assert resp.status_code == 200, resp.text
|
||
data = resp.json()["data"]
|
||
assert data["rule_type"] == "customer_concentration"
|
||
assert data["entity_id"] == 1
|
||
|
||
def test_reject_unknown_rule_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_T2")
|
||
|
||
resp = client.post("/api/cma/alert-rules", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id,
|
||
"rule_type": "no_such_type",
|
||
})
|
||
assert resp.status_code == 400
|
||
|
||
def test_batch_create_skips_unknown_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_T3")
|
||
|
||
resp = client.post("/api/cma/alert-rules/batch", headers=auth_header(token), json={
|
||
"rules": [
|
||
{"kpi_id": kpi.id, "rule_type": "customer_concentration", "params": {"threshold_pct": 50}},
|
||
{"kpi_id": kpi.id, "rule_type": "bogus_type"},
|
||
]
|
||
})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["created"] == 1
|
||
|
||
|
||
class TestCustomerRevenueAggregation:
|
||
"""客户收入聚合（zxt_masterbill）"""
|
||
|
||
def test_shares_top1_and_top2(self, db: Session):
|
||
_seed_masterbill(db, [
|
||
(101, 600.0, "2026-08-01"),
|
||
(102, 300.0, "2026-08-02"),
|
||
(103, 100.0, "2026-08-03"),
|
||
])
|
||
shares, total, rng = _customer_revenue_shares(db, 1, lookback_months=12, top_n=1)
|
||
assert total == 1000.0
|
||
assert len(shares) == 1
|
||
assert shares[0][0] == 101
|
||
assert shares[0][2] == 60.0
|
||
|
||
shares2, total2, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=2)
|
||
assert total2 == 1000.0
|
||
assert round(sum(s[2] for s in shares2), 2) == 90.0
|
||
|
||
def test_no_data_returns_empty(self, db: Session):
|
||
shares, total, rng = _customer_revenue_shares(db, 1)
|
||
assert shares == [] and total == 0.0 and rng is None
|
||
|
||
def test_entity_isolation(self, db: Session):
|
||
_seed_masterbill(db, [(201, 500.0, "2026-08-01")], entity_id=2)
|
||
shares, total, _ = _customer_revenue_shares(db, 1)
|
||
assert total == 0.0 and shares == []
|
||
|
||
def test_placeholder_unit_excluded(self, db: Session):
|
||
"""unit_id<=0 是占位（无客户/散客），不计入集中度分母"""
|
||
_seed_masterbill(db, [
|
||
(-1, 9000.0, "2026-08-01"),
|
||
(101, 600.0, "2026-08-02"),
|
||
(102, 400.0, "2026-08-03"),
|
||
])
|
||
shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=1)
|
||
assert total == 1000.0
|
||
assert shares[0][0] == 101 and shares[0][2] == 60.0
|
||
|
||
|
||
class TestConcentrationChecker:
|
||
"""占比判定：阈值 / 黄红灯 / 双数据源"""
|
||
|
||
def test_kpi_source_yellow(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_10", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=68.0, entity_id=1))
|
||
db.commit()
|
||
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1)
|
||
assert level == "yellow"
|
||
assert "68.0%" in msg
|
||
|
||
def test_kpi_source_red(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_11", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=85.0, entity_id=1))
|
||
db.commit()
|
||
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1)
|
||
assert level == "red"
|
||
|
||
def test_below_threshold_no_alert(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_12", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=40.0, entity_id=1))
|
||
db.commit()
|
||
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1)
|
||
assert level is None and msg is None
|
||
|
||
def test_no_data_no_alert(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_13", kpi_name="客户集中度", dimension="customer")
|
||
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "kpi"}, db, 1)
|
||
assert level is None and msg is None
|
||
|
||
def test_masterbill_source_with_kpi_fallback_disabled(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_14", kpi_name="客户集中度", dimension="customer")
|
||
# KPI 值虽高，但显式指定 source=zxt_masterbill 且无客户收入数据 → 不触发
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=90.0, entity_id=1))
|
||
db.commit()
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "zxt_masterbill"}, db, 1)
|
||
assert level is None
|
||
|
||
_seed_masterbill(db, [(101, 600.0, "2026-08-01"), (102, 400.0, "2026-08-05")])
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50, "source": "zxt_masterbill"}, db, 1)
|
||
assert level == "yellow"
|
||
assert "单一客户收入占比=60.0%" in msg
|
||
|
||
def test_auto_prefers_masterbill_then_kpi(self, db: Session):
|
||
create_test_user(db)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_15", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=70.0, entity_id=1))
|
||
db.commit()
|
||
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50}, db, 1)
|
||
assert level == "yellow"
|
||
assert "KPI值=70.0%" in msg # 无客户收入数据 → 回退KPI口径
|
||
|
||
_seed_masterbill(db, [(101, 900.0, "2026-08-01"), (102, 100.0, "2026-08-05")])
|
||
level, msg = _check_customer_concentration(kpi.id, {"threshold_pct": 50}, db, 1)
|
||
assert level == "red" # 90% ≥ 50×1.5
|
||
assert "单一客户收入占比=90.0%" in msg
|
||
|
||
|
||
class TestConcentrationTriggerChain:
|
||
"""触发链路：check-all / 专用端点 + 去重"""
|
||
|
||
def _mk_rule(self, client, db, token, kpi, params=None):
|
||
return client.post("/api/cma/alert-rules", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id,
|
||
"rule_type": "customer_concentration",
|
||
"params": params or {"threshold_pct": 50, "source": "kpi"},
|
||
})
|
||
|
||
def test_run_endpoint_generates_alert(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_20", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=72.0, entity_id=1))
|
||
db.commit()
|
||
assert self._mk_rule(client, db, token, kpi).status_code == 200
|
||
|
||
resp = client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token))
|
||
assert resp.status_code == 200, resp.text
|
||
assert resp.json()["generated"] == 1
|
||
|
||
alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all()
|
||
assert len(alerts) == 1
|
||
assert alerts[0].alert_level == "yellow"
|
||
assert alerts[0].entity_id == 1
|
||
assert "客户集中度" in alerts[0].alert_message
|
||
|
||
def test_run_endpoint_dedupes(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_21", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=72.0, entity_id=1))
|
||
db.commit()
|
||
self._mk_rule(client, db, token, kpi)
|
||
|
||
client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token))
|
||
client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token))
|
||
alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all()
|
||
assert len(alerts) == 1 # 第二次是更新而非新增
|
||
|
||
def test_check_all_includes_concentration(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="C_CONC_22", kpi_name="客户集中度", dimension="customer")
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-08", actual_value=80.0, entity_id=1))
|
||
db.commit()
|
||
self._mk_rule(client, db, token, kpi)
|
||
|
||
resp = client.post("/api/cma/alert-rules/check-all", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
alerts = db.query(KPIAlert).filter(KPIAlert.alert_type == "concentration").all()
|
||
assert len(alerts) == 1
|
||
assert alerts[0].alert_level == "red"
|
||
|
||
def test_run_endpoint_without_rule(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post("/api/cma/alert-rules/run-customer-concentration", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["generated"] == 0
|