diff --git a/backend/app/api/ai_suggestions.py b/backend/app/api/ai_suggestions.py index 273696f8..4a25d502 100644 --- a/backend/app/api/ai_suggestions.py +++ b/backend/app/api/ai_suggestions.py @@ -236,7 +236,10 @@ def _apply_action_plan(db: Session, sug: AISuggestion, params: dict, current_use kpi_id = params.get("kpi_id") or sug.target_id if not kpi_id: raise HTTPException(400, "缺少 kpi_id") - kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == kpi_id, + KPIDefinition.entity_id == sug.entity_id, # 2026-09-18 派生单 2030b: 目标KPI须归属建议账套 + ).first() if not kpi: raise HTTPException(404, f"KPI {kpi_id} 不存在") @@ -311,7 +314,11 @@ def preview_suggestion(suggestion_id: int, db: Session = Depends(get_db), kpi = None kpi_id = sd.get("kpi_id") or sug.target_id if kpi_id: - kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first() + # 2026-09-18 派生单 2030b: 目标KPI须归属建议账套;对端KPI降级为 KPI#(不回显名/目标值) + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == kpi_id, + KPIDefinition.entity_id == sug.entity_id, + ).first() if sug.suggestion_type == "kpi_target": return {"data": { diff --git a/backend/tests/test_ai_suggestion_kpi_attribution_residual_20260918.py b/backend/tests/test_ai_suggestion_kpi_attribution_residual_20260918.py new file mode 100644 index 00000000..19992e1a --- /dev/null +++ b/backend/tests/test_ai_suggestion_kpi_attribution_residual_20260918.py @@ -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#`、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# +- 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}"