# -*- coding: utf-8 -*- """统一阈值解析器 + 红黄绿判定顺序（红灯优先）单测 任务：project-20260911-threshold-fix 覆盖：解析器（正/负/小数/区间/多条件/非法）、判定顺序、双调用方接线、存量表达式回归。 """ import os import sys import pytest BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if BACKEND not in sys.path: sys.path.insert(0, BACKEND) from app.utils.threshold_parser import ( # noqa: E402 LEVEL_PRIORITY, calc_level, calc_level_from_exprs, describe_level, eval_threshold, is_valid_threshold, parse_clauses, parse_threshold, ) class _Kpi: """轻量替身：只需要 3 个阈值属性""" def __init__(self, g=None, y=None, r=None): self.threshold_green = g self.threshold_yellow = y self.threshold_red = r # ───────────────────────── ① 解析器 ───────────────────────── @pytest.mark.parametrize("expr,expected", [ (">=800", (">=", 800.0)), ("<=25", ("<=", 25.0)), (">60", (">", 60.0)), ("<20", ("<", 20.0)), ("=100", ("=", 100.0)), ("!=0", ("!=", 0.0)), (">=-100000", (">=", -100000.0)), # 缺陷3：负数 ("<-100000", ("<", -100000.0)), ("<-1.5", ("<", -1.5)), (">=1.0", (">=", 1.0)), (">= 800", (">=", 800.0)), # 允许空白 (" >=800 ", (">=", 800.0)), ("52.33~107.67", ("~", (52.33, 107.67))), # 区间 ("-182678~183078", ("~", (-182678.0, 183078.0))), ]) def test_parse_threshold_ok(expr, expected): assert parse_threshold(expr) == expected @pytest.mark.parametrize("expr", ["偏离55%", "偏离>55%", "80%", ">=abc", "~", ">=", "52.33~", ">=1>=2", "abc", ">=800元", ">=- 100"]) def test_parse_threshold_invalid(expr): assert parse_threshold(expr) == (None, None) assert is_valid_threshold(expr) is False @pytest.mark.parametrize("expr", [None, "", " "]) def test_blank_is_valid_but_no_clause(expr): """空串 = 不设置阈值：视为合法，但无子句。""" assert is_valid_threshold(expr) is True assert parse_clauses(expr) == [] assert parse_threshold(expr) == (None, None) def test_multi_clause(): assert is_valid_threshold("<=36|>=124") is True assert is_valid_threshold("<=36 | >=124") is True assert is_valid_threshold("<=36｜>=124") is True # 全角竖线 assert is_valid_threshold("<=36|abc") is False assert parse_clauses("<=36|>=124") == [("<=", 36.0), (">=", 124.0)] @pytest.mark.parametrize("actual,expr,expected", [ (800, ">=800", True), (799.99, ">=800", False), (25, "<=25", True), (25.01, "<=25", False), (61, ">60", True), (60, ">60", False), (19, "<20", True), (20, "<20", False), (-100000, ">=-100000", True), (-100001, ">=-100000", False), (-100001, "<-100000", True), (80, "52.33~107.67", True), (52.33, "52.33~107.67", True), (107.67, "52.33~107.67", True), (107.68, "52.33~107.67", False), (200, "<24.66|>135.34", True), (10, "<24.66|>135.34", True), (80, "<24.66|>135.34", False), (None, ">=0", False), (5, "偏离55%", False), # 非法表达式 → 不命中（不静默乱判） (5, "", False), ]) def test_eval_threshold(actual, expr, expected): assert eval_threshold(actual, expr) is expected # ─────────────────── ② 判定顺序：红灯优先 ─────────────────── def test_level_priority_is_red_first(): assert LEVEL_PRIORITY == ("red", "green", "yellow") def test_defect5_yellow_no_longer_masks_red(): """缺陷5 实测：id=6 G='<=76' Y='>76' R='>91'，值 95 必须判红（原来判黄）。""" kpi = _Kpi(g="<=76", y=">76", r=">91") assert calc_level(95, kpi) == "red" # 关键用例 assert calc_level(92, kpi) == "red" assert calc_level(91, kpi) == "yellow" # 未过 >91 → 黄 assert calc_level(80, kpi) == "yellow" assert calc_level(76, kpi) == "green" # <=76 → 绿 assert calc_level(70, kpi) == "green" def test_green_beats_wide_yellow(): """id=1 实测：G='>=800' Y='>=700'，值 900 不能判黄。""" kpi = _Kpi(g=">=800", y=">=700", r="<600") assert calc_level(900, kpi) == "green" assert calc_level(750, kpi) == "yellow" assert calc_level(500, kpi) == "red" assert calc_level(650, kpi) == "info" # 600~700 无命中 def test_negative_thresholds(): """id=32：Y='>=-100000' R='<-100000' 存量负数表达式（缺陷3）。""" kpi = _Kpi(g=">=0", y=">=-100000", r="<-100000") assert calc_level(100, kpi) == "green" assert calc_level(-50000, kpi) == "yellow" assert calc_level(-200000, kpi) == "red" def test_moderate_range_semantics(): """区间型（middle_best）：绿内区间 → 黄外环 → 红区间外。""" kpi = _Kpi(g="52.33~107.67", y="24.66~135.34", r="<24.66|>135.34") assert calc_level(80, kpi) == "green" assert calc_level(52.8, kpi) == "green" assert calc_level(124.1, kpi) == "yellow" assert calc_level(30, kpi) == "yellow" assert calc_level(200, kpi) == "red" assert calc_level(10, kpi) == "red" def test_no_threshold_and_none_actual(): assert calc_level(100, _Kpi()) == "info" assert calc_level(None, _Kpi(g=">=0", y=">=0", r=">=0")) == "info" assert calc_level_from_exprs(50, green=None, yellow=None, red=None) == "info" def test_describe_level(): kpi = _Kpi(g="<=76", y=">76", r=">91") assert describe_level(95, kpi) == ("red", ">91") assert describe_level(80, kpi) == ("yellow", ">76") assert describe_level(70, kpi) == ("green", "<=76") assert describe_level(0, kpi) == ("green", "<=76") assert describe_level(None, kpi) == ("info", None) # ─────────────────── ③ 双调用方接线（消除双实现） ─────────────────── def test_alert_generator_uses_unified_parser(): """alert_generator 必须委托统一解析器（负数/区间/多条件都生效）。""" from scripts import alert_generator as ag assert ag._parse_threshold(">=-100000") == (">=", -100000.0) assert ag._parse_threshold("52.33~107.67") == ("~", (52.33, 107.67)) trig, msg = ag._check_threshold(95, ">91", "red") assert trig is True and ">91" in msg assert ag._check_threshold(95, ">91x", "red") == (False, "") assert ag._check_threshold(None, ">91", "red") == (False, "") # 区间 + 多条件（旧实现做不到） assert ag._check_threshold(124.1, "24.66~135.34", "yellow")[0] is True assert ag._check_threshold(10, "<24.66|>135.34", "red")[0] is True def test_alert_generator_loop_order_is_red_first(): """判定循环顺序必须与 LEVEL_PRIORITY 一致（红灯优先）。""" import inspect from scripts import alert_generator as ag src = inspect.getsource(ag.run_alert_check) assert "for level in LEVEL_PRIORITY" in src assert 'for level in ["green", "yellow", "red"]' not in src def test_kpis_static_threshold_level_uses_unified_parser(): from app.api.kpis import _static_threshold_level kpi = _Kpi(g=">=0", y=">=-100000", r="<-100000") assert _static_threshold_level(-200000, kpi) == "red" assert _static_threshold_level(-50000, kpi) == "yellow" assert _static_threshold_level(100, kpi) == "green" def test_kpis_validate_threshold_format(): from app.api.kpis import validate_threshold_format assert validate_threshold_format({}) == [] assert validate_threshold_format({"threshold_green": ">=800", "threshold_yellow": "", "threshold_red": None}) == [] errs = validate_threshold_format({"threshold_yellow": "偏离55%"}) assert len(errs) == 1 and "threshold_yellow" in errs[0] errs2 = validate_threshold_format({"threshold_green": "80%", "threshold_red": "偏离>55%"}) assert len(errs2) == 2 def test_suggest_labels_are_parseable(): """suggest 生成的 label 必须可解析（缺陷1 根因）。""" from app.api import thresholds as th def _walk(obj, acc): if isinstance(obj, dict): if "label" in obj and obj.get("label") is not None: acc.append(obj["label"]) for v in obj.values(): _walk(v, acc) elif isinstance(obj, list): for v in obj: _walk(v, acc) class _K: def __init__(self, code, target): self.id = 1 self.kpi_code = code self.kpi_name = code self.target_value = target for code in ("SALES_TOTAL", "COST_CONTROL_RATE", "TOP5_CUSTOMER_RATIO", "MADE_UP_CODE"): out = th.suggest_by_type(_K(code, 80.0)) labels = [] _walk(out, labels) assert labels, code for lb in labels: assert is_valid_threshold(lb), f"{code} 生成了不可解析 label: {lb!r}" def test_existing_266_expressions_still_parse(): """回归：库里现存的表达式语法必须全部可解析（除已修正的存量 4 条外为 0 不可解析）。""" from app.database import get_session_local from app.models import KPIDefinition try: db = get_session_local()() except Exception as e: # 无DB环境（CI）时跳过 pytest.skip(f"DB 不可用: {e}") try: rows = db.query(KPIDefinition).filter( (KPIDefinition.threshold_green.isnot(None)) | (KPIDefinition.threshold_yellow.isnot(None)) | (KPIDefinition.threshold_red.isnot(None)) ).all() bad = [] for k in rows: for f in ("threshold_green", "threshold_yellow", "threshold_red"): e = getattr(k, f) if e and str(e).strip() and not is_valid_threshold(e): bad.append((k.id, k.kpi_code, f, str(e))) assert bad == [], f"仍存在不可解析阈值: {bad}" finally: db.close()