266 lines
12 KiB
Python
266 lines
12 KiB
Python
"""
|
||
统一阈值解析器 / 建议格式 / 红灯优先判定 测试
|
||
（project-20260911-threshold-fix，2026-09-11）
|
||
|
||
覆盖 4 类缺陷：
|
||
① suggest 生成格式必须可解析（原生成区间/文字 → 保存后静默失效）
|
||
② 前端/接口格式校验（非法表达式被拒）
|
||
③ 解析器支持负数（>=-100000）
|
||
⑤ 红灯优先（宽黄灯不得遮蔽红灯）
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.utils.threshold_parser import (
|
||
LEVEL_PRIORITY,
|
||
calc_level,
|
||
calc_level_from_exprs,
|
||
eval_threshold,
|
||
is_valid_threshold,
|
||
parse_threshold,
|
||
)
|
||
from app.api.thresholds import build_suggestion
|
||
from app.api.kpis import validate_threshold_format
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
|
||
|
||
class TestParseThreshold:
|
||
"""解析器：操作符 / 负数 / 区间 / 多条件 / 非法格式"""
|
||
|
||
def test_scalar_ops(self):
|
||
assert parse_threshold(">=32000000") == (">=", 32000000.0)
|
||
assert parse_threshold("<20") == ("<", 20.0)
|
||
assert parse_threshold("<=55") == ("<=", 55.0)
|
||
assert parse_threshold(">60") == (">", 60.0)
|
||
assert parse_threshold("=100") == ("=", 100.0)
|
||
assert parse_threshold("!=0") == ("!=", 0.0)
|
||
|
||
def test_negative_numbers(self):
|
||
"""缺陷3：原正则 [\\d.]+ 不支持负号 → 现金流类阈值静默失效"""
|
||
assert parse_threshold(">=-100000") == (">=", -100000.0)
|
||
assert parse_threshold("<-100000") == ("<", -100000.0)
|
||
assert eval_threshold(-200000, "<-100000") is True
|
||
assert eval_threshold(-50000, "<-100000") is False
|
||
|
||
def test_whitespace_tolerated(self):
|
||
assert parse_threshold(" >= 800 ") == (">=", 800.0)
|
||
|
||
def test_range(self):
|
||
assert parse_threshold("52.33~107.67") == ("~", (52.33, 107.67))
|
||
assert eval_threshold(80, "52.33~107.67") is True
|
||
assert eval_threshold(52.33, "52.33~107.67") is True # 含端点
|
||
assert eval_threshold(120, "52.33~107.67") is False
|
||
|
||
def test_or_clauses(self):
|
||
assert eval_threshold(130, "<=36|>=124") is True
|
||
assert eval_threshold(20, "<=36|>=124") is True
|
||
assert eval_threshold(80, "<=36|>=124") is False
|
||
|
||
@pytest.mark.parametrize("bad", ["偏离55%", "偏离>55%", "80%", ">=abc", "52.33~", ">=", ">=800%"])
|
||
def test_invalid(self, bad):
|
||
assert is_valid_threshold(bad) is False
|
||
assert parse_threshold(bad) == (None, None)
|
||
assert eval_threshold(100, bad) is False
|
||
|
||
def test_blank_is_valid_means_unset(self):
|
||
assert is_valid_threshold("") is True
|
||
assert is_valid_threshold(None) is True
|
||
assert eval_threshold(100, "") is False
|
||
|
||
def test_partial_or_clause_is_invalid(self):
|
||
"""多条件中任一子句非法 → 整体非法（防半截表达式静默生效）"""
|
||
assert is_valid_threshold("<=36|abc") is False
|
||
|
||
|
||
class TestCalcLevel:
|
||
def test_red_priority_over_yellow(self):
|
||
"""缺陷5：id=6 实测 Y='>76' R='>91'，值 95 原判黄（红灯不可达）→ 必须判红"""
|
||
assert calc_level_from_exprs(95, "<=76", ">76", ">91") == "red"
|
||
assert calc_level_from_exprs(152.62, "<=76", ">76", ">91") == "red"
|
||
|
||
def test_green_priority_over_yellow(self):
|
||
"""green 是主条件、yellow 是更宽兜底带（id=1 G='>=800' Y='>=700'）：
|
||
值 900 同时满足两条，用户意图是绿，不能被黄截胡"""
|
||
assert calc_level_from_exprs(900, ">=800", ">=700", "<600") == "green"
|
||
assert calc_level_from_exprs(750, ">=800", ">=700", "<600") == "yellow"
|
||
assert calc_level_from_exprs(500, ">=800", ">=700", "<600") == "red"
|
||
|
||
def test_lower_better(self):
|
||
assert calc_level_from_exprs(20, "<=25", "<=30", ">30") == "green"
|
||
assert calc_level_from_exprs(28, "<=25", "<=30", ">30") == "yellow"
|
||
assert calc_level_from_exprs(35, "<=25", "<=30", ">30") == "red"
|
||
|
||
def test_negative_cashflow(self):
|
||
"""缺陷3 场景：F_EVA G='>=0' Y='>=-100000' R='<-100000'"""
|
||
assert calc_level_from_exprs(-200000, ">=0", ">=-100000", "<-100000") == "red"
|
||
assert calc_level_from_exprs(-50000, ">=0", ">=-100000", "<-100000") == "yellow"
|
||
assert calc_level_from_exprs(100, ">=0", ">=-100000", "<-100000") == "green"
|
||
|
||
def test_moderate_band(self):
|
||
"""适中型：绿=内区间、黄=环带、红=外区间"""
|
||
g, y, r = "72~88", "<72|>88", "<=64|>=96"
|
||
assert calc_level_from_exprs(80, g, y, r) == "green"
|
||
assert calc_level_from_exprs(70, g, y, r) == "yellow"
|
||
assert calc_level_from_exprs(98, g, y, r) == "red"
|
||
|
||
def test_priority_constant(self):
|
||
assert LEVEL_PRIORITY == ("red", "green", "yellow")
|
||
|
||
def test_no_threshold_is_info(self):
|
||
assert calc_level_from_exprs(100, None, None, None) == "info"
|
||
assert calc_level_from_exprs(None, ">=1", None, None) == "info"
|
||
|
||
def test_calc_level_with_kpi_object(self):
|
||
class K:
|
||
threshold_green = "<=76"
|
||
threshold_yellow = ">76"
|
||
threshold_red = ">91"
|
||
assert calc_level(95, K()) == "red"
|
||
assert calc_level(70, K()) == "green"
|
||
|
||
|
||
class TestSuggestFormat:
|
||
"""缺陷1：suggest 生成值必须全部可解析（点“自动建议”填入后不再静默失效）"""
|
||
|
||
@pytest.mark.parametrize("kpi_type,target", [
|
||
("higher_better", 1200.0), ("lower_better", 80.0),
|
||
("middle_best", 80.0),
|
||
])
|
||
def test_all_labels_parseable(self, kpi_type, target):
|
||
s = build_suggestion(kpi_type, target, tolerance=10.0)
|
||
for level in ("green", "yellow", "red"):
|
||
label = s[level]["label"]
|
||
assert label, f"{kpi_type}.{level} 缺 label"
|
||
assert is_valid_threshold(label), f"{kpi_type}.{level} 不可解析: {label}"
|
||
|
||
def test_higher_better_labels(self):
|
||
s = build_suggestion("higher_better", 1000.0)
|
||
assert s["green"]["label"] == ">=800"
|
||
assert s["yellow"]["label"] == ">=500"
|
||
assert s["red"]["label"] == "<500"
|
||
# 三档构成完整划分
|
||
assert calc_level_from_exprs(900, ">=800", ">=500", "<500") == "green"
|
||
assert calc_level_from_exprs(600, ">=800", ">=500", "<500") == "yellow"
|
||
assert calc_level_from_exprs(400, ">=800", ">=500", "<500") == "red"
|
||
|
||
def test_lower_better_labels(self):
|
||
s = build_suggestion("lower_better", 100.0)
|
||
assert s["green"]["label"] == "<=120"
|
||
assert s["yellow"]["label"] == "<=200"
|
||
assert s["red"]["label"] == ">200"
|
||
assert calc_level_from_exprs(100, "<=120", "<=200", ">200") == "green"
|
||
assert calc_level_from_exprs(150, "<=120", "<=200", ">200") == "yellow"
|
||
assert calc_level_from_exprs(250, "<=120", "<=200", ">200") == "red"
|
||
|
||
def test_middle_best_labels_partition(self):
|
||
s = build_suggestion("middle_best", 100.0, tolerance=10.0)
|
||
assert s["green"]["label"] == "90~110"
|
||
# 黄 = 外环带 T±2tol（green ⊂ yellow）；因判定顺序 red→green→yellow，绿先命中，
|
||
# 等价于严格分区：绿=90~110 / 黄=80~120 / 红=之外
|
||
assert s["yellow"]["label"] == "80~120"
|
||
assert s["red"]["label"] == "<80|>120"
|
||
g = s["green"]["label"]
|
||
y = s["yellow"]["label"]
|
||
r = s["red"]["label"]
|
||
assert calc_level_from_exprs(100, g, y, r) == "green"
|
||
assert calc_level_from_exprs(95, g, y, r) == "green" # 90~110 内 = 绿
|
||
assert calc_level_from_exprs(85, g, y, r) == "yellow" # 环带
|
||
assert calc_level_from_exprs(119, g, y, r) == "yellow" # 环带
|
||
# 边界归属与 higher_better/lower_better 保持一致：恰好落在 ±2tol 边界 → 黄（红为开区间）
|
||
assert calc_level_from_exprs(80, g, y, r) == "yellow"
|
||
assert calc_level_from_exprs(120, g, y, r) == "yellow"
|
||
assert calc_level_from_exprs(79, g, y, r) == "red"
|
||
assert calc_level_from_exprs(121, g, y, r) == "red"
|
||
|
||
def test_labels_avoid_scientific_notation(self):
|
||
"""大数目标（>=1e6）不得生成 1.6e+06 这类科学计数法（解析器不认）。"""
|
||
s = build_suggestion("higher_better", 2_000_000.0)
|
||
assert s["green"]["label"] == ">=1600000"
|
||
assert s["yellow"]["label"] == ">=1000000"
|
||
assert s["red"]["label"] == "<1000000"
|
||
for level in ("green", "yellow", "red"):
|
||
assert is_valid_threshold(s[level]["label"]), s[level]["label"]
|
||
s2 = build_suggestion("middle_best", 1_500_000.0, tolerance=100_000.0)
|
||
for level in ("green", "yellow", "red"):
|
||
assert is_valid_threshold(s2[level]["label"]), s2[level]["label"]
|
||
|
||
def test_kpi_type_default_preserves_legacy_middle_best(self):
|
||
"""_kpi_type 默认分支必须保持历史语义 = middle_best（原实现的 else 分支）。"""
|
||
from app.api.thresholds import _kpi_type
|
||
|
||
class K:
|
||
def __init__(self, code):
|
||
self.kpi_code = code
|
||
|
||
assert _kpi_type(K("SALES_TOTAL")) == "higher_better"
|
||
assert _kpi_type(K("COST_CONTROL_RATE")) == "lower_better"
|
||
assert _kpi_type(K("C_REBATE_RATE")) == "middle_best" # 渠补率：历史为适中型
|
||
assert _kpi_type(K("BH_SALES_10")) == "middle_best" # 销售-任富海：历史为适中型
|
||
assert _kpi_type(K("MADE_UP_CODE")) == "middle_best" # 未知编码 → 默认适中型
|
||
|
||
def test_c_rebate_rate_suggestion_is_middle_best_range(self):
|
||
"""回归：C_REBATE_RATE 建议仍走适中型区间口径（不得退化为更高的单边界）。"""
|
||
from app.api.thresholds import _kpi_type, build_suggestion
|
||
|
||
class K:
|
||
kpi_code = "C_REBATE_RATE"
|
||
|
||
t = _kpi_type(K())
|
||
s = build_suggestion(t, 80.0, tolerance=27.67)
|
||
assert t == "middle_best"
|
||
assert s["green"]["label"] == "52.33~107.67"
|
||
assert is_valid_threshold(s["yellow"]["label"])
|
||
assert is_valid_threshold(s["red"]["label"])
|
||
|
||
def test_middle_best_zero_tolerance_guard(self):
|
||
"""容忍度退化时不得生成 A~A 之类不可用表达式"""
|
||
s = build_suggestion("middle_best", 0.0, tolerance=0.0)
|
||
for level in ("green", "yellow", "red"):
|
||
assert is_valid_threshold(s[level]["label"])
|
||
|
||
|
||
class TestThresholdValidation:
|
||
"""缺陷2：非法格式必须被拒（接口层 422），不再静默入库"""
|
||
|
||
def test_validate_helper(self):
|
||
assert validate_threshold_format({"threshold_green": ">=800"}) == []
|
||
assert validate_threshold_format({"threshold_yellow": ""}) == []
|
||
assert validate_threshold_format({"threshold_red": None}) == []
|
||
errs = validate_threshold_format({"threshold_yellow": "偏离55%"})
|
||
assert len(errs) == 1 and "阈值格式非法" in errs[0]
|
||
|
||
def test_update_kpi_rejects_bad_threshold(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
resp = client.put(f"/api/cma/kpis/{kpi.id}", headers=auth_header(token),
|
||
json={"threshold_yellow": "偏离55%"})
|
||
assert resp.status_code == 422
|
||
assert "阈值格式非法" in str(resp.json())
|
||
|
||
def test_update_kpi_accepts_valid_thresholds(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
for good in (">=800", ">=-100000", "52.33~107.67", "<=36|>=124", ""):
|
||
resp = client.put(f"/api/cma/kpis/{kpi.id}", headers=auth_header(token),
|
||
json={"threshold_green": good})
|
||
assert resp.status_code == 200, f"{good} 应被接受"
|
||
|
||
|
||
class TestSuggestApi:
|
||
"""接口级：suggest 返回的 label 可解析 + 挂 format_hint"""
|
||
|
||
def test_suggest_by_type_parseable(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="F_REVENUE", target_value=1000)
|
||
resp = client.get(f"/api/cma/thresholds/suggest/{kpi.id}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
labels = [body["suggestion"][lv]["label"] for lv in ("green", "yellow", "red")]
|
||
for lb in labels:
|
||
assert is_valid_threshold(lb), f"建议值不可解析: {lb}"
|
||
assert body.get("format_hint")
|