fix(cma): scope ai-suggestion action_plan and preview KPI to session entity

_apply_action_plan: constrain the target KPIDefinition lookup to sug.entity_id so a suggestion whose target_id/suggestion_data.kpi_id points at a foreign-entity KPI now returns 404 instead of creating an ActionPlan row with a cross-entity kpi_id. preview_suggestion: same constraint; a foreign-entity KPI now degrades to the existing placeholder (KPI#<id>, current_target=None) instead of echoing the foreign KPI name and current_target. Degradation (not 404) keeps the read-side preview usable when the stored kpi_id is inconsistent. Regression: backend/tests/test_ai_suggestion_kpi_attribution_residual_20260918.py (RED 2 failed before, GREEN 5 passed after; full suite 1228+2F -> 1230 passed / 1 xfailed).
2026-09-18 21:04:30 +08:00
parent 9cb960da25
commit 557771a803
2 changed files with 300 additions and 2 deletions
@@ -0,0 +1,291 @@
+"""CMA 跨实体隔离 — AI 建议同文件残余两处 KPI 归属（2026-09-18 派生单 2030b）
+
+任务：fullstack-20260918-2030b-aisuggestion-kpi-attribution-residual
+父单：fullstack-20260918-2030-dismiss-suggestion-entity-scope
+      （其执行期对本文件 app/api/ai_suggestions.py 的只读复查 §五 范围外发现）
+
+缺陷（同文件 ai_suggestions.py 两处 KPIDefinition 只按 id 直取、不校验归属）：
+  1. `_apply_action_plan` (原 239 行) — 建议自身归属校验通过后，其
+     suggestion_data.kpi_id / target_id 指向对端账套 KPI 时，会建出
+     ActionPlan(entity_id=本账套, kpi_id=对端账套KPI) 的跨账套引用行
+     （计划列表按账套可见、KPI 名解析取到对端账套对象）。
+  2. `preview_suggestion` (原 314 行) — 同一写法；建议自身已被账套收敛（读侧单 2025），
+     但其 suggestion_data.kpi_id 指向对端 KPI 时，预览会回显对端账套的 KPI 名与
+     current_target（只读泄漏）。
+
+修复口径：
+  - `_apply_action_plan`：KPI 查询补 `entity_id == sug.entity_id`，
+    不匹配抛 404「KPI {id} 不存在」（对齐 2030 单 `_apply_kpi_target`/`_apply_budget_adjust`
+    写法），且零落行。
+  - `preview_suggestion`：沿用现有降级语义（kpi=None → 展示 `KPI#<id>`、current_target=None），
+    不抛 404 —— 理由见报告 §三取舍说明（读侧端点不应因数据不自洽而整体不可用）。
+
+对端记录口径 = 404「KPI {id} 不存在」（写侧）/ 降级占位（读侧）。
+
+本用例（改前 RED 原文见 _reports/evidence-fullstack-20260918-2030b-red.txt）：
+- T1 应用 action_plan：目标 KPI 非本账套 → 404 且零落 ActionPlan 行（含对端 kpi_id 归属行），
+     建议状态保持 unapplied；两条取数路径（suggestion_data 回退 / 请求体显式 kpi_id）均 404
+- T2 应用 action_plan 正对照：本账套建议 + 本账套 KPI → 200 且落行 entity_id=会话账套、
+     kpi_id=本账套 KPI
+- T3 预览：目标 KPI 非本账套 → 不回显对端 KPI 名与 current_target，降级为 KPI#<id>
+- T4 预览正对照：本账套 KPI → kpi_name / current_target 完整回显
+- T5 鉴权底线：preview / apply(action_plan) 无 token → 401（改前改后都应通过）
+"""
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models import AISuggestion, ActionPlan, Entity, KPIDefinition
+from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
+
+SUGG = "/api/cma/ai/suggestions"
+
+# kpi_code 全局唯一 → 每账套独立 code（ZZ_ 前缀为本单专用，避免与同族用例撞名）
+E1_KPI = "ZZ_AKR_E1_KPI"
+E2_KPI = "ZZ_AKR_E2_KPI"
+E1_SUG_TITLE = "ZZ_AKR e1 私有建议"
+E2_SUG_TITLE = "ZZ_AKR e2 私有建议"
+CROSS_SUG_TITLE = "ZZ_AKR e2 指向对端 KPI 的建议"
+
+E1_TARGET = 1000.0
+E2_TARGET = 3000.0
+
+
+# ──────────────────────────────────────────────
+# 夹具（沿用 2030 单风格：双向登录 + 同账套正对照）
+# ──────────────────────────────────────────────
+
+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 _mk_kpi(db: Session, code: str, eid: int, target: float = E1_TARGET) -> KPIDefinition:
+    kpi = KPIDefinition(
+        entity_id=eid,
+        kpi_code=code,
+        kpi_name=f"{code}-名称",
+        dimension="finance",
+        category="revenue_growth",
+        formula="-",
+        data_source_type="excel",
+        data_source="Excel导入",
+        data_owner="财务部",
+        frequency="monthly",
+        unit="元",
+        target_value=target,
+        kpi_level="operational",
+        status="active",
+    )
+    db.add(kpi)
+    db.commit()
+    db.refresh(kpi)
+    return kpi
+
+
+def _seed_two_entities(db: Session):
+    _ensure_entity(db, 1, "酣客")
+    _ensure_entity(db, 2, "博海")
+    create_test_user(db)
+
+
+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 _mk_suggestion(client: TestClient, token: str, title: str,
+                   target_id: int, kpi_id: int,
+                   target_value: float = 2000.0,
+                   suggestion_type: str = "kpi_target") -> int:
+    body = {
+        "suggestion_type": suggestion_type,
+        "target_type": "kpi",
+        "target_id": target_id,
+        "title": title,
+        "content": "目标偏低，建议上调",
+        "suggestion_data": {"kpi_id": kpi_id, "target_value": target_value},
+    }
+    r = client.post(SUGG, json=body, headers=auth_header(token))
+    assert r.status_code == 200, f"创建建议失败: {r.status_code} {r.text[:300]}"
+    return r.json()["data"]["id"]
+
+
+def _sug_status(db: Session, sid: int) -> str:
+    db.expire_all()
+    row = db.query(AISuggestion).filter(AISuggestion.id == sid).first()
+    assert row is not None, f"建议 #{sid} 不存在"
+    return row.status
+
+
+def _plan_rows(db: Session, kpi_id: int = None):
+    db.expire_all()
+    q = db.query(ActionPlan)
+    if kpi_id is not None:
+        q = q.filter(ActionPlan.kpi_id == kpi_id)
+    return q.all()
+
+
+# ──────────────────────────────────────────────
+# T1. 应用 action_plan：目标 KPI 归属校验（两条取数路径）
+# ──────────────────────────────────────────────
+
+def test_apply_action_plan_foreign_kpi_404(client: TestClient, db: Session):
+    """建议属本账套但其目标 KPI 属对端 → 404 且零落行；对端 KPI 行数逐值未变"""
+    _seed_two_entities(db)
+    k1 = _mk_kpi(db, E1_KPI, 1, target=E1_TARGET)
+    k2 = _mk_kpi(db, E2_KPI, 2, target=E2_TARGET)
+    tok2 = _login(client, 2)
+
+    # e2 会话创建建议（归属 e2），但 target_id/suggestion_data.kpi_id 指向 e1 的 KPI
+    # （模拟手填数据 / 迁移遗留 / Bot 通道写入的不自洽数据）
+    s_cross = _mk_suggestion(client, tok2, CROSS_SUG_TITLE, k1.id, k1.id)
+
+    rows_before_e1 = len(_plan_rows(db, k1.id))
+    rows_before_all = len(_plan_rows(db))
+    assert rows_before_e1 == 0, f"前置：e1 KPI 已有行动方案 {rows_before_e1} 行"
+
+    # 路径 A：请求体不给 kpi_id → 回退 suggestion_data.kpi_id（对端账套）
+    r_a = client.post(f"{SUGG}/{s_cross}/apply", headers=auth_header(tok2),
+                      json={"action": "action_plan", "title": "ZZ_AKR 跨账套计划A"})
+    assert r_a.status_code == 404, (
+        f"action_plan 目标 KPI 属对端账套应 404，实际 {r_a.status_code} {r_a.text[:200]}"
+        " —— _apply_action_plan 只按 id 查 KPIDefinition，未校验归属会话账套"
+    )
+    # 路径 B：请求体显式给 kpi_id（对端账套）
+    r_b = client.post(f"{SUGG}/{s_cross}/apply", headers=auth_header(tok2),
+                      json={"action": "action_plan", "kpi_id": k1.id,
+                            "title": "ZZ_AKR 跨账套计划B"})
+    assert r_b.status_code == 404, (
+        f"显式 kpi_id 指向对端账套应 404，实际 {r_b.status_code} {r_b.text[:200]}"
+    )
+
+    rows_after_e1 = len(_plan_rows(db, k1.id))
+    rows_after_all = len(_plan_rows(db))
+    assert rows_after_e1 == rows_before_e1, (
+        f"越权 action_plan 落了跨账套引用行（对端 kpi_id={k1.id}）："
+        f"{rows_before_e1} → {rows_after_e1}"
+    )
+    assert rows_after_all == rows_before_all, (
+        f"越权 action_plan 落了行（总数）：{rows_before_all} → {rows_after_all}"
+    )
+    assert _sug_status(db, s_cross) == "unapplied", (
+        "越权应用竟改动了建议状态（应保持 unapplied）"
+    )
+    # 正对照前置：本账套 KPI 未被误伤
+    assert len(_plan_rows(db, k2.id)) == 0
+
+
+# ──────────────────────────────────────────────
+# T2. 应用 action_plan 正对照：本账套正常落行
+# ──────────────────────────────────────────────
+
+def test_apply_action_plan_own_kpi_ok(client: TestClient, db: Session):
+    """本账套建议 + 本账套 KPI → 200，落行 entity_id=会话账套、kpi_id=本账套 KPI"""
+    _seed_two_entities(db)
+    _mk_kpi(db, E1_KPI, 1, target=E1_TARGET)
+    k2 = _mk_kpi(db, E2_KPI, 2, target=E2_TARGET)
+    tok2 = _login(client, 2)
+
+    s_ok = _mk_suggestion(client, tok2, E2_SUG_TITLE, k2.id, k2.id)
+    r_ok = client.post(f"{SUGG}/{s_ok}/apply", headers=auth_header(tok2),
+                       json={"action": "action_plan", "title": "ZZ_AKR 本账套计划",
+                             "assignee": "张三", "priority": "high",
+                             "due_date": "2026-09-30"})
+    assert r_ok.status_code == 200, (
+        f"本账套 action_plan 应用应 200，实际 {r_ok.status_code} {r_ok.text[:300]}"
+    )
+
+    rows = _plan_rows(db, k2.id)
+    assert len(rows) == 1, f"本账套应用未落行（kpi_id={k2.id} 现有 {len(rows)} 行）"
+    plan = rows[0]
+    assert plan.entity_id == 2, (
+        f"落行 entity_id 应为会话账套 2，实际 {plan.entity_id}"
+    )
+    assert plan.kpi_id == k2.id, f"落行 kpi_id 应为 {k2.id}，实际 {plan.kpi_id}"
+    assert plan.title == "ZZ_AKR 本账套计划"
+    assert _sug_status(db, s_ok) == "applied", "本账套应用未落建议状态 applied"
+
+
+# ──────────────────────────────────────────────
+# T3. 预览：对端 KPI 不回显
+# ──────────────────────────────────────────────
+
+def test_preview_foreign_kpi_not_leaked(client: TestClient, db: Session):
+    """建议属本账套、其 kpi_id 属对端 → 预览不回显对端 KPI 名/current_target"""
+    _seed_two_entities(db)
+    k1 = _mk_kpi(db, E1_KPI, 1, target=E1_TARGET)
+    _mk_kpi(db, E2_KPI, 2, target=E2_TARGET)
+    tok2 = _login(client, 2)
+
+    s_cross = _mk_suggestion(client, tok2, CROSS_SUG_TITLE, k1.id, k1.id)
+
+    r = client.get(f"{SUGG}/{s_cross}/preview", headers=auth_header(tok2))
+    assert r.status_code == 200, f"预览应 200（降级语义），实际 {r.status_code} {r.text[:200]}"
+    data = r.json().get("data") or {}
+    blob = str(r.json())
+
+    assert f"{E1_KPI}-名称" not in blob, (
+        "预览回显了对端账套的 KPI 名（只读泄漏）"
+        " —— preview_suggestion 的 KPIDefinition 查询未校验归属会话账套"
+    )
+    assert f"KPI#{k1.id}" == data.get("kpi_name"), (
+        f"对端 KPI 应降级为占位名 KPI#{k1.id}，实际 {data.get('kpi_name')!r}"
+    )
+    assert data.get("current_target") is None, (
+        f"预览泄漏了对端账套 KPI 目标值：{data.get('current_target')!r}"
+    )
+
+
+# ──────────────────────────────────────────────
+# T4. 预览正对照：本账套 KPI 字段完整
+# ──────────────────────────────────────────────
+
+def test_preview_own_kpi_complete(client: TestClient, db: Session):
+    """本账套建议 + 本账套 KPI → kpi_name / current_target 完整回显"""
+    _seed_two_entities(db)
+    _mk_kpi(db, E1_KPI, 1, target=E1_TARGET)
+    k2 = _mk_kpi(db, E2_KPI, 2, target=E2_TARGET)
+    tok2 = _login(client, 2)
+
+    s_ok = _mk_suggestion(client, tok2, E2_SUG_TITLE, k2.id, k2.id, target_value=4500.0)
+    r = client.get(f"{SUGG}/{s_ok}/preview", headers=auth_header(tok2))
+    assert r.status_code == 200, f"本账套预览应 200，实际 {r.status_code} {r.text[:200]}"
+    data = r.json().get("data") or {}
+    assert data.get("kpi_name") == f"{E2_KPI}-名称", (
+        f"本账套预览未回显 KPI 名，实际 {data.get('kpi_name')!r}"
+    )
+    assert data.get("current_target") == E2_TARGET, (
+        f"本账套预览 current_target 应为 {E2_TARGET}，实际 {data.get('current_target')!r}"
+    )
+    assert data.get("new_target") == 4500.0, (
+        f"本账套预览 new_target 应为 4500.0，实际 {data.get('new_target')!r}"
+    )
+
+
+# ──────────────────────────────────────────────
+# T5. 鉴权底线（preview / apply 无 token → 401）
+# ──────────────────────────────────────────────
+
+def test_preview_and_apply_without_token_401(client: TestClient, db: Session):
+    """无 token 一律 401（改前改后都应通过）"""
+    _seed_two_entities(db)
+    k1 = _mk_kpi(db, E1_KPI, 1, target=E1_TARGET)
+    tok2 = _login(client, 2)
+    s = _mk_suggestion(client, tok2, CROSS_SUG_TITLE, k1.id, k1.id)
+
+    r1 = client.get(f"{SUGG}/{s}/preview")
+    assert r1.status_code == 401, f"preview 无 token 应 401，实际 {r1.status_code}"
+
+    r2 = client.post(f"{SUGG}/{s}/apply", json={"action": "action_plan"})
+    assert r2.status_code == 401, f"apply 无 token 应 401，实际 {r2.status_code}"