sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(cma): entity isolation batch4 (cash plan writes, budget ownership, read leaks, okr template)

Browse Source 
P0-A cash.py: scope api_update_plan / api_delete_plan / api_complete_plan /
api_register_payment to session entity (cross-entity id -> 404, aligning with
the file's existing get_entity_id endpoints).
P0-B budget.py: write entity_id on BudgetPlan creation + require the KPI to
belong to the session entity; scope /comparison, /comparison/kpi/{id} and
/deviation-report reads to the session entity.
P1 notifications.py: explicit field whitelist (name/config/enabled) on
update_channel to close mass-assignment.
P1 read leaks scoped to session entity: bot_kpis list, data_quality check/stats
(KPI calibre only; quality-log table has no entity_id column), kpis causality
chain + associate-map (KPI and map ownership), dashboard my-kpis,
customer_dashboard list, kpi_governance audit, budget_generate candidates,
driver_budget history (+ entity_id on its write path), expenses stats
(+ entity_id on its write path).
P2 okr_templates.py: write entity_id on the StrategicMap created by apply.

tests: tests/test_cross_entity_isolation_batch4_20260921.py (30 cases;
RED before fix = 22 failed, GREEN after = 30 passed; full suite 1302 passed /
1 xfailed / 0 failed).
This commit is contained in:
Hermes CI Fix
2026-09-21 09:37:38 +08:00
parent 8c5853f64b
commit a0762b367f
14 changed files with 749 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/bot_kpis.py
+3 -1
View File
@@ -62,11 +62,13 @@ def list_bot_kpis(
source: str = Query("finance-bot", description="Bot标识"),
period: Optional[str] = None,
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""获取某Bot的所有KPI（含评分）"""
"""获取某Bot的所有KPI（含评分）（账套隔离 2026-09-21：仅本账套，对齐 update_bot_kpi_value）"""
kpis = db.query(KPIDefinition).filter(
KPIDefinition.bot_source == source,
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).order_by(KPIDefinition.kpi_code).all()
if not kpis:
backend/app/api/budget.py
+24 -6
View File
@@ -77,8 +77,9 @@ def create_budget_plan(
data: dict,
db: Session = Depends(get_db),
current_user=Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""创建或更新单条预算计划"""
"""创建或更新单条预算计划（账套隔离 2026-09-21：KPI 归属校验 + 写会话账套）"""
kpi_id = data.get("kpi_id")
period = data.get("period")
budget_value = data.get("budget_value")
@@ -86,8 +87,11 @@ def create_budget_plan(
if not all([kpi_id, period, budget_value is not None]):
raise HTTPException(400, "缺少必要参数: kpi_id, period, budget_value")
# 检查KPI是否存在
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
# 检查KPI是否存在（账套隔离 2026-09-21：仅本账套 KPI，防把预算挂到对端 KPI）
kpi = db.query(KPIDefinition).filter(
KPIDefinition.id == kpi_id,
KPIDefinition.entity_id == entity_id,
).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
@@ -112,6 +116,7 @@ def create_budget_plan(
return {"message": "预算已更新", "id": existing.id}
else:
plan = BudgetPlan(
entity_id=entity_id, # 账套隔离 2026-09-21：原漏写→落模型默认 1
kpi_id=kpi_id,
map_id=map_id,
period=period,
@@ -394,6 +399,7 @@ def get_deviation_report(
dimension: Optional[str] = Query(None),
alert_level: Optional[str] = Query(None),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""获取差异分析报告（汇总多个KPI的实际vs预算差异）"""
if period is None:
@@ -404,7 +410,10 @@ def get_deviation_report(
else:
period = datetime.now().strftime("%Y-%m")
query = db.query(KPIDefinition).filter(KPIDefinition.status == "active")
query = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
)
if kpi_id:
query = query.filter(KPIDefinition.id == kpi_id)
if dimension:
@@ -640,6 +649,7 @@ def get_budget_comparison(
kpi_id: Optional[int] = Query(None),
year: Optional[int] = Query(None),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""
获取实际值vs预测值对比数据
@@ -685,7 +695,11 @@ def get_budget_comparison(
)
if kpi_id:
query = query.filter(BudgetPlan.kpi_id == kpi_id)
query = query.filter(BudgetPlan.period.in_(periods), BudgetPlan.status == "active")
query = query.filter(
BudgetPlan.period.in_(periods),
BudgetPlan.status == "active",
BudgetPlan.entity_id == entity_id, # 账套隔离 2026-09-21
)
budget_plans = query.all()
# 按KPI+期间索引
@@ -761,6 +775,7 @@ def get_kpi_comparison(
kpi_id: int,
year: Optional[int] = Query(None),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""
获取单个KPI的实际vs预测对比数据（用于图表展示）
@@ -771,7 +786,10 @@ def get_kpi_comparison(
now = datetime.now()
y = year or now.year
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
kpi = db.query(KPIDefinition).filter(
KPIDefinition.id == kpi_id,
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
backend/app/api/budget_generate.py
+7 -2
View File
@@ -5,6 +5,7 @@
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_auth, require_role
from app.models import KPIDefinition, BudgetPlan, KPIValue, OperationLog
import json
@@ -75,12 +76,16 @@ def _calc_budget(kpi: KPIDefinition) -> dict:
def get_kpi_budget_candidates(
year: int = None,
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""获取可用于生成预算的KPI列表，按类型分类"""
"""获取本账套可用于生成预算的KPI列表，按类型分类（账套隔离 2026-09-21）"""
if not year:
year = datetime.now().year
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).all()
# 获取每个KPI的最新实际值
latest_values = {}
backend/app/api/cash.py
+25 -12
View File
@@ -198,9 +198,12 @@ def api_create_plan(request: Request, data: dict, db: Session = Depends(get_db))
@router.put("/plans/{plan_id}")
def api_update_plan(plan_id: int, data: dict, db: Session = Depends(get_db)):
"""更新收付款计划"""
plan = db.query(CashPlan).filter(CashPlan.id == plan_id).first()
def api_update_plan(plan_id: int, data: dict, db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id)):
"""更新收付款计划（账套隔离 2026-09-21：跨账套 id → 404）"""
plan = db.query(CashPlan).filter(
CashPlan.id == plan_id, CashPlan.entity_id == entity_id,
).first()
if not plan:
raise HTTPException(404, "计划不存在")
if "plan_type" in data:
@@ -248,9 +251,12 @@ def api_update_plan(plan_id: int, data: dict, db: Session = Depends(get_db)):
@router.delete("/plans/{plan_id}")
def api_delete_plan(plan_id: int, db: Session = Depends(get_db)):
"""删除收付款计划"""
plan = db.query(CashPlan).filter(CashPlan.id == plan_id).first()
def api_delete_plan(plan_id: int, db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id)):
"""删除收付款计划（账套隔离 2026-09-21：跨账套 id → 404）"""
plan = db.query(CashPlan).filter(
CashPlan.id == plan_id, CashPlan.entity_id == entity_id,
).first()
if not plan:
raise HTTPException(404, "计划不存在")
db.delete(plan)
@@ -259,9 +265,12 @@ def api_delete_plan(plan_id: int, db: Session = Depends(get_db)):
@router.post("/plans/{plan_id}/complete")
def api_complete_plan(plan_id: int, db: Session = Depends(get_db)):
"""标记计划为已完成（收款到账/付款完成）"""
plan = db.query(CashPlan).filter(CashPlan.id == plan_id).first()
def api_complete_plan(plan_id: int, db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id)):
"""标记计划为已完成（收款到账/付款完成）（账套隔离 2026-09-21：跨账套 id → 404）"""
plan = db.query(CashPlan).filter(
CashPlan.id == plan_id, CashPlan.entity_id == entity_id,
).first()
if not plan:
raise HTTPException(404, "计划不存在")
plan.status = "completed"
@@ -616,9 +625,13 @@ def api_receivables(
@router.post("/receivables/{plan_id}/payment")
def api_register_payment(plan_id: int, data: dict, db: Session = Depends(get_db)):
"""回款登记 — 更新计划状态 + 自动消除到期未收款预警 + 联动催收行动/F_AR_DAYS"""
plan = db.query(CashPlan).filter(CashPlan.id == plan_id).first()
def api_register_payment(plan_id: int, data: dict, db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id)):
"""回款登记 — 更新计划状态 + 自动消除到期未收款预警 + 联动催收行动/F_AR_DAYS
（账套隔离 2026-09-21：跨账套 id → 404）"""
plan = db.query(CashPlan).filter(
CashPlan.id == plan_id, CashPlan.entity_id == entity_id,
).first()
if not plan:
raise HTTPException(404, "应收计划不存在")
if plan.plan_type != "receive":
backend/app/api/customer_dashboard.py
+3
View File
@@ -5,6 +5,7 @@ from sqlalchemy import func, desc
from typing import Optional
from datetime import datetime, timedelta
from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_auth, require_role
from app.models import KPIDefinition, KPIValue, KPIAlert, User
import logging
@@ -45,6 +46,7 @@ def list_customer_kpis(
end_date: str = Query(None),
db: Session = Depends(get_db),
current_user: User = Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""获取客户维度KPI列表（含最新值、预警、趋势）"""
start, end = parse_period(period, start_date, end_date)
@@ -54,6 +56,7 @@ def list_customer_kpis(
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.dimension == "customer",
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
).order_by(KPIDefinition.kpi_code).all()
result = []
backend/app/api/dashboard.py
+5 -1
View File
@@ -360,6 +360,7 @@ def get_my_kpis(
current_user: User = Depends(require_auth),
period: str = Query("month"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""获取当前用户负责的KPI
- business角色：只看自己负责的KPI
@@ -370,7 +371,10 @@ def get_my_kpis(
name = current_user.name
period_str = datetime.now().strftime("%Y-%m")
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
).all()
result = []
for k in kpis:
backend/app/api/data_quality.py
+17 -8
View File
@@ -10,7 +10,7 @@ import json
import logging
from app.database import get_db
from app.deps import assert_entity_scope
from app.deps import assert_entity_scope, get_entity_id
from app.auth_middleware import require_auth, require_role
from app.models import KPIDefinition, KPIValue, KpiDataQualityLog, OperationLog
from app.api.kpis import kpi_to_dict
@@ -36,9 +36,12 @@ def _log_to_dict(log):
# ============================================================
@router.get("/check")
def run_quality_check(db: Session = Depends(get_db)):
"""扫描全部KPI，生成数据质量报告"""
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
def run_quality_check(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""扫描本账套KPI，生成数据质量报告（账套隔离 2026-09-21）"""
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).all()
issues = []
current_period = datetime.now().strftime("%Y-%m")
@@ -228,9 +231,12 @@ def delete_quality_log(log_id: int, db: Session = Depends(get_db), user=WRITE_RO
# ============================================================
@router.get("/stats")
def quality_stats(db: Session = Depends(get_db)):
"""数据质量统计"""
total_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").count()
def quality_stats(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)):
"""数据质量统计（账套隔离 2026-09-21：KPI 口径按会话账套；质量日志表无 entity_id 列，见回报）"""
total_kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).count()
total_logs = db.query(KpiDataQualityLog).count()
open_logs = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.status == "open").count()
@@ -256,7 +262,10 @@ def quality_stats(db: Session = Depends(get_db)):
# ── 数据审计看板统计 ──
# KPI完整度评分
all_kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
all_kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id, # 账套隔离 2026-09-21
).all()
total = len(all_kpis)
complete_kpis = 0
missing_metadata_count = 0
backend/app/api/driver_budget.py
+8 -2
View File
@@ -7,6 +7,7 @@ from typing import Optional, List
from datetime import datetime
from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_auth, require_role
from app.models import SystemConfig, OperationLog
from app.models.driver_budget import DriverFactorTemplate, DriverFactorBudget
@@ -254,6 +255,7 @@ def driver_calculate(
data: dict,
db: Session = Depends(get_db),
current_user=Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""
驱动因子计算预算。
@@ -313,6 +315,7 @@ def driver_calculate(
# 记录计算结果
budget_record = DriverFactorBudget(
entity_id=entity_id, # 账套隔离 2026-09-21：写侧先写归属（否则读侧过滤误杀本账套行）
name=name,
industry=industry,
template_id=template_id,
@@ -457,9 +460,12 @@ def driver_sensitivity(
def list_driver_history(
limit: int = Query(50, ge=1, le=200),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
):
"""查看最近的驱动因子预算计算记录"""
records = db.query(DriverFactorBudget).order_by(
"""查看本账套最近的驱动因子预算计算记录（账套隔离 2026-09-21）"""
records = db.query(DriverFactorBudget).filter(
DriverFactorBudget.entity_id == entity_id,
).order_by(
DriverFactorBudget.created_at.desc()
).limit(limit).all()
backend/app/api/expenses.py
+11 -2
View File
@@ -343,6 +343,7 @@ def submit_reimbursement(
data: dict,
db: Session = Depends(get_db),
current_user=Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""提交报销单 → 自动校验规则（超限自动打回）"""
applicant = data.get("applicant") or (current_user.name if hasattr(current_user, "name") else current_user.username)
@@ -366,6 +367,7 @@ def submit_reimbursement(
expense_date = None
reimb = ExpenseReimbursement(
entity_id=entity_id, # 账套隔离 2026-09-21：写侧先写归属（否则读侧过滤误杀本账套行）
reimb_no="BX" + datetime.now().strftime("%Y%m%d%H%M%S") + f"{random.randint(100, 999)}",
applicant=applicant,
department=data.get("department", ""),
@@ -537,6 +539,7 @@ def resubmit_reimbursement(
def expense_stats(
period: str = Query(None, description="期间 YYYY-MM，默认当前月"),
db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
_=Depends(require_auth),
):
"""费用审核看板统计: 状态计数 / 费用类型统计 / 超限预警 / 本月总额"""
@@ -544,7 +547,9 @@ def expense_stats(
period = period or now.strftime("%Y-%m")
counts = {"pending": 0, "approved": 0, "rejected": 0, "returned": 0, "total": 0}
for st, cnt in db.query(ExpenseReimbursement.status, func.count(ExpenseReimbursement.id)).group_by(
for st, cnt in db.query(ExpenseReimbursement.status, func.count(ExpenseReimbursement.id)).filter(
ExpenseReimbursement.entity_id == entity_id, # 账套隔离 2026-09-21
).group_by(
ExpenseReimbursement.status
).all():
if st in counts:
@@ -560,6 +565,7 @@ def expense_stats(
)
.filter(
ExpenseReimbursement.status.in_(["pending", "approved"]),
ExpenseReimbursement.entity_id == entity_id, # 账套隔离 2026-09-21
func.date_format(ExpenseReimbursement.expense_date, "%Y-%m") == period,
)
.group_by(ExpenseReimbursement.expense_type)
@@ -583,6 +589,7 @@ def expense_stats(
.filter(
ExpenseReimbursement.status == "returned",
ExpenseReimbursement.check_result == "fail",
ExpenseReimbursement.entity_id == entity_id, # 账套隔离 2026-09-21
)
.order_by(ExpenseReimbursement.id.desc())
.limit(50)
@@ -604,7 +611,9 @@ def expense_stats(
for r in over_limit_rows
]
recent_rows = db.query(ExpenseReimbursement).order_by(ExpenseReimbursement.id.desc()).limit(10).all()
recent_rows = db.query(ExpenseReimbursement).filter(
ExpenseReimbursement.entity_id == entity_id, # 账套隔离 2026-09-21
).order_by(ExpenseReimbursement.id.desc()).limit(10).all()
recent = [_reimb_to_dict(r) for r in recent_rows]
# 本月预算使用率(招待费15万/月 管理费91万/月)
backend/app/api/kpi_governance.py
+7 -2
View File
@@ -15,6 +15,7 @@ from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_auth, require_role
from app.models import KPIDefinition
@@ -166,9 +167,13 @@ def validate_kpi(
def governance_audit(
db: Session = Depends(get_db),
current_user=Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""全量校验所有活跃KPI，输出按4条规则分组的不合规清单。"""
kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all()
"""校验本账套活跃KPI，输出按4条规则分组的不合规清单（账套隔离 2026-09-21）。"""
kpis = db.query(KPIDefinition).filter(
KPIDefinition.status == "active",
KPIDefinition.entity_id == entity_id,
).all()
non_compliant = []
by_rule: dict = {1: [], 2: [], 3: [], 4: []}
backend/app/api/kpis.py
+17 -6
View File
@@ -437,9 +437,13 @@ def get_kpi_causality_chain(
kpi_id: int,
db: Session = Depends(get_db),
current_user = Depends(require_auth),
entity_id: int = Depends(get_entity_id),
):
"""KPI因果链追踪 — 返回单个KPI的上下游因果链"""
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
"""KPI因果链追踪 — 返回单个KPI的上下游因果链（账套隔离 2026-09-21：跨账套 id → 404）"""
kpi = db.query(KPIDefinition).filter(
KPIDefinition.id == kpi_id,
KPIDefinition.entity_id == entity_id,
).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
@@ -884,14 +888,21 @@ def kpi_to_dict(k):
@router.put("/{kpi_id}/associate-map")
def associate_kpi_map(kpi_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
"""关联KPI到战略地图"""
kpi = db.query(KPIDefinition).filter(KPIDefinition.id == kpi_id).first()
def associate_kpi_map(kpi_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES,
entity_id: int = Depends(get_entity_id)):
"""关联KPI到战略地图（账套隔离 2026-09-21：KPI 与地图均须属本账套，跨账套 → 404）"""
kpi = db.query(KPIDefinition).filter(
KPIDefinition.id == kpi_id,
KPIDefinition.entity_id == entity_id,
).first()
if not kpi:
raise HTTPException(404, "KPI不存在")
map_id = data.get("map_id")
if map_id is not None:
m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
m = db.query(StrategicMap).filter(
StrategicMap.id == map_id,
StrategicMap.entity_id == entity_id,
).first()
if not m:
raise HTTPException(404, "战略地图不存在")
kpi.map_id = map_id
backend/app/api/notifications.py
+3 -1
View File
@@ -52,8 +52,10 @@ def update_channel(ch_id: int, data: dict, db: Session = Depends(get_db)):
ch = db.query(NotificationChannel).filter(NotificationChannel.id == ch_id).first()
if not ch:
raise HTTPException(404, "渠道不存在")
# 字段白名单（2026-09-21 mass-assignment 收口）：仅业务字段，其余键忽略
_updatable = ("name", "config", "enabled")
for k, v in data.items():
if hasattr(ch, k) and k not in ("id", "created_at"):
if k in _updatable:
setattr(ch, k, v)
db.commit()
db.refresh(ch)
backend/app/api/okr_templates.py
+4 -1
View File
@@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import Optional
from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_role
from app.models import OKRTemplate, StrategicMap
from app.utils.kr_name import sanitize_kr_name
@@ -253,7 +254,8 @@ def increment_use_count(template_id: int, db: Session = Depends(get_db)):
@router.post("/{template_id}/apply")
def apply_okr_template(template_id: int, data: dict, db: Session = Depends(get_db)):
def apply_okr_template(template_id: int, data: dict, db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id)):
"""应用 OKR 模板 — 创建战略地图并填入 O+KR"""
t = db.query(OKRTemplate).filter(OKRTemplate.id == template_id).first()
if not t:
@@ -293,6 +295,7 @@ def apply_okr_template(template_id: int, data: dict, db: Session = Depends(get_d
})
m = StrategicMap(
entity_id=entity_id, # 账套隔离 2026-09-21：原漏写→落模型默认 1（对齐 maps.py）
title=map_title,
version=data.get("version", "v1.0"),
status="draft",
backend/tests/test_cross_entity_isolation_batch4_20260921.py
+615
View File
@@ -0,0 +1,615 @@
"""CMA 跨实体隔离收口（第四批）账套隔离回归测试 — 2026-09-21
任务：project-20260921-cma-entity-isolation-batch4（项目Bot 派单；来源 OpenCode 周审 20260921）
缺陷（本单范围）：
P0-A cash.py 4 端点无账套校验（跨租户改/删/完成/回款）
PUT /api/cma/cash/plans/{plan_id}
DELETE /api/cma/cash/plans/{plan_id}
POST /api/cma/cash/plans/{plan_id}/complete
POST /api/cma/cash/receivables/{plan_id}/payment
P0-B budget.py create_budget_plan 建行未写 entity_id（落默认 1，错归属）
P1 notifications.py update_channel 无字段白名单（mass-assignment）
P1 跨实体只读泄漏点（bot_kpis / data_quality / kpis / dashboard /
customer_dashboard / kpi_governance / budget_generate / budget 读 /
driver_budget / expenses）
P2 okr_templates.py apply_okr_template 建 StrategicMap 未写 entity_id
每条均「跨账套负例 + 本账套正对照」；改前 RED 原文见回报 §三。
本文件只跑内存 SQLite（tests/conftest.py 机制），零生产库 / 零生产 Redis 连接。
"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import (
BudgetPlan,
CashPlan,
Entity,
ExpenseReimbursement,
KPIDefinition,
KpiDataQualityLog,
NotificationChannel,
OKRTemplate,
StrategicMap,
DriverFactorBudget,
)
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
CASH = "/api/cma/cash"
BUDGET = "/api/cma/budget"
NOTIF = "/api/cma/notifications"
KPIS = "/api/cma/kpis"
DQ = "/api/cma/data-quality"
DASH = "/api/cma/dashboard"
CUST = "/api/cma/customer-dashboard"
GOV = "/api/cma/kpi"
OKRT = "/api/cma/okr-templates"
EXPENSES = "/api/cma/expenses"
BOT_SRC = "batch4-bot"
# 两账套资源编码前缀（kpi_code 库内唯一）
E1_KPI = "ZZ_B4_E1_KPI_001"
E2_KPI = "ZZ_B4_E2_KPI_001"
E1_CUST = "ZZ_B4_E1_CUST_001"
E2_CUST = "ZZ_B4_E2_CUST_001"
E1_BOT = "ZZ_B4_E1_BOT_001"
E2_BOT = "ZZ_B4_E2_BOT_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 _mk_kpi(db: Session, code: str, eid: int, dimension: str = "finance", **kw) -> KPIDefinition:
kpi = KPIDefinition(
entity_id=eid,
kpi_code=code,
kpi_name=f"{code}-名称",
dimension=dimension,
category="revenue_growth",
formula="-",
data_source_type="excel",
data_source="Excel导入",
data_owner="财务部",
frequency="monthly",
unit="元",
target_value=0,
kpi_level="operational",
status="active",
**kw,
)
db.add(kpi)
db.commit()
db.refresh(kpi)
return kpi
def _mk_plan(db: Session, eid: int, amount: float = 10.0, plan_type: str = "receive",
status: str = "pending", paid_amount: float = 0.0, **kw) -> CashPlan:
from datetime import datetime
plan = CashPlan(
entity_id=eid, plan_type=plan_type, amount=amount,
plan_date=datetime(2026, 9, 30), counterparty=f"e{eid}客户",
description=f"e{eid}计划", status=status, source="manual",
paid_amount=paid_amount, **kw,
)
db.add(plan)
db.commit()
db.refresh(plan)
return plan
def _mk_map(db: Session, eid: int, title: str) -> StrategicMap:
m = StrategicMap(entity_id=eid, title=title, status="draft",
dimensions=[], canvas_data={"connections": []})
db.add(m)
db.commit()
db.refresh(m)
return m
def _mk_channel(db: Session, name: str) -> NotificationChannel:
ch = NotificationChannel(name=name, channel_type="wecom", config={"webhook": "x"}, enabled=True)
db.add(ch)
db.commit()
db.refresh(ch)
return ch
def _mk_budget(db: Session, kpi_id: int, eid: int, period: str = "2026-09", value: float = 100.0):
bp = BudgetPlan(entity_id=eid, kpi_id=kpi_id, period=period, budget_value=value,
budget_year=2026, budget_month=9, version="v1.0", status="active")
db.add(bp)
db.commit()
db.refresh(bp)
return bp
def _mk_template(db: Session, name: str) -> OKRTemplate:
t = OKRTemplate(name=name, description="批4测试模板", dimension="finance",
layer="level1", preset_krs=[{"name": "KR测试", "target_value": 100, "weight": 50}],
source="system", is_active=1)
db.add(t)
db.commit()
db.refresh(t)
return t
def _mk_reimb(db: Session, eid: int, no: str, amount: float = 100.0):
from datetime import datetime
r = ExpenseReimbursement(entity_id=eid, reimb_no=no, applicant="张三", department="财务部",
expense_type="office", title=f"{no}-报销", amount=amount,
expense_date=datetime(2026, 9, 10), status="pending",
check_result="pass")
db.add(r)
db.commit()
db.refresh(r)
return r
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 _seed_two_entities(db: Session):
_ensure_entity(db, 1, "测试企业1")
_ensure_entity(db, 2, "测试企业2")
create_test_user(db)
# ──────────────────────────────────────────────
# P0-A. cash.py 4 端点按账套（跨账套 404）
# ──────────────────────────────────────────────
def test_cash_update_plan_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
other = _mk_plan(db, 2, amount=50.0)
token1 = _login(client, 1)
r = client.put(f"{CASH}/plans/{other.id}", headers=auth_header(token1), json={"amount": 999.0})
assert r.status_code == 404, f"跨账套改计划应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
assert row is not None, "对端计划被删除"
assert row.amount == 50.0, f"跨账套改计划生效了：amount={row.amount}"
def test_cash_update_plan_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
mine = _mk_plan(db, 1, amount=50.0)
token1 = _login(client, 1)
r = client.put(f"{CASH}/plans/{mine.id}", headers=auth_header(token1), json={"amount": 88.0})
assert r.status_code == 200, r.text[:200]
db.expire_all()
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first().amount == 88.0
def test_cash_delete_plan_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
other = _mk_plan(db, 2)
token1 = _login(client, 1)
r = client.delete(f"{CASH}/plans/{other.id}", headers=auth_header(token1))
assert r.status_code == 404, f"跨账套删计划应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
assert db.query(CashPlan).filter(CashPlan.id == other.id).first() is not None, \
"跨账套删除竟生效（对端计划已消失）"
def test_cash_delete_plan_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
mine = _mk_plan(db, 1)
token1 = _login(client, 1)
r = client.delete(f"{CASH}/plans/{mine.id}", headers=auth_header(token1))
assert r.status_code == 200, r.text[:200]
db.expire_all()
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first() is None
def test_cash_complete_plan_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
other = _mk_plan(db, 2, amount=50.0)
token1 = _login(client, 1)
r = client.post(f"{CASH}/plans/{other.id}/complete", headers=auth_header(token1), json={})
assert r.status_code == 404, f"跨账套完成计划应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
assert row.status == "pending", f"跨账套完成生效了：status={row.status}"
assert (row.paid_amount or 0) == 0, f"跨账套完成写入了 paid_amount={row.paid_amount}"
def test_cash_complete_plan_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
mine = _mk_plan(db, 1, amount=50.0)
token1 = _login(client, 1)
r = client.post(f"{CASH}/plans/{mine.id}/complete", headers=auth_header(token1), json={})
assert r.status_code == 200, r.text[:300]
db.expire_all()
row = db.query(CashPlan).filter(CashPlan.id == mine.id).first()
assert row.status == "completed" and row.paid_amount == 50.0
def test_cash_register_payment_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
other = _mk_plan(db, 2, amount=50.0, plan_type="receive")
token1 = _login(client, 1)
r = client.post(f"{CASH}/receivables/{other.id}/payment", headers=auth_header(token1),
json={"amount": 5.0, "paid_date": "2026-09-20"})
assert r.status_code == 404, f"跨账套回款登记应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
assert (row.paid_amount or 0) == 0, f"跨账套回款登记生效了：paid_amount={row.paid_amount}"
assert row.status == "pending", f"跨账套回款登记改了 status={row.status}"
def test_cash_register_payment_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
mine = _mk_plan(db, 1, amount=50.0, plan_type="receive")
token1 = _login(client, 1)
r = client.post(f"{CASH}/receivables/{mine.id}/payment", headers=auth_header(token1),
json={"amount": 5.0, "paid_date": "2026-09-20"})
assert r.status_code == 200, r.text[:300]
db.expire_all()
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first().paid_amount == 5.0
# ──────────────────────────────────────────────
# P0-B. budget.py 建预算写会话账套
# ──────────────────────────────────────────────
def test_create_budget_plan_writes_session_entity(client: TestClient, db: Session):
_seed_two_entities(db)
k2 = _mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.post(f"{BUDGET}/plans", headers=auth_header(token2),
json={"kpi_id": k2.id, "period": "2026-09", "budget_value": 123.0})
assert r.status_code == 200, r.text[:300]
new_id = r.json()["id"]
db.expire_all()
row = db.query(BudgetPlan).filter(BudgetPlan.id == new_id).first()
assert row is not None, "预算未落库"
assert row.entity_id == 2, (
f"会话账套=2 新建预算落库归属 entity_id={row.entity_id}（缺陷：未写归属→模型默认 1）"
)
def test_create_budget_plan_e1_control(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
token1 = _login(client, 1)
r = client.post(f"{BUDGET}/plans", headers=auth_header(token1),
json={"kpi_id": k1.id, "period": "2026-09", "budget_value": 55.0})
assert r.status_code == 200, r.text[:300]
db.expire_all()
row = db.query(BudgetPlan).filter(BudgetPlan.id == r.json()["id"]).first()
assert row.entity_id == 1, f"e1 会话建预算 entity_id={row.entity_id}"
# ──────────────────────────────────────────────
# P1. notifications.py 字段白名单
# ──────────────────────────────────────────────
def test_update_channel_rejects_non_whitelist_fields(client: TestClient, db: Session):
_seed_two_entities(db)
ch = _mk_channel(db, "批4渠道")
token1 = _login(client, 1)
r = client.put(f"{NOTIF}/channels/{ch.id}", headers=auth_header(token1), json={
"name": "批4渠道-改名",
"channel_type": "email",
"id": 999999,
"created_at": None,
})
assert r.status_code in (200, 400), f"非白名单字段应被忽略或 400，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
row = db.query(NotificationChannel).filter(NotificationChannel.id == ch.id).first()
assert row.name == "批4渠道-改名", "白名单字段 name 未更新"
assert row.channel_type == "wecom", f"非白名单字段 channel_type 被篡改 → {row.channel_type}"
assert row.id == ch.id, "主键被篡改"
def test_update_channel_allows_business_fields(client: TestClient, db: Session):
_seed_two_entities(db)
ch = _mk_channel(db, "批4渠道2")
token1 = _login(client, 1)
r = client.put(f"{NOTIF}/channels/{ch.id}", headers=auth_header(token1), json={
"name": "批4渠道2-新", "config": {"webhook": "y"}, "enabled": False,
})
assert r.status_code == 200, r.text[:200]
db.expire_all()
row = db.query(NotificationChannel).filter(NotificationChannel.id == ch.id).first()
assert row.name == "批4渠道2-新"
assert row.enabled in (False, 0)
assert (row.config or {}).get("webhook") == "y"
# ──────────────────────────────────────────────
# P1. 跨实体只读泄漏点
# ──────────────────────────────────────────────
def test_bot_kpis_list_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_BOT, 1, bot_source=BOT_SRC)
_mk_kpi(db, E2_BOT, 2, bot_source=BOT_SRC)
token2 = _login(client, 2)
r = client.get(f"/api/cma/bot-kpis?source={BOT_SRC}", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
codes = {k["kpi_code"] for k in r.json()["kpis"]}
assert E1_BOT not in codes, f"e2 会话读到 e1 Bot KPI: {sorted(codes)}"
assert E2_BOT in codes, f"e2 会话读不到自己的 Bot KPI: {sorted(codes)}"
def test_data_quality_check_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
_mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.get(f"{DQ}/check", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
body = r.json()
assert body["total_kpis"] == 1, f"e2 会话扫描了 {body['total_kpis']} 个 KPI（应只扫本账套 1 个）"
db.expire_all()
leaked = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.kpi_id == k1.id).count()
assert leaked == 0, f"e2 会话的质量检查为 e1 的 KPI 写了 {leaked} 条日志"
def test_data_quality_stats_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_KPI, 1)
_mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.get(f"{DQ}/stats", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert r.json()["total_kpis"] == 1, f"e2 统计口径含他账套：total_kpis={r.json()['total_kpis']}"
def test_kpis_causality_chain_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
k2 = _mk_kpi(db, E2_KPI, 2)
token1 = _login(client, 1)
r = client.get(f"{KPIS}/{k2.id}/causality-chain", headers=auth_header(token1))
assert r.status_code == 404, f"跨账套读因果链应 404，实际 {r.status_code} {r.text[:200]}"
assert E2_KPI not in r.text, "跨账套响应体泄漏了对端 kpi_code"
def test_kpis_causality_chain_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
token1 = _login(client, 1)
r = client.get(f"{KPIS}/{k1.id}/causality-chain", headers=auth_header(token1))
assert r.status_code == 200, r.text[:200]
assert r.json()["kpi"]["kpi_code"] == E1_KPI
def test_kpis_associate_map_cross_entity_kpi_404(client: TestClient, db: Session):
_seed_two_entities(db)
k2 = _mk_kpi(db, E2_KPI, 2)
m1 = _mk_map(db, 1, "e1地图")
token1 = _login(client, 1)
r = client.put(f"{KPIS}/{k2.id}/associate-map", headers=auth_header(token1), json={"map_id": m1.id})
assert r.status_code == 404, f"跨账套改 KPI 归属应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
assert db.query(KPIDefinition).filter(KPIDefinition.id == k2.id).first().map_id is None, \
"跨账套 associate-map 竟改动了对端 KPI"
def test_kpis_associate_map_cross_entity_map_404(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
m2 = _mk_map(db, 2, "e2地图")
token1 = _login(client, 1)
r = client.put(f"{KPIS}/{k1.id}/associate-map", headers=auth_header(token1), json={"map_id": m2.id})
assert r.status_code == 404, f"本账套 KPI 关联对端地图应 404，实际 {r.status_code} {r.text[:200]}"
db.expire_all()
assert db.query(KPIDefinition).filter(KPIDefinition.id == k1.id).first().map_id is None
def test_kpis_associate_map_own_entity_ok(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
m1 = _mk_map(db, 1, "e1地图")
token1 = _login(client, 1)
r = client.put(f"{KPIS}/{k1.id}/associate-map", headers=auth_header(token1), json={"map_id": m1.id})
assert r.status_code == 200, r.text[:200]
db.expire_all()
assert db.query(KPIDefinition).filter(KPIDefinition.id == k1.id).first().map_id == m1.id
def test_dashboard_my_kpis_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_KPI, 1)
_mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.get(f"{DASH}/my-kpis", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert E1_KPI not in r.text, "e2 会话在 /dashboard/my-kpis 看到 e1 KPI 数据"
assert E2_KPI in r.text, "e2 会话看不到自己的 KPI"
def test_customer_dashboard_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_CUST, 1, dimension="customer")
_mk_kpi(db, E2_CUST, 2, dimension="customer")
token2 = _login(client, 2)
r = client.get(CUST, headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert E1_CUST not in r.text, "e2 会话在客户看板看到 e1 的客户维度 KPI"
assert E2_CUST in r.text, "e2 会话看不到自己的客户维度 KPI"
def test_kpi_governance_audit_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_KPI, 1)
_mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.get(f"{GOV}/governance/audit", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
body = r.json()
assert body["total"] == 1, f"e2 治理审计扫了 {body['total']} 个 KPI（应只含本账套 1 个）"
assert E1_KPI not in r.text, "e2 治理审计泄漏 e1 KPI"
def test_budget_generate_candidates_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_KPI, 1)
_mk_kpi(db, E2_KPI, 2)
token2 = _login(client, 2)
r = client.get(f"{BUDGET}/kpi-budget-candidates?year=2026", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert E1_KPI not in r.text, "e2 会话在预算候选里看到 e1 KPI"
assert E2_KPI in r.text, "e2 会话看不到自己的预算候选 KPI"
def test_budget_comparison_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
k1 = _mk_kpi(db, E1_KPI, 1)
_mk_budget(db, k1.id, 1)
token2 = _login(client, 2)
r = client.get(f"{BUDGET}/comparison?year=2026", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert r.json()["total_kpis"] == 0, (
f"e2 会话的预算对比含他账套预算：total_kpis={r.json()['total_kpis']}"
)
def test_budget_kpi_comparison_cross_entity_404(client: TestClient, db: Session):
_seed_two_entities(db)
k2 = _mk_kpi(db, E2_KPI, 2)
token1 = _login(client, 1)
r = client.get(f"{BUDGET}/comparison/kpi/{k2.id}?year=2026", headers=auth_header(token1))
assert r.status_code == 404, f"跨账套读预算对比应 404，实际 {r.status_code} {r.text[:200]}"
def test_budget_deviation_report_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_kpi(db, E1_KPI, 1)
token2 = _login(client, 2)
r = client.get(f"{BUDGET}/deviation-report?period=2026-09", headers=auth_header(token2))
assert r.status_code == 200, r.text[:200]
assert E1_KPI not in r.text, "e2 会话在预算差异报告看到 e1 KPI"
assert r.json()["summary"]["total_kpis"] == 0, (
f"e2 差异报告含他账套 KPI：total_kpis={r.json()['summary']['total_kpis']}"
)
def test_driver_budget_write_and_history_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
token2 = _login(client, 2)
payload = {
"name": "批4驱动预算",
"formula_text": "a * b",
"factor_defs": [{"key": "a", "label": "A", "default": 2}, {"key": "b", "label": "B", "default": 3}],
"factors": {},
}
r = client.post(f"{BUDGET}/driver/calculate", headers=auth_header(token2), json=payload)
assert r.status_code == 200, r.text[:300]
rec_id = r.json()["id"]
db.expire_all()
rec = db.query(DriverFactorBudget).filter(DriverFactorBudget.id == rec_id).first()
assert rec is not None and rec.entity_id == 2, (
f"e2 会话的驱动因子预算落库归属 entity_id={getattr(rec, 'entity_id', None)}（应为 2）"
)
# 历史列表只含本账套
r1 = client.get(f"{BUDGET}/driver/history", headers=auth_header(_login(client, 1)))
assert r1.status_code == 200, r1.text[:200]
assert "批4驱动预算" not in r1.text, "e1 会话在驱动因子历史里看到 e2 记录"
r2 = client.get(f"{BUDGET}/driver/history", headers=auth_header(token2))
assert "批4驱动预算" in r2.text, "e2 会话看不到自己的驱动因子历史"
def test_expenses_write_and_stats_scoped(client: TestClient, db: Session):
_seed_two_entities(db)
_mk_reimb(db, 1, "ZZ_B4_E1_BX001", amount=111.0)
token2 = _login(client, 2)
r = client.post(f"{EXPENSES}/reimbursements", headers=auth_header(token2), json={
"applicant": "李四", "department": "财务部", "expense_type": "office",
"title": "批4 e2 报销", "amount": 222.0, "expense_date": "2026-09-15",
})
assert r.status_code == 200, r.text[:300]
new_id = r.json()["data"]["id"]
db.expire_all()
row = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.id == new_id).first()
assert row.entity_id == 2, f"e2 会话报销单落库归属 entity_id={row.entity_id}（应为 2）"
r2 = client.get(f"{EXPENSES}/stats", headers=auth_header(token2))
assert r2.status_code == 200, r2.text[:200]
counts = r2.json()["counts"]
assert counts["total"] == 1, (
f"e2 费用看板统计含他账套：total={counts['total']}（应只统计本账套 1 条）"
)
# ──────────────────────────────────────────────
# P2. okr_templates 应用模板写会话账套
# ──────────────────────────────────────────────
def test_apply_okr_template_writes_session_entity(client: TestClient, db: Session):
_seed_two_entities(db)
t = _mk_template(db, "批4模板")
token2 = _login(client, 2)
r = client.post(f"{OKRT}/{t.id}/apply", headers=auth_header(token2), json={})
assert r.status_code == 200, r.text[:300]
map_id = r.json().get("map_id") or (r.json().get("data") or {}).get("id")
db.expire_all()
m = None
if map_id:
m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
if m is None:
m = db.query(StrategicMap).order_by(StrategicMap.id.desc()).first()
assert m is not None, f"应用模板未创建地图：{r.text[:300]}"
assert m.entity_id == 2, (
f"会话账套=2 应用 OKR 模板生成的地图归属 entity_id={m.entity_id}（缺陷：未传 entity_id→默认 1）"
)