feat(strategy-map): live board reading map_objectives + kpi_targets (read-only)

Task: finance-20261011-strategy-map-frontend-fullstack (P2, read-only integration)

- backend/app/utils/strategy_board.py: single source of truth for the level
  caliber (layer colors, inverse-kpi explicit whitelist, status class by
  unit, zero guard, period ranking, latest-actual picking). Pure functions.
- backend/app/api/maps.py: GET /api/cma/maps/{map_id}/strategy-board
  - nodes straight from map_objectives (sorted by sort_order, id)
  - node<->kpi binding with 4-step fallback, binding_source echoed back
  - targets from kpi_targets (target_type compatible: yearly + annual),
    deduped per (kpi_id, period) with yearly preferred
  - actuals from kpi_values actual-only scope, latest period
  - causality edges from kpi_causality, out-of-map endpoints dropped+counted
  - entity_id taken from the token only (no cross-tenant reads)
  - zero writes; nothing is persisted by this endpoint
- frontend/src/views/StrategyMap.vue: /strategy-map live board (4 swim lanes
  bottom-up, node cards, 20 causality arrows in SVG, period switch, A3 print).
  No emoji, null guards everywhere (no toFixed on null).
- api/index.ts + router/index.ts + permission.ts: route + menu registration
  (/strategy-map, zone A read-only, roles ceo/finance/business/it).
- backend/tests/test_strategy_board_20261011.py: 33 cases (spec section 4 /
  4.1 / 4.2 / 4.3 + section 7 fixtures + threshold boundaries + period rank).
- backend/tests/test_map_objectives_kpis_form_20261011.py: 14 cases for the
  four kpis column shapes (list, json string, object list, NULL/[]/weird).

Not included: push + backend restart (production write, separate gate).

