@@ -0,0 +1,375 @@
""" CMA 跨实体隔离 — ActionPlan 归属写侧「残余三处构造点」（2026-09-18 派生单）
任务：fullstack-20260918-2015-actionplan-entity-write-residual-sites
父单：fullstack-20260918-1905b-actionplan-entity-write-and-okr-read
（其 §五 范围外只读侦查登记的三处同族缺口）
缺陷（`ActionPlan.entity_id` 模型默认 1，以下三处构造均不写会话账套）：
1. app/api/alerts.py::create_action_plan_from_alert — 预警关联生成改善计划
2. app/api/deviation_push.py::push_deviation_to_map — 差异反打自动生成回顾会议题
3. app/api/ai_suggestions.py::_apply_action_plan — AI 建议转行动计划
后果：非 e1 账套经这三个入口创建的计划落 `entity_id=1`（不可信字段）。一旦读路径按
`ActionPlan.entity_id` 过滤（1905b 已在 OKR 分解视图加该条件），即出现跨账套泄漏或
「自己建的计划看不见」的假阴性。
本用例（改前失败条目见 _reports/evidence-fullstack-20260918-2015-red.txt）：
- T1 预警关联生成计划：e2 会话 → 库内 entity_id=2；e1 会话 → 1（双向 + 正对照）；
并附跨账套负对照（e2 token 打 e1 预警 → 404 且不落行）
- T2 差异反打生成议题：e2 会话 → 库内 entity_id=2（
e1 会话 → 1（双向）
- T3 差异反打去重按账套隔离：对端账套存在同名 pending 议题，不得抑制本账套创建
- T4 AI 建议转计划：e2 会话 → 库内 entity_id=2；
- T5 AI 建议越权负对照：对端账套的建议不得被本账套会话应用（404 + 零落行）
- T6 鉴权底线：三个入口无 token → 401（改前改后都应通过）
"""
import datetime as _dt
from datetime import time as _time
import pytest
from fastapi . testclient import TestClient
from sqlalchemy . orm import Session
from app . models import ActionPlan , AISuggestion , Entity , KPIAlert , KPIDefinition , StrategicMap
from tests . conftest import TEST_USER_PASSWORD , auth_header , create_test_user
ALERTS = " /api/cma/alerts "
DPUSH = " /api/cma/deviation-push/push-to-map "
SUGG = " /api/cma/ai/suggestions "
# kpi_code 全局唯一（models.KPIDefinition.kpi_code unique=True）→ 每个账套用独立 code
E1_ALERT_KPI = " ZZ_APRES_E1_ALERT_KPI "
E2_ALERT_KPI = " ZZ_APRES_E2_ALERT_KPI "
E1_DPUSH_KPI = " ZZ_APRES_E1_DPUSH_KPI "
E2_DPUSH_KPI = " ZZ_APRES_E2_DPUSH_KPI "
E1_SUG_KPI = " ZZ_APRES_E1_SUG_KPI "
E2_SUG_KPI = " ZZ_APRES_E2_SUG_KPI "
# ──────────────────────────────────────────────
# 夹具（沿用 1905b 同族风格：双向登录 + 同账套正对照）
# ──────────────────────────────────────────────
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 = 1000.0 ) - > 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 _mk_alert ( db : Session , eid : int , kpi_id : int , message : str ) - > KPIAlert :
alert = KPIAlert (
entity_id = eid ,
kpi_id = kpi_id ,
alert_level = " red " ,
alert_message = message ,
status = " pending " ,
)
db . add ( alert )
db . commit ( )
db . refresh ( alert )
return alert
def _mk_map ( db : Session , eid : int , title : str , kpi_code : str , obj_name : str ) - > StrategicMap :
m = StrategicMap (
entity_id = eid ,
title = title ,
version = " v1.0 " ,
status = " draft " ,
dimensions = [ {
" key " : " finance " , " name " : " 财务维度 " , " icon " : " F " , " color " : " #409eff " ,
" objectives " : [ { " name " : obj_name , " kpis " : [ kpi_code ] } ] ,
} ] ,
canvas_data = { " connections " : [ ] } ,
)
db . add ( m )
db . commit ( )
db . refresh ( m )
return m
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 _plans_by_title ( db : Session , title : str ) :
db . expire_all ( )
return db . query ( ActionPlan ) . filter ( ActionPlan . title == title ) . all ( )
def _alert_plan_title ( message : str ) - > str :
return f " 改善: { message } "
def _dpush_topic_title ( kpi : KPIDefinition , obj_name : str ) - > str :
return f " 【差异反打】 { kpi . kpi_name } 偏差回写 — { obj_name } "
def _period ( ) - > str :
return _dt . date . today ( ) . strftime ( " % Y- % m " )
# ──────────────────────────────────────────────
# T1. 预警关联生成计划 → 会话账套（双向 + 跨账套负对照）
# ──────────────────────────────────────────────
def test_alert_create_action_plan_writes_session_entity ( client : TestClient , db : Session ) :
""" e1 会话 → 库内 entity_id=1； """
_seed_two_entities ( db )
k1 = _mk_kpi ( db , E1_ALERT_KPI , 1 )
k2 = _mk_kpi ( db , E2_ALERT_KPI , 2 )
a1 = _mk_alert ( db , 1 , k1 . id , " ZZ_APRES e1 营收严重下滑 " )
a2 = _mk_alert ( db , 2 , k2 . id , " ZZ_APRES e2 营收严重下滑 " )
t1 = _alert_plan_title ( a1 . alert_message )
r1 = client . post ( f " { ALERTS } / { a1 . id } /create-action-plan " ,
headers = auth_header ( _login ( client , 1 ) ) , json = { " assignee " : " 甲 " } )
assert r1 . status_code == 200 , f " e1 创建应 200，实际 { r1 . status_code } { r1 . text [ : 200 ] } "
rows1 = _plans_by_title ( db , t1 )
assert len ( rows1 ) == 1 , f " e1 预警未生成计划: { rows1 } "
assert rows1 [ 0 ] . entity_id == 1 , (
f " e1 会话经预警生成计划的 entity_id 应为 1，实际 { rows1 [ 0 ] . entity_id } "
)
t2 = _alert_plan_title ( a2 . alert_message )
r2 = client . post ( f " { ALERTS } / { a2 . id } /create-action-plan " ,
headers = auth_header ( _login ( client , 2 ) ) , json = { " assignee " : " 乙 " } )
assert r2 . status_code == 200 , f " e2 创建应 200，实际 { r2 . status_code } { r2 . text [ : 200 ] } "
rows2 = _plans_by_title ( db , t2 )
assert len ( rows2 ) == 1 , f " e2 预警未生成计划: { rows2 } "
assert rows2 [ 0 ] . entity_id == 2 , (
f " e2 会话经预警生成计划的 entity_id 应为 2（会话账套），实际 { rows2 [ 0 ] . entity_id } "
" —— alerts.py 构造 ActionPlan 未写 entity_id，落模型默认值 1 "
)
assert rows2 [ 0 ] . kpi_id == k2 . id , (
f " e2 计划应挂 e2 自己的 KPI# { k2 . id } ，实际 # { rows2 [ 0 ] . kpi_id } "
)
def test_alert_create_action_plan_cross_entity_404 ( client : TestClient , db : Session ) :
""" e2 token 打 e1 预警 → 404（读侧既有护栏，不得因写侧修复而退化）+ 零落行 """
_seed_two_entities ( db )
k1 = _mk_kpi ( db , E1_ALERT_KPI , 1 )
a1 = _mk_alert ( db , 1 , k1 . id , " ZZ_APRES e1 跨账套负对照 " )
r = client . post ( f " { ALERTS } / { a1 . id } /create-action-plan " ,
headers = auth_header ( _login ( client , 2 ) ) , json = { " assignee " : " 乙 " } )
assert r . status_code == 404 , f " 跨账套应 404，实际 { r . status_code } { r . text [ : 200 ] } "
assert _plans_by_title ( db , _alert_plan_title ( a1 . alert_message ) ) == [ ] , (
" 跨账套 create-action-plan 竟为对端账套预警建了计划 "
)
# ──────────────────────────────────────────────
# T2. 差异反打生成议题 → 会话账套（双向）
# ──────────────────────────────────────────────
def test_deviation_push_writes_session_entity ( client : TestClient , db : Session ) :
""" e1 会话 → 议题 entity_id=1； ） """
_seed_two_entities ( db )
k1 = _mk_kpi ( db , E1_DPUSH_KPI , 1 )
k2 = _mk_kpi ( db , E2_DPUSH_KPI , 2 )
m1 = _mk_map ( db , 1 , " ZZ_APRES e1 地图 " , E1_DPUSH_KPI , " e1 目标 " )
m2 = _mk_map ( db , 2 , " ZZ_APRES e2 地图 " , E2_DPUSH_KPI , " e2 目标 " )
t1 = _dpush_topic_title ( k1 , " e1 目标 " )
r1 = client . post ( DPUSH , headers = auth_header ( _login ( client , 1 ) ) , json = {
" mapId " : m1 . id , " nodeId " : " finance-0 " , " newValue " : 900.0 ,
" period " : _period ( ) , " createReviewTopic " : True ,
} )
assert r1 . status_code == 200 , f " e1 反打应 200，实际 { r1 . status_code } { r1 . text [ : 300 ] } "
assert r1 . json ( ) [ " review_topic_created " ] is True , r1 . text [ : 200 ]
rows1 = _plans_by_title ( db , t1 )
assert len ( rows1 ) == 1 , f " e1 反打未生成议题: { rows1 } "
assert rows1 [ 0 ] . entity_id == 1 , (
f " e1 会话经差异反打生成议题的 entity_id 应为 1，实际 { rows1 [ 0 ] . entity_id } "
)
t2 = _dpush_topic_title ( k2 , " e2 目标 " )
r2 = client . post ( DPUSH , headers = auth_header ( _login ( client , 2 ) ) , json = {
" mapId " : m2 . id , " nodeId " : " finance-0 " , " newValue " : 900.0 ,
" period " : _period ( ) , " createReviewTopic " : True ,
} )
assert r2 . status_code == 200 , f " e2 反打应 200，实际 { r2 . status_code } { r2 . text [ : 300 ] } "
assert r2 . json ( ) [ " review_topic_created " ] is True , r2 . text [ : 200 ]
rows2 = _plans_by_title ( db , t2 )
assert len ( rows2 ) == 1 , f " e2 反打未生成议题: { rows2 } "
assert rows2 [ 0 ] . entity_id == 2 , (
f " e2 会话经差异反打生成议题的 entity_id 应为 2（会话账套），实际 { rows2 [ 0 ] . entity_id } "
" —— deviation_push.py 构造 ActionPlan 未写 entity_id，落模型默认值 1 "
)
assert rows2 [ 0 ] . kpi_id == k2 . id , (
f " e2 议题应挂 e2 自己的 KPI# { k2 . id } ，实际 # { rows2 [ 0 ] . kpi_id } "
)
# ──────────────────────────────────────────────
# T3. 差异反打去重按账套隔离（对端同名议题不得抑制本账套创建）
# ──────────────────────────────────────────────
def test_deviation_push_dedupe_scoped_to_entity ( client : TestClient , db : Session ) :
""" e1 已有同名 pending 议题（含历史遗留 entity_id=1 行）不得抑制 e2 创建自己的议题 """
_seed_two_entities ( db )
_mk_kpi ( db , E1_DPUSH_KPI , 1 )
k2 = _mk_kpi ( db , E2_DPUSH_KPI , 2 )
m2 = _mk_map ( db , 2 , " ZZ_APRES e2 地图去重 " , E2_DPUSH_KPI , " e2 去重目标 " )
t2 = _dpush_topic_title ( k2 , " e2 去重目标 " )
# 对端账套（e1）先落一条同标题的 pending 议题（模拟历史遗留/另一账套真实办理中）
foreign = ActionPlan (
entity_id = 1 , kpi_id = db . query ( KPIDefinition ) . filter (
KPIDefinition . kpi_code == E1_DPUSH_KPI ) . first ( ) . id ,
title = t2 , assignee = " 甲 " , priority = " medium " , status = " pending " , progress = 0 ,
)
db . add ( foreign )
db . commit ( )
r2 = client . post ( DPUSH , headers = auth_header ( _login ( client , 2 ) ) , json = {
" mapId " : m2 . id , " nodeId " : " finance-0 " , " newValue " : 900.0 ,
" period " : _period ( ) , " createReviewTopic " : True ,
} )
assert r2 . status_code == 200 , f " e2 反打应 200，实际 { r2 . status_code } { r2 . text [ : 300 ] } "
assert r2 . json ( ) [ " review_topic_created " ] is True , (
" 对端账套同名议题抑制了本账套创建（去重条件未按账套隔离） "
)
rows = _plans_by_title ( db , t2 )
assert len ( rows ) == 2 , f " 应有 e1 遗留 + e2 新建共 2 条，实际 { len ( rows ) } "
own = [ x for x in rows if x . entity_id == 2 ]
assert len ( own ) == 1 , (
f " e2 本账套议题未落库或落错账套: { [ ( x . id , x . entity_id ) for x in rows ] } "
)
# ──────────────────────────────────────────────
# T4. AI 建议转行动计划 → 会话账套（双向）
# ──────────────────────────────────────────────
def _mk_suggestion ( client : TestClient , token : str , kpi_id : int , title : str ) :
body = {
" suggestion_type " : " action_plan " ,
" target_type " : " kpi " ,
" target_id " : kpi_id ,
" title " : title ,
" content " : " 达成率低于阈值 " ,
" suggestion_data " : { " kpi_id " : kpi_id } ,
}
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 test_ai_suggestion_action_plan_writes_session_entity ( client : TestClient , db : Session ) :
""" e1 会话 → 计划 entity_id=1； ） """
_seed_two_entities ( db )
k1 = _mk_kpi ( db , E1_SUG_KPI , 1 )
k2 = _mk_kpi ( db , E2_SUG_KPI , 2 )
tok1 = _login ( client , 1 )
s1 = _mk_suggestion ( client , tok1 , k1 . id , " ZZ_APRES e1 建议转计划 " )
r1 = client . post ( f " { SUGG } / { s1 } /apply " , headers = auth_header ( tok1 ) ,
json = { " action " : " action_plan " , " title " : " ZZ_APRES e1 转出的计划 " } )
assert r1 . status_code == 200 , f " e1 应用应 200，实际 { r1 . status_code } { r1 . text [ : 300 ] } "
rows1 = _plans_by_title ( db , " ZZ_APRES e1 转出的计划 " )
assert len ( rows1 ) == 1 , f " e1 应用未生成计划: { rows1 } "
assert rows1 [ 0 ] . entity_id == 1 , (
f " e1 会话经 AI 建议生成的计划 entity_id 应为 1，实际 { rows1 [ 0 ] . entity_id } "
)
tok2 = _login ( client , 2 )
s2 = _mk_suggestion ( client , tok2 , k2 . id , " ZZ_APRES e2 建议转计划 " )
r2 = client . post ( f " { SUGG } / { s2 } /apply " , headers = auth_header ( tok2 ) ,
json = { " action " : " action_plan " , " title " : " ZZ_APRES e2 转出的计划 " } )
assert r2 . status_code == 200 , f " e2 应用应 200，实际 { r2 . status_code } { r2 . text [ : 300 ] } "
rows2 = _plans_by_title ( db , " ZZ_APRES e2 转出的计划 " )
assert len ( rows2 ) == 1 , f " e2 应用未生成计划: { rows2 } "
assert rows2 [ 0 ] . entity_id == 2 , (
f " e2 会话经 AI 建议生成的计划 entity_id 应为 2（会话账套），实际 { rows2 [ 0 ] . entity_id } "
" —— ai_suggestions.py 构造 ActionPlan 未写 entity_id，落模型默认值 1 "
)
def test_ai_suggestion_apply_cross_entity_404 ( client : TestClient , db : Session ) :
""" e1 会话应用 e2 的建议 → 404（建议须归属会话账套）+ 零落行；本账套应用正常（正对照） """
_seed_two_entities ( db )
_mk_kpi ( db , E1_SUG_KPI , 1 )
k2 = _mk_kpi ( db , E2_SUG_KPI , 2 )
tok2 = _login ( client , 2 )
s2 = _mk_suggestion ( client , tok2 , k2 . id , " ZZ_APRES e2 私有建议 " )
tok1 = _login ( client , 1 )
r = client . post ( f " { SUGG } / { s2 } /apply " , headers = auth_header ( tok1 ) ,
json = { " action " : " action_plan " , " title " : " ZZ_APRES 越权转出的计划 " } )
assert r . status_code == 404 , f " 跨账套应用应 404，实际 { r . status_code } { r . text [ : 300 ] } "
assert _plans_by_title ( db , " ZZ_APRES 越权转出的计划 " ) == [ ] , (
" 跨账套应用竟为对端账套建议建了计划 "
)
db . expire_all ( )
sug = db . query ( AISuggestion ) . filter ( AISuggestion . id == s2 ) . first ( )
assert sug . status == " unapplied " , f " 越权应用改动了建议状态: { sug . status } "
# 正对照：本账套会话应用自己账套的建议仍然正常
r_ok = client . post ( f " { SUGG } / { s2 } /apply " , headers = auth_header ( tok2 ) ,
json = { " action " : " action_plan " , " title " : " ZZ_APRES e2 自救计划 " } )
assert r_ok . status_code == 200 , f " 本账套应用应 200，实际 { r_ok . status_code } { r_ok . text [ : 300 ] } "
assert len ( _plans_by_title ( db , " ZZ_APRES e2 自救计划 " ) ) == 1
# ──────────────────────────────────────────────
# T5. 鉴权底线（三个入口无 token → 401）
# ──────────────────────────────────────────────
def test_residual_sites_without_token_401 ( client : TestClient , db : Session ) :
""" 三个入口无 token 一律 401（改前改后都应通过） """
_seed_two_entities ( db )
k1 = _mk_kpi ( db , E1_ALERT_KPI , 1 )
a1 = _mk_alert ( db , 1 , k1 . id , " ZZ_APRES e1 鉴权底线 " )
m1 = _mk_map ( db , 1 , " ZZ_APRES e1 鉴权地图 " , E1_ALERT_KPI , " e1 鉴权目标 " )
r1 = client . post ( f " { ALERTS } / { a1 . id } /create-action-plan " , json = { } )
assert r1 . status_code == 401 , f " 预警入口无 token 应 401，实际 { r1 . status_code } "
r2 = client . post ( DPUSH , json = { " mapId " : m1 . id , " nodeId " : " finance-0 " ,
" newValue " : 800.0 , " createReviewTopic " : False } )
assert r2 . status_code == 401 , f " 差异反打无 token 应 401，实际 { r2 . status_code } "
r3 = client . post ( SUGG , json = { " suggestion_type " : " action_plan " , " title " : " 无 token " } )
assert r3 . status_code == 401 , f " AI 建议无 token 应 401，实际 { r3 . status_code } "
Hermes CI Fix