""" 统一阈值解析器 / 建议格式 / 红灯优先判定 测试 （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_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")