fix(cma): isolate KPI alignment tree by entity (project-20260918-alignment-entity-leak)

GET /api/cma/alignment/tree queried all active KPIs without entity_id filter, so switching the three alignment modes showed other tenants' KPIs. Add get_entity_id dependency and filter KPIDefinition by entity_id. Three mode builders (_build_vertical_tree/_build_horizontal_tree/_build_bsc_chain) unchanged. Auth semantics unchanged; cross-entity explicit entity_id still returns 403. Test: tests/test_alignment_entity_isolation_20260918.py (3 modes x both directions, 403, 401).
2026-09-18 17:18:52 +08:00
parent d33b5e1ced
commit a2fedd892a
2 changed files with 201 additions and 3 deletions
@@ -0,0 +1,188 @@
+"""战略执行看板（/alignment）账套隔离回归测试 — 2026-09-18
+
+任务：project-20260918-alignment-entity-leak（任总 2026-09-18 派单）
+缺陷：GET /api/cma/alignment/tree 未接账套（entity_id）→ 返回全实体 active KPI，
+      三种对齐模式共用同一份 KPI 列表 → 酣客账套视图混入博海 KPI。
+
+本用例覆盖（三模式 × 双向隔离）：
+1. 纵向分解 / 横向支撑 / BSC瀑布链 三种模式，均只返回会话账套的 KPI；
+2. 双向：entity1 会话看不到 entity2 的 KPI，entity2 会话看不到 entity1 的 KPI；
+3. 跨实体显式传参 ?entity_id=<对方> → 403（对齐 get_entity_id 既有语义）；
+4. 鉴权语义未改：无 token → 401；
+5. total 计数 = 本账套 active KPI 数（不是全库）。
+"""
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import Entity, KPIDefinition, RolePermission
+from tests.conftest import (
+    TEST_USER_PASSWORD,
+    auth_header,
+    create_test_user,
+    get_token_for_user,
+)
+
+ALIGN = "/api/cma/alignment"
+MODES = ["vertical_decomposition", "horizontal_support", "bsc_chain"]
+
+# 两个账套各自的 KPI 编码（kpi_code 库内唯一，故两侧编码前缀不同）
+E1_CODES = ["HK_REVENUE_001", "HK_REVENUE_002", "HK_PROFIT_001"]
+E2_CODES = ["BH_REVENUE_001", "BH_PROFIT_001"]
+
+
+def _ensure_entity(db: Session, eid: int, name: str) -> Entity:
+    ent = db.query(Entity).filter(Entity.id == eid).first()
+    if not ent:
+        ent = Entity(id=eid, name=name, short_name=name, status="active")
+        db.add(ent)
+        db.commit()
+        db.refresh(ent)
+    return ent
+
+
+def _set_mode(db: Session, mode: str) -> None:
+    """写入全局对齐模式配置（RolePermission.alignment_config，不分账套）"""
+    perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first()
+    if perm:
+        perm.value = {"mode": mode}
+    else:
+        db.add(RolePermission(key="alignment_config", value={"mode": mode}))
+    db.commit()
+
+
+def _mk_kpi(db: Session, code: str, eid: int, dimension: str, category: str) -> KPIDefinition:
+    kpi = KPIDefinition(
+        entity_id=eid,
+        kpi_code=code,
+        kpi_name=f"{code}-名称",
+        dimension=dimension,
+        category=category,
+        formula="-",
+        data_source_type="excel",
+        data_source="Excel导入",
+        data_owner="财务部",
+        frequency="monthly",
+        unit="元",
+        target_value=0,
+        kpi_level="operational",
+        status="active",
+    )
+    db.add(kpi)
+    db.commit()
+    db.refresh(kpi)
+    return kpi
+
+
+def _seed(db: Session) -> None:
+    """两个账套各 2-3 个 active KPI（三模式均可构建出层级）"""
+    _ensure_entity(db, 1, "酣客")
+    _ensure_entity(db, 2, "博海")
+    for c in E1_CODES:
+        _mk_kpi(db, c, 1, "finance", "revenue_growth")
+    for c in E2_CODES:
+        _mk_kpi(db, c, 2, "finance", "revenue_growth")
+
+
+def _login(client: TestClient, entity_id: int) -> str:
+    resp = client.post("/api/cma/auth/login", json={
+        "username": "testadmin",
+        "password": TEST_USER_PASSWORD,
+        "entity_id": entity_id,
+    })
+    assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
+    return resp.json()["token"]
+
+
+def _collect_codes(nodes) -> set:
+    """递归收集树内所有 kpi_code（维度组/类别组节点无 kpi_code，跳过）"""
+    found = set()
+    for node in nodes or []:
+        if not isinstance(node, dict):
+            continue
+        if node.get("kpi_code"):
+            found.add(node["kpi_code"])
+        found |= _collect_codes(node.get("children"))
+    return found
+
+
+@pytest.mark.parametrize("mode", MODES)
+def test_tree_isolated_by_entity_both_directions(client: TestClient, db: Session, mode: str):
+    """三模式 × 双向：只返回本账套 KPI，不含对方账套 code"""
+    _seed(db)
+    create_test_user(db)
+    _set_mode(db, mode)
+
+    token1 = _login(client, 1)
+    token2 = _login(client, 2)
+
+    r1 = client.get(f"{ALIGN}/tree", headers=auth_header(token1))
+    assert r1.status_code == 200, r1.text
+    body1 = r1.json()
+    codes1 = _collect_codes(body1["tree"])
+
+    r2 = client.get(f"{ALIGN}/tree", headers=auth_header(token2))
+    assert r2.status_code == 200, r2.text
+    body2 = r2.json()
+    codes2 = _collect_codes(body2["tree"])
+
+    # 酣客（e1）：只见酣客 code，绝不含博海 code
+    assert codes1 == set(E1_CODES), f"[{mode}] entity1 返回 {sorted(codes1)}"
+    assert not any(c.startswith("BH_") for c in codes1), f"[{mode}] entity1 混入博海 KPI"
+    # 博海（e2）：反向同理
+    assert codes2 == set(E2_CODES), f"[{mode}] entity2 返回 {sorted(codes2)}"
+    assert not any(c.startswith("HK_") for c in codes2), f"[{mode}] entity2 混入酣客 KPI"
+
+    # total = 本账套 active KPI 数（修复前 = 全库 5）
+    assert body1["total"] == len(E1_CODES), f"[{mode}] entity1 total={body1['total']}"
+    assert body2["total"] == len(E2_CODES), f"[{mode}] entity2 total={body2['total']}"
+    assert body1["mode"] == mode and body2["mode"] == mode
+
+
+def test_tree_cross_entity_explicit_param_403(client: TestClient, db: Session):
+    """跨实体显式传参 → 403（会话 token 绑定账套为准，未改鉴权语义外的任何行为）"""
+    _seed(db)
+    create_test_user(db)
+    _set_mode(db, "horizontal_support")
+
+    token1 = _login(client, 1)
+    token2 = _login(client, 2)
+
+    r = client.get(f"{ALIGN}/tree?entity_id=2", headers=auth_header(token1))
+    assert r.status_code == 403, f"entity1 会话读 entity2 应 403，实际 {r.status_code}"
+    assert "无权访问" in r.json()["detail"]
+
+    r = client.get(f"{ALIGN}/tree?entity_id=1", headers=auth_header(token2))
+    assert r.status_code == 403, f"entity2 会话读 entity1 应 403，实际 {r.status_code}"
+
+
+def test_tree_without_token_401(client: TestClient, db: Session):
+    """鉴权底线不变：无 Bearer token → 401"""
+    _seed(db)
+    _set_mode(db, "bsc_chain")
+    r = client.get(f"{ALIGN}/tree")
+    assert r.status_code == 401, f"无 token 应 401，实际 {r.status_code}"
+
+
+def test_tree_same_entity_explicit_param_ok(client: TestClient, db: Session):
+    """同账套显式传参与 token 一致 → 200（不是一律拒绝）"""
+    _seed(db)
+    create_test_user(db)
+    _set_mode(db, "bsc_chain")
+    token1 = get_token_for_user(client)  # entity1
+    r = client.get(f"{ALIGN}/tree?entity_id=1", headers=auth_header(token1))
+    assert r.status_code == 200, r.text
+    assert _collect_codes(r.json()["tree"]) == set(E1_CODES)
+
+
+def test_tree_kpi_id_of_other_entity_not_returned(client: TestClient, db: Session):
+    """kpi_id 指向对方账套的 KPI → 不返回该 KPI（隔离后树内查无此 id）"""
+    _seed(db)
+    create_test_user(db)
+    _set_mode(db, "vertical_decomposition")
+    other = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == E2_CODES[0]).first()
+    token1 = _login(client, 1)
+    r = client.get(f"{ALIGN}/tree?kpi_id={other.id}", headers=auth_header(token1))
+    assert r.status_code == 200
+    body = r.json()
+    assert body is None or not _collect_codes([body]), f"entity1 不应取到 entity2 的 KPI: {body}"