Hermes CI Fix
1613f7f961
ci/woodpecker/push/main Pipeline failed
Customer revenue/concentration now filters bill_type IN (1,4,14) (XSD/XSTHD/XSHHD). Non-sales docs (JHD purchase, YFKD prepay, YSKD accept, QTRK, SKD, YBFYD) carry SUPPLIER unit_id and were polluting 'customer concentration' (old Top1/Top2 were our biggest suppliers; root cause of 6.04x base gap). Unify sign filter to (sum_money+sum_tax_money)>0 (equivalent to money_only on current data, diff 0 rows) and correct the withdrawn 407-row/17.2% wording in cpa_caliber docstring. Deriver: anchors from cpa_caliber single source; new --concentration overwrite and --only concentration (minimal 9-row write face) with pre-overwrite backup. Tests: fixture gains bill_type/sum_tax_money; +5 whitelist cases (44 passed targeted).
356 lines
17 KiB
Python
356 lines
17 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 触发链路 + 去重
|
||
5. **销售类单据白名单**（P0 纠错 2026-10-04）：bill_type IN (1,4,14) 生效、
|
||
非销售单据（供应商）被排除、预警路径与 CPA 口径同源同式
|
||
"""
|
||
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[, bill_type[, sum_tax_money]]) 测试数据
|
||
|
||
2026-10-04 P0 口径纠错后聚合过滤含**销售类白名单** `bill_type IN (1,4,14)`
|
||
与**统一符号口径** `(sum_money + sum_tax_money) > 0` ⇒ 夹具必须带 `bill_type` /
|
||
`sum_tax_money` 两列（旧夹具只有 sum_money，改用例夹具属为适配新契约的**显式改写**）：
|
||
* bill_type 默认 1 = 销售单 XSD；逐行可指定应被排除的非销售单据
|
||
（0=JHD 进货 / 129=YFKD 预付款 / 17=QTRK 其它入库 / 103=YBFYD 费用 / 128=YSKD 验收
|
||
—— 这些行的 unit_id 是供应商）。
|
||
* sum_tax_money 默认 0.0。
|
||
"""
|
||
db.execute(sa_text("""
|
||
CREATE TABLE IF NOT EXISTS zxt_masterbill (
|
||
id INTEGER PRIMARY KEY,
|
||
entity_id INTEGER,
|
||
unit_id INTEGER,
|
||
bill_type INTEGER,
|
||
sum_money FLOAT,
|
||
sum_tax_money FLOAT,
|
||
bill_date DATE
|
||
)
|
||
"""))
|
||
for i, row in enumerate(rows, start=1):
|
||
unit_id, amount, bill_date = row[0], row[1], row[2]
|
||
bill_type = row[3] if len(row) > 3 else 1
|
||
tax = row[4] if len(row) > 4 else 0.0
|
||
db.execute(sa_text(
|
||
"INSERT INTO zxt_masterbill (id, entity_id, unit_id, bill_type, sum_money, "
|
||
"sum_tax_money, bill_date) VALUES (:id, :eid, :uid, :bt, :amt, :tax, :bd)"
|
||
), {"id": i, "eid": entity_id, "uid": unit_id, "bt": bill_type, "amt": amount,
|
||
"tax": tax, "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 TestSalesBillTypeWhitelist:
|
||
"""P0 口径纠错（2026-10-04）：单据类型白名单（销售类 bill_type IN (1,4,14)）
|
||
|
||
背景：非销售单据（进货/预付款/验收/入库/收款/费用）的 unit_id 是**供应商**
|
||
（abst 实读「向【酣客酒业（广州）】进货」「预付款单」），混入会把
|
||
「客户集中度」度量成「往来单位（含供应商）集中度」。
|
||
"""
|
||
|
||
def test_purchase_and_prepay_rows_excluded(self, db: Session):
|
||
_seed_masterbill(db, [
|
||
(152, 600.0, "2026-08-01", 1), # XSD 销售单 → 计入
|
||
(193, 300.0, "2026-08-02", 4), # XSTHD 销售退货 → 计入（销售类）
|
||
(146, 9000.0, "2026-08-03", 0), # JHD 进货（供应商）→ 排除
|
||
(146, 8000.0, "2026-08-04", 129), # YFKD 预付款（供应商）→ 排除
|
||
(474, 7000.0, "2026-08-05", 17), # QTRK 其它入库（供应商）→ 排除
|
||
(101, 5000.0, "2026-08-06", 103), # YBFYD 一般费用单 → 排除
|
||
(102, 4000.0, "2026-08-07", 128), # YSKD 验收单 → 排除
|
||
])
|
||
shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=2)
|
||
assert total == 900.0
|
||
assert [s[0] for s in shares] == [152, 193]
|
||
assert shares[0][2] == round(600.0 / 900.0 * 100, 2)
|
||
|
||
def test_sales_whitelist_in_generated_sql(self):
|
||
from app.utils import cpa_caliber
|
||
sql = cpa_caliber._sql_window()
|
||
assert "bill_type IN (1, 4, 14)" in sql # 白名单生效
|
||
assert "(sum_money + sum_tax_money) > 0" in sql # 统一书面口径
|
||
assert sql == cpa_caliber._sql_window("tax_incl", "sales")
|
||
all_sql = cpa_caliber._sql_window(bill_scope="all")
|
||
assert "bill_type" not in all_sql # 对拍用全单据口径不带白名单
|
||
|
||
def test_unknown_scope_and_sign_rejected(self):
|
||
from app.utils import cpa_caliber
|
||
with pytest.raises(ValueError):
|
||
cpa_caliber._sql_window(bill_scope="bogus")
|
||
with pytest.raises(ValueError):
|
||
cpa_caliber._sql_window(sign_basis="bogus")
|
||
|
||
def test_sign_caliber_semantics_on_negative_row(self, db: Session):
|
||
"""负额行（sum_money<0、含税>0）：tax_incl 计入 / money_only 排除 ⇒ 两式**非恒等**。
|
||
|
||
本库当前数据**不存在**此类行（2026-01~09 双向差集 0 行；全表亦 0 行），
|
||
故「两口径等价」是**数据事实**（财务复验 §5.1），不是恒等式——本用例固化语义，
|
||
防后人据「等价」误删符号过滤。
|
||
"""
|
||
from app.database import get_engine
|
||
from app.utils import cpa_caliber
|
||
_seed_masterbill(db, [(999, -100.0, "2026-08-03", 1, 130.0)])
|
||
incl = cpa_caliber.fetch_monthly_revenue(get_engine(), 1, sign_basis="tax_incl")
|
||
money = cpa_caliber.fetch_monthly_revenue(get_engine(), 1, sign_basis="money_only")
|
||
assert incl.get("2026-08") == [(999, -100.0)] # 含税净额 +30 > 0 ⇒ 计入
|
||
assert "2026-08" not in money # sum_money -100 ⇒ 排除
|
||
|
||
def test_alert_path_shares_single_caliber_with_cpa(self, db: Session):
|
||
"""预警路径与 CPA 派生器同源同式（防「同一客户收入两份结果」）"""
|
||
from datetime import date, timedelta
|
||
from app.database import get_engine
|
||
from app.utils import cpa_caliber
|
||
_seed_masterbill(db, [(152, 600.0, "2026-08-01", 1), (146, 9000.0, "2026-08-03", 0)])
|
||
shares, total, _ = _customer_revenue_shares(db, 1, lookback_months=12, top_n=5)
|
||
cpa = cpa_caliber.fetch_revenue_rows(
|
||
get_engine(), 1, date.today() - timedelta(days=360), date.today(), 5)
|
||
assert shares == cpa[0] and total == cpa[1] == 600.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
|