wecom-fullstack
477515c453
ci/woodpecker/push/main Pipeline was canceled
任务 dama-entity-isolation-sweep-20260923（DAMA 独立回查漏项 / 任总 2026-09-23「按照建议执行」）
① 补漏同源全局键 kpi_alert_higher_better（system_configs，零 DDL）
- deviation_engine: 新增 higher_better_key(entity_id)；get_higher_better_codes(db, entity_id)
改为只读本账套键、不回退旧全局键；未配置 → DEFAULT_HIGHER_BETTER（兜底，与原行为逐字一致）
- budget.py: GET/PUT /alert-direction 挂 get_entity_id，只读写会话账套键；body entity_id 不信
- scripts/migrate_higher_better_key_entity_20260923.py: --check/--apply/--down + 库名保护 + 幂等
- 生产实况：库内该键 0 行 ⇒ 无原值可搬运（--apply 走 skip 零写入），e1 行为不变
② kpi_visibility 口径裁定：角色级天然全局 = 非账套维度，不拆（关闭本项）
- auth_middleware 注释 + docstring 登记「有意不做账套后缀」；口径字典
docs/global-config-keys-caliber-20260923.md（含 budget_mode/budget_driver_mode 评估结论）
③ tenant/switch 审计收口
- detail 显式 from_entity_id → to_entity_id + source + operator{id,username,name,role}
- 审计失败不静默：同事务回滚 + AUDIT_WRITE_FAILED ERROR + HTTP 500（不留「改了但没审计」）
- GET /current 无写路径不落审计；历史缺口不补写
证据：新增/改写用例 17 条；全量 1397 passed（去新增文件 1380 + 17）；RED 态 8 failed
488 lines
22 KiB
Python
488 lines
22 KiB
Python
"""预算系统技术改进测试 (2026-08-28 yanxue-budget-tech-improve)
|
||
|
||
覆盖:
|
||
P1-③ 告警归因(alert_type/attribution/scenario_id + 详情接口)
|
||
P1-④ 实际值自动归集(映射CRUD/采集器/覆盖率)
|
||
P2-① 真零基逐项论证(CRUD/generate/method-comparison is_demo)
|
||
P2-② 派生规则可配置(规则CRUD/apply-method rule_source)
|
||
P2-⑤ 双路径合并(两出口级别一致, 无第二套阈值逻辑)
|
||
P2-⑥ 现金流分类规则(待分类队列/一键归类)
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
import datetime
|
||
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
from app.models import (
|
||
BudgetPlan, KPIValue, BudgetDeviationAlert, KPIAlert,
|
||
KPIValueSource, KPIValueCollectLog, VoucherDetail,
|
||
BudgetZeroBasedItem, BudgetDerivationRule,
|
||
CashPlanClassifyRule, CashPlanUnclassified, CashPlan,
|
||
ScenarioSuggestion, KPIDefinition,
|
||
)
|
||
|
||
|
||
def _seed_vouchers(db, period="2026-06", entity_id=1, credit=1500.0):
|
||
"""给自动归集造源数据（voucher_details 贷方发生额）。
|
||
|
||
采集器口径（d71c7fe「自动归集空汇总不写0」）：源表无记录 → 返回 None 并记 skipped，
|
||
不落 0 值 KPIValue。故本类用例必须先造源数据，否则 collected 恒为 0。
|
||
"""
|
||
y, m = int(period[:4]), int(period[5:7])
|
||
db.add(VoucherDetail(
|
||
entity_id=entity_id, voucher_no="COLLECT-001",
|
||
voucher_date=datetime.datetime(y, m, 15),
|
||
subject_code="6001", subject_name="主营业务收入",
|
||
debit_amount=0.0, credit_amount=credit,
|
||
summary="自动归集单测", carry_forward=0, period=period,
|
||
))
|
||
db.flush()
|
||
|
||
|
||
|
||
class TestP1AlertAttribution:
|
||
"""P1-③ 告警归因: 告警从'差多少'到'差在哪+怎么办'"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def _setup_alert(self, client, db, kpi_code="ATTRIB_KPI", kpi_name="销售费用", actual=150.0, budget=100.0):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code=kpi_code, kpi_name=kpi_name)
|
||
db.add(BudgetPlan(kpi_id=kpi.id, period="2026-06", budget_value=budget,
|
||
budget_year=2026, budget_month=6, status="active"))
|
||
db.add(KPIValue(kpi_id=kpi.id, period="2026-06", actual_value=actual))
|
||
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
|
||
return token, kpi, resp
|
||
|
||
def test_deviation_check_writes_attribution(self, client: TestClient, db: Session):
|
||
"""生成告警时同步写 alert_type/attribution/scenario_id"""
|
||
token, kpi, resp = self._setup_alert(client, db, actual=150.0, budget=100.0)
|
||
assert resp.json()["alerts_generated"] == 1
|
||
|
||
alert = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.kpi_id == kpi.id).first()
|
||
assert alert is not None
|
||
assert alert.alert_level == "warning"
|
||
# 归因JSON结构
|
||
assert alert.attribution is not None
|
||
attr = alert.attribution
|
||
assert "dimensions" in attr and "subjects" in attr
|
||
assert "variance_type" in attr and "trend" in attr
|
||
assert attr["variance_type"] in ("quantity_diff", "price_diff", "mixed")
|
||
assert "anomaly" in attr["trend"]
|
||
# 场景建议关联（费用类KPI → cost_high 模板）
|
||
if alert.scenario_id:
|
||
s = db.query(ScenarioSuggestion).filter(ScenarioSuggestion.id == alert.scenario_id).first()
|
||
assert s is not None
|
||
assert s.alert_type in ("cash_low", "cash_critical", "cost_high", "revenue_drop")
|
||
|
||
def test_attribution_detail_endpoint(self, client: TestClient, db: Session):
|
||
"""GET /deviation-alerts/{id}/attribution 返回归因+场景建议"""
|
||
token, kpi, _ = self._setup_alert(client, db, actual=200.0, budget=100.0)
|
||
alert = db.query(BudgetDeviationAlert).filter(BudgetDeviationAlert.kpi_id == kpi.id).first()
|
||
|
||
resp = client.get(f"{self.BASE}/deviation-alerts/{alert.id}/attribution",
|
||
headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["attribution"] != {}
|
||
assert "dimensions" in data["attribution"]
|
||
# scenario 建议联查（无匹配时可空, 有模板时必须带文本）
|
||
if data["scenario"]:
|
||
assert data["scenario"]["title"]
|
||
|
||
def test_list_alerts_has_attribution_fields(self, client: TestClient, db: Session):
|
||
"""列表响应新增 alert_type/attribution/scenario_id 字段（可空）"""
|
||
token, kpi, _ = self._setup_alert(client, db)
|
||
resp = client.get(f"{self.BASE}/deviation-alerts", headers=auth_header(token))
|
||
row = resp.json()["data"][0]
|
||
assert "alert_type" in row
|
||
assert "attribution" in row
|
||
assert "scenario_id" in row
|
||
|
||
def test_alert_direction_config(self, client: TestClient, db: Session):
|
||
"""P2-⑤ 方向配置 GET/PUT system_configs"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.get(f"{self.BASE}/alert-direction", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert "SALES_TOTAL" in resp.json()["codes"]
|
||
|
||
resp2 = client.put(f"{self.BASE}/alert-direction", headers=auth_header(token),
|
||
json={"codes": ["SALES_TOTAL", "CUSTOM_COUNT"]})
|
||
assert resp2.status_code == 200
|
||
assert resp2.json()["codes"] == ["SALES_TOTAL", "CUSTOM_COUNT"]
|
||
|
||
resp3 = client.get(f"{self.BASE}/alert-direction", headers=auth_header(token))
|
||
assert resp3.json()["codes"] == ["SALES_TOTAL", "CUSTOM_COUNT"]
|
||
assert resp3.json()["is_configured"] is True
|
||
|
||
|
||
class TestP1ValueCollect:
|
||
"""P1-④ 实际值自动归集"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_value_source_crud_and_collect(self, client: TestClient, db: Session):
|
||
"""映射CRUD → 采集器 → kpi_values 出现 auto_collect"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="COLLECT_KPI")
|
||
_seed_vouchers(db) # 源数据（无源数据采集器按设计跳过）
|
||
|
||
# 建映射
|
||
resp = client.post(f"{self.BASE}/value-sources", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id,
|
||
"source_table": "voucher_details",
|
||
"source_field": "credit_amount",
|
||
"aggregate": "sum",
|
||
"filter_rule": {"direction": "credit"},
|
||
"period_field": "period",
|
||
"unit_conversion": 1,
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
# 采集器试跑（不写库）
|
||
test_resp = client.post(f"{self.BASE}/value-sources/test", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id,
|
||
"source_table": "voucher_details",
|
||
"source_field": "credit_amount",
|
||
"aggregate": "sum",
|
||
"filter_rule": {"direction": "credit"},
|
||
"period_field": "period",
|
||
"period": "2026-06", # 显式期间（不传则默认当月，与造数期间错配 → value=None）
|
||
})
|
||
assert test_resp.status_code == 200
|
||
assert test_resp.json()["value"] is not None
|
||
assert test_resp.json()["value"] == 1500.0 # 贷方发生额合计（unit_conversion=1）
|
||
|
||
# 手动触发采集
|
||
run_resp = client.post(f"{self.BASE}/value-collect/run", headers=auth_header(token),
|
||
json={"period": "2026-06"})
|
||
assert run_resp.status_code == 200
|
||
assert run_resp.json()["collected"] >= 1
|
||
|
||
# 验证 kpi_values 落库
|
||
val = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.period == "2026-06",
|
||
KPIValue.source_type == "auto_collect",
|
||
).first()
|
||
assert val is not None
|
||
assert val.actual_value is not None
|
||
assert val.remark and "自动归集" in val.remark
|
||
|
||
# 采集日志
|
||
logs = db.query(KPIValueCollectLog).filter(KPIValueCollectLog.kpi_id == kpi.id).all()
|
||
assert len(logs) >= 1
|
||
|
||
# 覆盖率
|
||
cov = client.get(f"{self.BASE}/value-sources/coverage", headers=auth_header(token))
|
||
assert cov.status_code == 200
|
||
assert cov.json()["mapped_count"] >= 1
|
||
|
||
def test_collector_idempotent(self, client: TestClient, db: Session):
|
||
"""同kpi+period 重复采集 → 更新不新增"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="COLLECT_IDEMP")
|
||
_seed_vouchers(db)
|
||
client.post(f"{self.BASE}/value-sources", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "source_table": "voucher_details",
|
||
"source_field": "credit_amount", "aggregate": "sum",
|
||
})
|
||
client.post(f"{self.BASE}/value-collect/run", headers=auth_header(token), json={"period": "2026-06"})
|
||
client.post(f"{self.BASE}/value-collect/run", headers=auth_header(token), json={"period": "2026-06"})
|
||
|
||
rows = db.query(KPIValue).filter(
|
||
KPIValue.kpi_id == kpi.id,
|
||
KPIValue.period == "2026-06",
|
||
KPIValue.source_type == "auto_collect",
|
||
).all()
|
||
assert len(rows) == 1
|
||
|
||
def test_collector_logs_filter(self, client: TestClient, db: Session):
|
||
"""采集日志 status 过滤"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="COLLECT_LOG")
|
||
_seed_vouchers(db)
|
||
client.post(f"{self.BASE}/value-sources", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "source_table": "voucher_details",
|
||
"source_field": "credit_amount", "aggregate": "sum",
|
||
})
|
||
client.post(f"{self.BASE}/value-collect/run", headers=auth_header(token), json={"period": "2026-06"})
|
||
resp = client.get(f"{self.BASE}/value-collect/logs", headers=auth_header(token),
|
||
params={"status": "success"})
|
||
assert resp.json()["total"] >= 1
|
||
|
||
|
||
class TestP2ZeroBased:
|
||
"""P2-① 真零基逐项论证"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def _setup_kpi_with_plans(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
# 核心4KPI（apply-method 需要）
|
||
for code in ("F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_GROSS_MARGIN"):
|
||
create_test_kpi(db, kpi_code=code, kpi_name=code)
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_REVENUE").first()
|
||
return token, kpi
|
||
|
||
def test_zero_based_items_crud_and_generate(self, client: TestClient, db: Session):
|
||
"""录入3个科目 → 逐项论证 → generate → budget_plans 出现且金额=Σ建议值"""
|
||
token, kpi = self._setup_kpi_with_plans(client, db)
|
||
|
||
# 录入3个论证项
|
||
items = [
|
||
{"item_name": "房租", "item_category": "fixed", "base_value": 15, "proposed_value": 15, "justification": "合同锁定"},
|
||
{"item_name": "招待费", "item_category": "discretionary", "base_value": 16, "proposed_value": 8, "justification": "压缩50%"},
|
||
{"item_name": "杂项", "item_category": "discretionary", "base_value": 12, "proposed_value": 8, "justification": "压缩30%"},
|
||
]
|
||
for it in items:
|
||
r = client.post(f"{self.BASE}/zero-based/items", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "period": "2026-06", **it,
|
||
})
|
||
assert r.status_code == 200
|
||
|
||
# 列表+合计
|
||
lst = client.get(f"{self.BASE}/zero-based/items", headers=auth_header(token),
|
||
params={"kpi_id": kpi.id, "period": "2026-06"})
|
||
assert lst.json()["total"] == 3
|
||
assert lst.json()["total_proposed"] == 31.0
|
||
|
||
# generate → budget_plans
|
||
gen = client.post(f"{self.BASE}/zero-based/generate", headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06"})
|
||
assert gen.status_code == 200
|
||
assert gen.json()["total"] == 31.0
|
||
|
||
plan = db.query(BudgetPlan).filter(
|
||
BudgetPlan.kpi_id == kpi.id,
|
||
BudgetPlan.period == "2026-06",
|
||
BudgetPlan.version.like("zbb-%"),
|
||
).first()
|
||
assert plan is not None
|
||
assert plan.budget_value == 31.0
|
||
assert plan.calc_logic == "zero_based_itemized"
|
||
|
||
def test_method_comparison_zero_based_is_demo_false(self, client: TestClient, db: Session):
|
||
"""method-comparison 传论证KPI → is_demo=false; 不传 → is_demo=true"""
|
||
token, kpi = self._setup_kpi_with_plans(client, db)
|
||
client.post(f"{self.BASE}/zero-based/items", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "period": "2026-06",
|
||
"item_name": "房租", "item_category": "fixed",
|
||
"base_value": 15, "proposed_value": 15,
|
||
})
|
||
|
||
# 有论证项 → 真零基
|
||
r1 = client.post(f"{self.BASE}/method-comparison", headers=auth_header(token), json={
|
||
"zero_based_kpi_id": kpi.id, "zero_based_period": "2026-06",
|
||
})
|
||
zbb1 = [m for m in r1.json()["methods"] if m["id"] == "zero_based"][0]
|
||
assert zbb1["is_demo"] is False
|
||
assert zbb1["item_count"] == 1
|
||
|
||
# 无论证项 → demo fallback
|
||
r2 = client.post(f"{self.BASE}/method-comparison", headers=auth_header(token), json={})
|
||
zbb2 = [m for m in r2.json()["methods"] if m["id"] == "zero_based"][0]
|
||
assert zbb2["is_demo"] is True
|
||
|
||
def test_apply_method_zero_based_writes_plan(self, client: TestClient, db: Session):
|
||
"""apply-method zero_based → 落库 zbb 版本"""
|
||
token, kpi = self._setup_kpi_with_plans(client, db)
|
||
client.post(f"{self.BASE}/zero-based/items", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "period": "2026-06",
|
||
"item_name": "房租", "item_category": "fixed",
|
||
"base_value": 15, "proposed_value": 15,
|
||
})
|
||
r = client.post(f"{self.BASE}/apply-method", headers=auth_header(token), json={
|
||
"method": "zero_based", "year": 2026,
|
||
"zero_based_kpi_id": kpi.id, "zero_based_period": "2026-06",
|
||
})
|
||
assert r.status_code == 200
|
||
|
||
|
||
class TestP2DerivationRules:
|
||
"""P2-② 派生规则可配置"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def _setup(self, client, db):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
for code in ("F_REVENUE", "F_NET_PROFIT", "F_COST_RATIO", "F_GROSS_MARGIN"):
|
||
create_test_kpi(db, kpi_code=code, kpi_name=code)
|
||
return token
|
||
|
||
def test_rule_crud_and_apply(self, client: TestClient, db: Session):
|
||
"""配置 F_NET_PROFIT 派生率 5% → apply-method → rule_source=configured 且结果变化"""
|
||
token = self._setup(client, db)
|
||
rev = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_REVENUE").first()
|
||
np_kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_NET_PROFIT").first()
|
||
|
||
# 无规则时 apply → default 比例(2%)
|
||
r_default = client.post(f"{self.BASE}/apply-method", headers=auth_header(token),
|
||
json={"method": "incremental", "year": 2026})
|
||
np_default = [a for a in r_default.json()["applied"] if a["kpi_code"] == "F_NET_PROFIT"][0]
|
||
assert np_default["rule_source"] == "default"
|
||
assert r_default.json()["rule_source"] == "default"
|
||
|
||
# 建规则: percentage_of → 来源F_REVENUE × 5%
|
||
# 先给来源KPI实际值（真实链路: base_kpi实际值 × rate）
|
||
db.add(KPIValue(kpi_id=rev.id, period="2026-05", actual_value=2000.0))
|
||
db.commit()
|
||
r_rule = client.post(f"{self.BASE}/derivation-rules", headers=auth_header(token), json={
|
||
"kpi_id": np_kpi.id,
|
||
"rule_type": "percentage_of",
|
||
"base_kpi_id": rev.id,
|
||
"params": {"rate": 0.05},
|
||
"formula_text": "净利润 = 营业收入 × 5%",
|
||
})
|
||
assert r_rule.status_code == 200
|
||
|
||
# 配置后 apply → rule_source=configured, 金额=2000×5%=100
|
||
r2 = client.post(f"{self.BASE}/apply-method", headers=auth_header(token),
|
||
json={"method": "incremental", "year": 2026})
|
||
assert r2.json()["rule_source"] == "configured"
|
||
np_after = [a for a in r2.json()["applied"] if a["kpi_code"] == "F_NET_PROFIT"][0]
|
||
assert np_after["rule_source"] == "configured"
|
||
assert np_after["budget_value"] == 100.0
|
||
|
||
# 规则列表
|
||
lst = client.get(f"{self.BASE}/derivation-rules", headers=auth_header(token))
|
||
assert lst.json()["total"] == 1
|
||
assert lst.json()["data"][0]["rule_type"] == "percentage_of"
|
||
|
||
|
||
class TestP2SingleAlertPath:
|
||
"""P2-⑤ 双路径合并: 单一告警逻辑, 两出口级别一致"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_single_build_function_two_exits(self, client: TestClient, db: Session):
|
||
"""run_deviation_check 走统一逻辑写 KPIAlert; deviation-check 写 budget_deviation_alerts"""
|
||
from app.utils.deviation_engine import build_deviation_alert, run_deviation_check
|
||
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="SINGLE_PATH_KPI", kpi_name="测试成本")
|
||
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=160.0)) # 60% 超支
|
||
db.commit()
|
||
|
||
# KPIAlert 出口: 级别 red(≥30)
|
||
r = build_deviation_alert(db, kpi, "2026-06")
|
||
assert r["triggered"] is True
|
||
assert r["kpi_alert_level"] == "red"
|
||
assert r["level"] == "critical" # >50
|
||
|
||
# budget 出口 API: deviation-check
|
||
resp = client.post(f"{self.BASE}/deviation-check", headers=auth_header(token),
|
||
json={"period": "2026-06", "threshold": 20})
|
||
assert resp.json()["alerts_generated"] == 1
|
||
|
||
# run_deviation_check 写 KPIAlert
|
||
n = run_deviation_check(db, "2026-06")
|
||
assert n >= 1
|
||
kpi_alert = db.query(KPIAlert).filter(
|
||
KPIAlert.kpi_id == kpi.id,
|
||
KPIAlert.alert_message.contains("[差异预警]"),
|
||
).first()
|
||
assert kpi_alert is not None
|
||
assert kpi_alert.alert_level == "red"
|
||
assert kpi_alert.suggestion # 非模板空文案
|
||
|
||
def test_no_second_threshold_logic(self, client: TestClient, db: Session):
|
||
"""deviation_engine 中不应再有独立阈值/方向列表（grep 验证在代码review, 此处测函数可用）"""
|
||
from app.utils.deviation_engine import build_deviation_alert, get_higher_better_codes
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="HB_KPI", kpi_name="营业收入")
|
||
# 2026-09-23 拆账套：签名改为 entity_id 必传（漏传即 TypeError，防落默认账套）
|
||
assert "SALES_TOTAL" in get_higher_better_codes(db, 1)
|
||
|
||
|
||
class TestP2CashClassify:
|
||
"""P2-⑥ 现金流分类规则表"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_unclassified_queue_and_classify(self, client: TestClient, db: Session):
|
||
"""无关键词KPI → sync-cash-plans → 待分类队列(不静默跳过) → 一键归类 → CashPlan"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
# 无任何关键词的KPI（不会命中默认关键词）
|
||
kpi = create_test_kpi(db, kpi_code="MYSTERY_KPI", kpi_name="部门专项投入待定")
|
||
# 移除'投入'关键词冲突: 名称改无关键词
|
||
kpi.kpi_name = "神秘专项"
|
||
db.commit()
|
||
|
||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 300.0})
|
||
|
||
# sync → 进待分类队列
|
||
resp = client.post(f"{self.BASE}/sync-cash-plans", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["unclassified_count"] >= 1
|
||
|
||
item = db.query(CashPlanUnclassified).filter(
|
||
CashPlanUnclassified.kpi_id == kpi.id,
|
||
CashPlanUnclassified.status == "pending",
|
||
).first()
|
||
assert item is not None
|
||
assert item.reason == "未匹配任何分类规则"
|
||
|
||
# 队列列表
|
||
lst = client.get(f"{self.BASE}/cash-unclassified", headers=auth_header(token),
|
||
params={"status": "pending"})
|
||
assert any(r["kpi_id"] == kpi.id for r in lst.json()["data"])
|
||
|
||
# 一键归类 receive
|
||
cls = client.post(f"{self.BASE}/cash-unclassified/{item.id}/classify", headers=auth_header(token),
|
||
json={"plan_type": "receive"})
|
||
assert cls.status_code == 200
|
||
assert cls.json()["rule_created"] is True
|
||
|
||
# 规则自动补建 + CashPlan 生成
|
||
rule = db.query(CashPlanClassifyRule).filter(
|
||
CashPlanClassifyRule.entity_id == 1,
|
||
CashPlanClassifyRule.kpi_id == kpi.id,
|
||
).first()
|
||
assert rule is not None and rule.plan_type == "receive"
|
||
|
||
plan = db.query(CashPlan).filter(CashPlan.related_kpi_id == kpi.id).first()
|
||
assert plan is not None and plan.plan_type == "receive"
|
||
|
||
# 队列状态 → classified
|
||
db.refresh(item)
|
||
assert item.status == "classified"
|
||
|
||
def test_rule_priority_over_keyword(self, client: TestClient, db: Session):
|
||
"""规则表精确匹配优先于默认关键词"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="OVERRIDE_KPI", kpi_name="营业收入") # 默认会命中 receive
|
||
|
||
# 规则表强制 pay
|
||
r = client.post(f"{self.BASE}/cash-classify-rules", headers=auth_header(token), json={
|
||
"kpi_id": kpi.id, "plan_type": "pay", "priority": 1,
|
||
})
|
||
assert r.status_code == 200
|
||
|
||
client.post(f"{self.BASE}/plans", headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 500.0})
|
||
client.post(f"{self.BASE}/sync-cash-plans", headers=auth_header(token))
|
||
|
||
plan = db.query(CashPlan).filter(CashPlan.related_kpi_id == kpi.id).first()
|
||
assert plan is not None
|
||
assert plan.plan_type == "pay" # 规则覆盖关键词
|