(cherry picked from commit c675bde6b4beaf1d8a9d02a80f5434203a53f348)
This commit is contained in:
Hermes CI Fix
2026-10-18 16:31:52 +08:00
parent 2cff58ca6a
commit 75a48f55d2
8 changed files with 1263 additions and 4 deletions
@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
"""`map_objectives.kpis` 读端形态兼容回归(承接 dama 知会件 2026-10-11 19:1x)
背景:map_id=65 的 19 行 `kpis` 现为 NULL;库内唯一非空先例是**编码字符串数组**
(map 53);finance 拟回填的是**对象数组**且未通过形态核验。
因此读端(`_sb_resolve_bindings`)必须对四种形态都不抛错、且两形都能正确绑定:
| 形态 | 样例 | 期望来源 |
|---|---|---|
| 甲 编码字符串数组 | `["F_REVENUE"]` | kpis_code(按 code 查 kpi_definitions) |
| 丙 同上但为 JSON 字符串 | `'["F_REVENUE"]'` | kpis_code |
| 乙 对象数组 | `[{"kpi_id":1,"kpi_code":"F_REVENUE"}]` | kpis(直取 kpi_id) |
| 乙' 对象数组仅 code | `[{"kpi_code":"F_REVENUE"}]` | kpis_code |
| 丁 NULL / [] / 异形 | `None` / `[]` / `"not-json"` | 兜底(name / note / none),不得抛错 |
同时断言:**不按 status='active' 过滤** kpi_definitions
(15 条新建定义 status=NULL,#11 inactive、#230 archived,按 active 过滤会漏)。
纯桩测试(FakeDB),不连库、不触生产。
"""
import json
from types import SimpleNamespace
from typing import Any
import pytest
from app.api.maps import _sb_resolve_bindings
from app.models import KPIDefinition
DEFS = [
SimpleNamespace(id=1, entity_id=1, kpi_code="F_REVENUE", kpi_name="营业收入"),
SimpleNamespace(id=35, entity_id=1, kpi_code="F_NET_PROFIT", kpi_name="净利润"),
# status 刻意置 None:读端不得按 status 过滤
SimpleNamespace(id=504, entity_id=1, kpi_code="F_CCC", kpi_name="现金转换周期", status=None),
]
TARGETS = [(35, "净利润 · 2027")]
def _db() -> Any:
"""桩:仅供单测,不连库(运行时鸭子类型,Session 的极小子集)"""
return _DB()
def _col_val(arg):
try:
return arg.left.key, arg.right.value
except Exception:
return None, None
class _Q:
def __init__(self, model, store):
self.model, self.store, self.eq = model, store, {}
def filter(self, *args):
for a in args:
k, v = _col_val(a)
if k:
self.eq[k] = v
return self
def first(self):
for r in self.store.get(self.model, []):
if all(getattr(r, k, None) == v for k, v in self.eq.items()):
return r
return None
class _DB:
def __init__(self):
self.store = {KPIDefinition: DEFS}
def query(self, model):
return _Q(model, self.store)
def execute(self, stmt, params):
pat = params.get("pat") or ""
class _R:
def first(self):
return next((t for t in TARGETS if pat and t[1].startswith(pat[:-2])), None)
return _R()
def _obj(oid, name, kpis):
return SimpleNamespace(id=oid, name=name, kpis=kpis, dimension_key="finance")
@pytest.mark.parametrize("kpis,expect_id,expect_src", [
(["F_REVENUE"], 1, "kpis_code"), # 甲
(json.dumps(["F_CCC"]), 504, "kpis_code"), # 丙
([{"kpi_id": 35, "kpi_code": "F_NET_PROFIT"}], 35, "kpis"), # 乙
([{"kpi_code": "F_NET_PROFIT"}], 35, "kpis_code"), # 乙'
([{"code": "F_REVENUE"}], 1, "kpis_code"), # 乙''(code 别名)
({"kpi_id": 35}, 35, "kpis"), # 单对象(非数组)
])
def test_kpis_forms_bind_correctly(kpis, expect_id, expect_src):
o = _obj(2001, "任意节点名", kpis)
b, s = _sb_resolve_bindings(_db(), 1, [o])
assert b.get(2001, (None,))[0] == expect_id, f"form={kpis!r} got={b.get(2001)}"
assert s.get(2001) == expect_src, f"form={kpis!r} source={s.get(2001)}"
@pytest.mark.parametrize("kpis", [None, [], "not-json", 12345])
def test_kpis_degenerate_forms_fall_back_and_never_raise(kpis):
"""丁:NULL / 空 / 异形 —— 走 name/note 兜底或 none,绝不抛异常"""
o = _obj(2002, "净利润", kpis) # 名称可回查 + note 兜底均命中 id=35
b, s = _sb_resolve_bindings(_db(), 1, [o])
assert b.get(2002, (None,))[0] == 35
assert s.get(2002) in ("name", "note")
def test_null_kpis_unmatched_node_reports_none_not_crash():
"""19 行全 NULL 时:节点仍须在结果里(source=none),不得丢节点/抛错"""
o = _obj(2003, "图上没有对应 KPI 的节点", None)
b, s = _sb_resolve_bindings(_db(), 1, [o])
assert o.id not in b and s.get(o.id) == "none"
def test_no_status_active_filter_on_kpi_definitions():
"""status=None 的定义(15 条新建 KPI 形态)必须能被读到 —— 读端不得按 status 过滤"""
o = _obj(2004, "现金转换周期", [{"kpi_code": "F_CCC"}])
b, _s = _sb_resolve_bindings(_db(), 1, [o])
assert b.get(2004, (None,))[0] == 504, "status=None 的定义被过滤掉了(应命中)"
@@ -0,0 +1,174 @@
# -*- coding: utf-8 -*-
"""战略地图活看板色阶口径回归(任务 finance-20261011-strategy-map-frontend-fullstack)
覆盖件 §四/§4.1/§4.2/§4.3 与 §七「可直接复用的对照值」(2026-01~08 现取)。
纯函数测试:不连库、不依赖 FastAPI。
"""
import pytest
from app.utils.strategy_board import (
INVERSE_KPI_IDS, LAYER_META, LEVEL_COLORS, STATUS_UNIT,
compute_level, is_inverse_kpi, is_status_kpi, layer_meta, period_rank, pick_latest,
)
# ── §七 对照值(前端自测用,逐条断言期望色)──────────────────────────
@pytest.mark.parametrize("name,kpi_id,unit,current,target,expect", [
("营业收入", 1, "万元", 808.80, 1050, "danger"), # r=0.77
("CM 率", 503, "%", 16.83, 43, "danger"),
("CCC(逆)", 504, "天", 206.4, 150, "danger"), # r=0.727 取倒数
("货币资金", 505, "万元", 5.31, 60, "danger"),
("高贡献线收入占比", 506, "%", 47.81, 55, "warning"), # r=0.87
])
def test_spec_section7_fixtures(name, kpi_id, unit, current, target, expect):
got = compute_level(kpi_id=kpi_id, unit=unit, current=current, target=target)
assert got["level"] == expect, f"{name}: {got}"
def test_inverse_ccc_rate_is_reciprocal():
"""§4.1 CCC 206.4 → 目标 150:r = 150/206.4 = 0.7267(不是 206.4/150=1.376 的绿灯)"""
got = compute_level(kpi_id=504, unit="天", current=206.4, target=150)
assert got["mode"] == "inverse"
assert got["rate"] == pytest.approx(0.7267, abs=1e-4)
assert got["level"] == "danger"
# 若误按正向算会得到 success —— 这正是本回归要拦住的假绿
assert compute_level(kpi_id=9999, unit="天", current=206.4, target=150)["level"] == "success"
def test_inverse_whitelist_is_explicit_not_direction_based():
"""§4.1 逆指标=显式白名单;#230/#23 库内 direction 为 NULL 也不得漏"""
assert INVERSE_KPI_IDS == frozenset({504, 511, 230, 23})
for kid in (504, 511, 230, 23):
assert is_inverse_kpi(kid) is True
for kid in (1, 11, 502, 503, 505, 506, 507, 508, 509, 510, 512, 513, 514, 515, 516):
assert is_inverse_kpi(kid) is False
# 短期借款 #230:unit=元、direction=NULL ⇒ 逆指标(不能因 direction 缺失判成状态类/正向)
got = compute_level(kpi_id=230, unit="元", current=211.5, target=211.5)
assert got["mode"] == "inverse"
assert got["level"] == "success"
# ── §4.2 状态类:按 unit 判,1=绿 / 0=NULL=红,且绝不做除法 ──────────
@pytest.mark.parametrize("kpi_id,current,expect", [
(512, 1, "success"),
(512, 0, "danger"),
(512, None, "danger"), # 件 §4.2 逐字:0 / NULL = 未达成 ⇒ 红
(513, 1, "success"),
(515, 1, "success"),
(516, 0, "danger"),
])
def test_status_class_by_unit(kpi_id, current, expect):
unit = STATUS_UNIT # '状态'
got = compute_level(kpi_id=kpi_id, unit=unit, current=current, target=1)
assert got["mode"] == "status"
assert got["level"] == expect
assert got["rate"] is None, "状态类不得产出达成率"
def test_status_class_never_divides():
"""状态类 + 目标 0/缺失 ⇒ 依然按 1/0 判色,不触发除零"""
for target in (None, 0):
got = compute_level(kpi_id=512, unit="状态", current=0, target=target)
assert got["level"] == "danger" and got["rate"] is None
assert is_status_kpi("状态") is True
assert is_status_kpi("状态数") is False # 精确匹配,不用 startswith
assert is_status_kpi(None) is False
def test_status_not_inferred_from_null_direction():
"""`direction IS NULL` 不等于状态类:#230/#23 同为 NULL 但不走状态分支"""
for kid, unit in ((230, "元"), (23, "%")):
assert compute_level(kpi_id=kid, unit=unit, current=10, target=10)["mode"] == "inverse"
for kid in (512, 513, 515, 516):
assert compute_level(kpi_id=kid, unit=STATUS_UNIT, current=1, target=1)["mode"] == "status"
# ── §4.3 零点保护 / 空值 ────────────────────────────────────────────
@pytest.mark.parametrize("current,target,reason", [
(None, 150, "无实际值"),
(206.4, None, "目标值缺失或为 0"),
(206.4, 0, "目标值缺失或为 0"),
(0, 150, "逆指标现值为 0(倒数无定义)"),
])
def test_zero_guard_returns_info(current, target, reason):
got = compute_level(kpi_id=504, unit="天", current=current, target=target)
assert got["level"] == "info"
assert got["color"] == LEVEL_COLORS["info"]
assert got["rate"] is None
assert got["reason"] == reason
def test_never_nan_or_inf_or_raise():
"""历史教训:`{{ current.cover.toFixed(1) }}` 对 null 无守卫崩卡 —— 后端必须不产出 NaN/Inf"""
bad_inputs = [
(None, None), (None, 0), ("", ""), ("abc", "def"), (float("nan"), 100),
(100, float("nan")), (float("inf"), 100), (100, float("inf")),
("206.4", "150"), (False, 1), ([], {}),
]
for cur, tgt in bad_inputs:
got = compute_level(kpi_id=504, unit="天", current=cur, target=tgt)
assert got["rate"] is None or got["rate"] == got["rate"]
assert got["rate"] not in (float("inf"), float("-inf"))
assert got["level"] in LEVEL_COLORS
# 字符串数字可按数值处理(Excel/JSON 常见)
assert compute_level(kpi_id=1, unit="万元", current="808.80", target="1050")["rate"] == pytest.approx(0.7703, abs=1e-4)
# ── 阈值边界 ────────────────────────────────────────────────────────
@pytest.mark.parametrize("current,target,expect", [
(100, 100, "success"), (120, 100, "success"),
(80, 100, "warning"), (99.9, 100, "warning"),
(79.99, 100, "danger"), (0.01, 100, "danger"),
])
def test_threshold_boundaries(current, target, expect):
assert compute_level(kpi_id=1, unit="万元", current=current, target=target)["level"] == expect
# ── 期间排序 / 取最新(实测区取值口径)─────────────────────────────
@pytest.mark.parametrize("a,b", [
("2026-08", "2026-H1"), ("2026-Q4", "2026-09"), ("2027", "2026-H2"),
("2026-12", "2026-01"), ("2026-Q3", "2026-Q2"),
])
def test_period_rank_order(a, b):
assert period_rank(a) > period_rank(b)
def test_period_rank_unparsable_is_zero():
for bad in (None, "", "未知", "abc", "2026-13"):
assert period_rank(bad) == 0
class _Row:
def __init__(self, period, data_status, actual_value, id):
self.period, self.data_status, self.actual_value, self.id = period, data_status, actual_value, id
def test_pick_latest_prefers_newer_period_then_verified():
rows = [
_Row("2026-01", "verified", 180.87, 1),
_Row("2026-08", "error", 14.45, 9),
_Row("2026-08", "verified", 14.45, 8),
_Row("2026-H1", "verified", 713.27, 10),
]
got = pick_latest(rows)
assert got.period == "2026-08"
assert got.data_status == "verified", "同期间内必须取 verified(不被 error 行顶掉)"
assert pick_latest([]) is None
# ── 层色规范(§三,固定)─────────────────────────────────────────────
def test_layer_colors_fixed():
assert LAYER_META["finance"]["color"] == "#F56C6C"
assert LAYER_META["customer"]["color"] == "#409EFF"
assert LAYER_META["process"]["color"] == "#67C23A"
assert LAYER_META["learning"]["color"] == "#E6A23C"
# 未知/缺失 dimension_key 不抛错、不丢节点(置灰占位)
assert layer_meta(None)["color"] == LEVEL_COLORS["info"]
assert layer_meta("mystery")["color"] == LEVEL_COLORS["info"]
def test_level_colors_fixed():
assert LEVEL_COLORS["success"] == "#16A34A"
assert LEVEL_COLORS["warning"] == "#EAB308"
assert LEVEL_COLORS["danger"] == "#DC2626"
assert LEVEL_COLORS["info"] == "#94A3B8"