diff --git a/backend/app/api/alignment.py b/backend/app/api/alignment.py index 6d660633..63f7b495 100644 --- a/backend/app/api/alignment.py +++ b/backend/app/api/alignment.py @@ -20,7 +20,30 @@ router = APIRouter(prefix="/api/cma/alignment", tags=["KPI目标对齐"], dependencies=[Depends(require_auth_401)], ) -# 三种对齐模式定义 +# ── B3(a) 对齐模式拆账套（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922）── +# 背景：`role_permissions.key` 为 unique 单行键 → `alignment_config` 全库一行， +# 任一账套的 ceo/it 改对齐模式 → **全账套生效**（跨账套配置污染）。 +# 治本（方案 A，零 DDL）：键名带账套后缀 → `alignment_config.`。 +# · 读：只读本账套键；未配置 → configured=false（/tree 提示先初始化）， +# **不回退旧全局键**（回退即恢复跨账套生效）。 +# · 写：只写本账套键（entity 来自会话 token）。 +# · 旧值搬运：脚本 scripts/migrate_cash_balance_key_entity_20260922.py --check/--apply +# （同一 500 口径：原全局值落到「当前生产账套」名下；--down 删新键并写回旧键）。 +ALIGNMENT_KEY_PREFIX = "alignment_config" + + +def alignment_key(entity_id: int) -> str: + """账套维度的对齐模式键名：alignment_config.""" + return f"{ALIGNMENT_KEY_PREFIX}.{int(entity_id)}" + + +def _get_alignment_perm(db: Session, entity_id: int): + """读本账套的对齐模式配置行（不存在 → None）""" + return db.query(RolePermission).filter( + RolePermission.key == alignment_key(entity_id) + ).first() + + ALIGNMENT_MODES = [ { "key": "vertical_decomposition", @@ -50,18 +73,25 @@ def list_modes(): @router.get("/config") -def get_alignment_config(db: Session = Depends(get_db)): - """获取当前系统对齐模式配置（公开接口，无需认证）""" - perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first() +def get_alignment_config(entity_id: int = Depends(get_entity_id), + db: Session = Depends(get_db)): + """获取**当前账套**的对齐模式配置（B3(a) 拆账套，2026-09-22） + + 修复前：读全局唯一键 `alignment_config` → 任一账套改模式全平台生效。 + 修复后：只读 `alignment_config.`；未配置 → configured=false。 + """ + perm = _get_alignment_perm(db, entity_id) if not perm: return { "mode": None, "configured": False, + "entity_id": entity_id, "modes": ALIGNMENT_MODES, } return { "mode": perm.value, "configured": True, + "entity_id": entity_id, "modes": ALIGNMENT_MODES, } @@ -69,23 +99,31 @@ def get_alignment_config(db: Session = Depends(get_db)): @router.post("/config") def set_alignment_config( data: dict, + entity_id: int = Depends(get_entity_id), db: Session = Depends(get_db), user = Depends(require_role("ceo", "it")), ): - """初始化/修改系统对齐模式（CEO/IT权限）""" + """初始化/修改**当前账套**的对齐模式（CEO/IT权限） + + B3(a)（2026-09-22）：只写 `alignment_config.`，不覆盖他账套； + entity 来自会话 token，显式传他账套 entity_id → get_entity_id 403。 + """ mode_key = data.get("mode") if mode_key not in [m["key"] for m in ALIGNMENT_MODES]: raise HTTPException(400, f"无效的对齐模式: {mode_key}") - perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first() + perm = _get_alignment_perm(db, entity_id) if perm: - perm.value = {"mode": mode_key, "set_at": datetime.now().isoformat()} + perm.value = {"mode": mode_key, "set_at": datetime.now().isoformat(), + "entity_id": entity_id} else: - perm = RolePermission(key="alignment_config", value={"mode": mode_key, "set_at": datetime.now().isoformat()}) + perm = RolePermission(key=alignment_key(entity_id), + value={"mode": mode_key, "set_at": datetime.now().isoformat(), + "entity_id": entity_id}) db.add(perm) db.commit() - return {"message": f"对齐模式已设置为: {mode_key}", "mode": mode_key} + return {"message": f"对齐模式已设置为: {mode_key}", "mode": mode_key, "entity_id": entity_id} @router.get("/tree") @@ -106,8 +144,8 @@ def get_alignment_tree( growth-quality / predict.accuracy）。三种模式共用本列表 → 三模式同时隔离。 跨实体显式传参（?entity_id=<他人账套>）由 get_entity_id 返回 403。 """ - # 获取当前模式 - config_perm = db.query(RolePermission).filter(RolePermission.key == "alignment_config").first() + # 获取当前账套的对齐模式（B3(a) 拆账套 2026-09-22：只读 alignment_config.） + config_perm = _get_alignment_perm(db, entity_id) mode = config_perm.value.get("mode") if config_perm else None if not mode: raise HTTPException(400, "系统未配置对齐模式，请先在系统设置中初始化") diff --git a/backend/app/api/bot_bridge_v2.py b/backend/app/api/bot_bridge_v2.py index 1770a96f..36419857 100644 --- a/backend/app/api/bot_bridge_v2.py +++ b/backend/app/api/bot_bridge_v2.py @@ -21,7 +21,7 @@ from app.database import get_db from app.models import ( MpmResult, BotBridgeConfig, KPIDefinition, KPIValue, KPIAlert, - ActionPlan, Entity, + ActionPlan, Entity, OperationLog, ) from app.risk_levels import risk_level # 裁定 R1/R5（2026-09-19 财务口径）：demo 账套任何写通道不产生业务预警；期间非规整月不入库（单一事实源） @@ -364,6 +364,7 @@ def receive_mpm_result( # ── 3. 触发预警 ── alerts = [] + alerts_skipped_no_kpi = [] # B3(b): KPI 定义缺失而跳过的预警（留痕用） for field, config in ALERT_THRESHOLDS.items(): field_value = results.get(field) if field_value is None: @@ -383,6 +384,12 @@ def receive_mpm_result( logger.info("R5 期间守卫：跳过不入库（不可归期，非规整月 YYYY-MM）period=%r", period) continue # 查找关联KPI + # B3(b)（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922）： + # 修复前 `kpi_id=kpi_id or 1` —— KPI 定义缺失时把预警错挂到「全库第 1 个 KPI」 + # （该 KPI 可能属他账套：预警 entity_id 正确、KPI 维度却跨了账套）。 + # 现改为**显式校验**：拿不到本账套的 KPI 定义 → 不建预警（跳过）+ 落 OperationLog 留痕。 + # 为什么不是 `kpi_id=None`：`kpi_alerts.kpi_id` 为 NOT NULL（models/__init__.py:145）， + # batch5 §B3(b) 方案 A（保留可见性）在本 schema 下不可行；方案 B（跳过 + 审计）落地。 kpi_id = None if kpi_code: kpi_def = db.query(KPIDefinition).filter( @@ -392,11 +399,38 @@ def receive_mpm_result( if kpi_def: kpi_id = kpi_def.id - # 创建预警 alert_msg = f"⚠️ {kpi_name}异常: {field_value:.1f}% (阈值: {threshold}%) [entity={entity.short_name or entity.name}, period={period}]" + + if kpi_id is None: + # 宁缺勿错：不入库，但必须留痕（防「预警静默丢失」） + alerts_skipped_no_kpi.append(f"{kpi_name}(kpi_code={kpi_code!r})") + db.add(OperationLog( + user_id=None, + action="skip_alert", + target_type="alert", + target_id=None, + detail={ + "reason": "kpi_definition_not_found_for_entity", + "kpi_code": kpi_code, + "kpi_name": kpi_name, + "entity_id": entity_id, + "period": period, + "field": field, + "field_value": field_value, + "threshold": threshold, + "source": "bot_bridge_v2.mpm-result", + }, + )) + logger.warning( + "B3(b) 跳过建预警（KPI定义缺失，不再回落 kpi_id=1）: kpi_code=%r entity_id=%s period=%s", + kpi_code, entity_id, period, + ) + continue + + # 创建预警 alert = KPIAlert( entity_id=entity_id, # 2026-09-13 F1: 显式写归属 - kpi_id=kpi_id or 1, # fallback to first KPI if not found + kpi_id=kpi_id, # B3(b): 显式校验后的本账套 KPI（禁止 or 1 兜底） alert_level="red", alert_message=alert_msg, alert_type="actual", @@ -413,10 +447,13 @@ def receive_mpm_result( "success": True, "kpi_updated": len(kpi_updates), "alerts_triggered": len(alerts), + # B3(b): 因本账套无对应 KPI 定义而跳过的预警（已落 OperationLog，非静默丢失） + "alerts_skipped_missing_kpi": len(alerts_skipped_no_kpi), "mpm_record_id": record.id, "details": { "kpi_codes": kpi_updates, "alerts": alerts, + "alerts_skipped_missing_kpi": alerts_skipped_no_kpi, }, "skipped_non_month_period": (not period_attributable), } diff --git a/backend/app/api/cash.py b/backend/app/api/cash.py index 24406e96..fa9a082c 100644 --- a/backend/app/api/cash.py +++ b/backend/app/api/cash.py @@ -116,15 +116,21 @@ def api_get_balance( @router.post("/balance") def api_set_balance( data: dict, + entity_id: int = Depends(get_entity_id), db: Session = Depends(get_db), ): - """设置当前现金余额（万元），作为预测基线""" + """设置当前现金余额（万元），作为预测基线 + + 账套隔离（B1，2026-09-22）：基线落 `cash.current_balance.`， + entity 来自会话 token（显式传他账套 entity_id → get_entity_id 403）。 + 修复前本端点无 entity 依赖 → 任一账套设基线会覆盖全平台唯一键。 + """ from app.utils.cash_forecast_engine import set_current_cash_balance value = float(data.get("current_cash", 0)) if value < 0: raise HTTPException(400, "现金余额不能为负") - set_current_cash_balance(db, value) - return {"message": "当前现金余额已更新", "current_cash": value} + set_current_cash_balance(db, value, entity_id) + return {"message": "当前现金余额已更新", "entity_id": entity_id, "current_cash": value} # ══════════════════════════════════════════════════════════ @@ -785,7 +791,11 @@ def _is_carry_forward(summary: str, subject_name: str) -> bool: def _sync_cash_balance_from_ledger(db: Session, entity_id: int, batch: str) -> float: - """货币资金类科目(1001/1002开头)期末余额 → set_current_cash_balance（万元）""" + """货币资金类科目(1001/1002开头)期末余额 → set_current_cash_balance（万元） + + 账套隔离（B1，2026-09-22）：本函数已按 entity_id 计算余额，写回时**带上该账套** + （修复前丢账套 → 任一家导入网银流水就覆盖全平台唯一键的基线）。 + """ from app.models import VoucherDetail from app.utils.cash_forecast_engine import set_current_cash_balance from app.utils.voucher_guard import apply_voucher_guard @@ -800,8 +810,11 @@ def _sync_cash_balance_from_ledger(db: Session, entity_id: int, batch: str) -> f and (r.subject_code.startswith("1001") or r.subject_code.startswith("1002")) ), 2) cash_wan = round(balance_yuan / 10000, 4) - set_current_cash_balance(db, cash_wan) - logger.info(f"网银流水导入[{batch}] 货币资金期末余额{balance_yuan}元 = {cash_wan}万元 → 现金余额联动") + set_current_cash_balance(db, cash_wan, entity_id) + logger.info( + f"网银流水导入[{batch}] 货币资金期末余额{balance_yuan}元 = {cash_wan}万元 " + f"→ 现金余额联动（账套 {entity_id}）" + ) return cash_wan diff --git a/backend/app/api/notifications.py b/backend/app/api/notifications.py index ea85b385..82145a71 100644 --- a/backend/app/api/notifications.py +++ b/backend/app/api/notifications.py @@ -2,10 +2,21 @@ 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.models import NotificationChannel, NotificationLog from app.auth_middleware import require_role, require_auth_401 from datetime import datetime +# ── B2（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922）──────────────── +# 本模块两张表在 `models/__init__.py` 均**无 entity_id 列**，按「分表处置」口径落地： +# · `notification_logs`（Log 类）→ **经父键 join 归属**，零迁移： +# alert_id → kpi_alerts.id → kpi_alerts.entity_id（GET /logs 已收口）。 +# · `notification_channels`（**配置类，无父键可桥接**）→ 任总 2026-09-22 已批 +# 「方案 1（零 DDL）」：**保持全局 = 集团共享配置**（一套 wecom/mail/sms 渠道 +# 供全账套复用；预警推送是集团行为，渠道与账套无关）。 +# ⚠️ 因此 `/channels*` 四个端点**有意不按账套过滤**——这是口径，不是漏隔离。 +# 若后续需要「账套私有渠道」，须单独立单加 entity_id 列（DDL + 回填 + 唯一键改造）。 + # 用户端点：统一要求会话 Bearer 鉴权（cma-api-auth-surface-audit-20260914） # 缺口：本 router 6 个端点全部无鉴权——无 token 可读写通知渠道（config 内含 webhook # 地址/令牌等凭据）、可触发渠道测试推送、可读推送日志。收口为 require_auth_401。 @@ -97,14 +108,28 @@ def test_channel(ch_id: int, db: Session = Depends(get_db)): @router.get("/logs") -def list_logs(page: int = 1, db: Session = Depends(get_db)): - """通知历史""" - total = db.query(NotificationLog).count() - logs = db.query(NotificationLog).order_by( +def list_logs(page: int = 1, + entity_id: int = Depends(get_entity_id), + db: Session = Depends(get_db)): + """通知历史（**按会话账套收口**，B2 零迁移） + + 2026-09-22 dama-workbench-entity-isolation-fullstack-20260922（§四 B2）： + `notification_logs` 无 entity_id 列 → 经父键 join 归属： + `notification_logs.alert_id → kpi_alerts.id → kpi_alerts.entity_id`。 + 口径与 batch5 对 `KpiDataQualityLog` 的处置一致（宁少勿错）： + `alert_id IS NULL` 的行无可归属父键 → **不返回**（不暴露给任一账套）。 + """ + from app.models import KPIAlert + scoped = (db.query(NotificationLog) + .join(KPIAlert, NotificationLog.alert_id == KPIAlert.id) + .filter(KPIAlert.entity_id == entity_id)) + total = scoped.count() + logs = scoped.order_by( NotificationLog.created_at.desc() ).offset((page - 1) * 20).limit(20).all() return { "total": total, + "entity_id": entity_id, "data": [{ "id": l.id, "alert_id": l.alert_id, diff --git a/backend/app/api/okr_templates.py b/backend/app/api/okr_templates.py index a7ac10dc..60bd894a 100644 --- a/backend/app/api/okr_templates.py +++ b/backend/app/api/okr_templates.py @@ -1,5 +1,18 @@ """ OKR模板库 API — 按维度筛选 + 用户自定义模板 + +⚠️ 集团级共享口径（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922，§四 B2） +------------------------------------------------------------------------------------ +`okr_templates` 表**无 entity_id 列**，且属**模板/配置类**（无父键可 join 桥接）。 +任总 2026-09-22「都批」已按 **方案 1（零 DDL）** 拍板： + + **保持全局 = 集团级共享配置** —— 模板库是集团知识资产，一套模板（维度/行业标签/ + 目标+KR 骨架）供全账套复用；账套差异体现在**使用模板生成的具体目标/KR 行** + （`objectives` / `map_objectives` 均带 entity_id，已隔离），而不是模板本身。 + +因此本模块的列表/详情/新增/改/删**有意不按账套过滤**——这是口径，不是漏隔离。 +若后续确需「账套私有模板」，须单独立单加 `entity_id` 列（DDL + 回填 + 唯一键 +`name+dimension` 改造），本单不预设。 """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session diff --git a/backend/app/api/tenant.py b/backend/app/api/tenant.py index 65319c06..39b56770 100644 --- a/backend/app/api/tenant.py +++ b/backend/app/api/tenant.py @@ -1,10 +1,29 @@ -"""租户状态API — CMA系统切换公司时记录当前tenant，供项目Bot分发A2A任务""" +"""租户状态API — CMA系统切换公司时记录当前tenant，供项目Bot分发A2A任务 + +⚠️ 语义收敛（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922，任总已批「都批」） +-------------------------------------------------------------------------------------- +`system_configs.current_tenant` 是**全局唯一键**（该表无 entity_id 列，`config_key` unique） +⇒ 一次写入全平台生效。本次事故（2026-09-04 前端企业切换器把它写成 entity_id=2）暴露的 +结构性缺陷正是「全局单值键被当成个人当前视图 / 数据维度」。 + +本文件的收敛口径： + 1. **它不是数据维度**。工作台/看板/报表取数一律用**登录会话的 entity** + （`app/deps.py::get_entity_id` ← JWT token 绑定），前端本地视图用 + `localStorage.cma_entity_id`（登录窗口写入）。禁止任何取数端点读本键。 + 2. 它的用途**只剩**：项目 Bot 分发 A2A 任务时的状态记录（本模块原始设计目的）。 + 3. **仅 admin 会话可读**（GET /current）：系统无 `admin` 角色（ROLES: ceo/finance/ + business/it），故映射为持有 `admin` 动作权限的 `it` + 业务负责人 `ceo`。 + 4. 写（POST /switch）**不额外收紧角色**：项目 Bot 分发任务时需要写该状态， + 收紧会切断其原始用途。 + 5. 每次写都落 `operation_logs`（action=`switch_tenant_state`）留痕 —— 修复前本端点 + 无审计（ops 止血单假定「走端点即有审计」不成立，已在本单补齐）。 +""" from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.orm import Session from app.database import get_db -from app.auth_middleware import require_auth_401 -from app.models import SystemConfig, Entity +from app.auth_middleware import require_auth_401, require_role +from app.models import SystemConfig, Entity, OperationLog # 用户端点：统一要求会话 Bearer 鉴权（cma-api-auth-surface-audit-20260914） # 缺口：本 router 原无鉴权——无 token 可读当前租户、并可 POST /switch 改写 @@ -25,8 +44,15 @@ def _tenant_name(entity_id: int) -> str: @router.get("/current") -def get_current_tenant(db: Session = Depends(get_db)): - """查询当前租户""" +def get_current_tenant( + db: Session = Depends(get_db), + user=Depends(require_role("it", "ceo")), +): + """查询当前租户状态记录（**仅 admin 会话**：it/ceo） + + 2026-09-22 收敛：本键**不是**数据维度——前端/接口取数一律按会话 entity， + 不得读本键决定「取哪个实体的数据」（本次工作台跨实体污染的直接触发点）。 + """ cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() if cfg and cfg.config_value: import json @@ -39,12 +65,23 @@ def get_current_tenant(db: Session = Depends(get_db)): @router.post("/switch") def switch_tenant(data: TenantSwitchRequest, db: Session = Depends(get_db)): - """切换当前租户（CMA前端企业切换器调用）""" + """切换当前租户状态记录（项目Bot 分发 A2A 任务用；CMA前端企业切换器已下线） + + 2026-09-22：写操作**不动**（保留 Bot 通道可用性），补 `operation_logs` 审计留痕。 + """ ent = db.query(Entity).filter(Entity.id == data.entity_id).first() if not ent: raise HTTPException(404, "企业不存在") import json + previous = None + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() + if cfg and cfg.config_value: + try: + previous = json.loads(cfg.config_value) + except Exception: + previous = None + state = { "tenant_id": _tenant_name(data.entity_id), "entity_id": data.entity_id, @@ -54,15 +91,29 @@ def switch_tenant(data: TenantSwitchRequest, db: Session = Depends(get_db)): "switched_at": __import__("datetime").datetime.now().isoformat(), } - cfg = db.query(SystemConfig).filter(SystemConfig.config_key == TENANT_KEY).first() if cfg: cfg.config_value = json.dumps(state, ensure_ascii=False) else: db.add(SystemConfig( config_key=TENANT_KEY, config_value=json.dumps(state, ensure_ascii=False), - description="当前租户状态（CMA企业切换联动）", + description="当前租户状态记录（仅供项目Bot分发A2A任务；非数据维度，2026-09-22 语义收敛）", )) + + # 审计留痕（2026-09-22 补齐：本端点原无 operation_logs 记录） + db.add(OperationLog( + user_id=None, + action="switch_tenant_state", + target_type="system_config", + target_id=None, + detail={ + "entity_id": data.entity_id, + "entity_name": ent.name, + "source": data.source, + "previous": previous, + "note": "current_tenant 状态记录变更（非数据维度）", + }, + )) db.commit() return {"success": True, "current": state} diff --git a/backend/app/utils/cash_forecast_engine.py b/backend/app/utils/cash_forecast_engine.py index 4d39cb1c..d8dddbb5 100644 --- a/backend/app/utils/cash_forecast_engine.py +++ b/backend/app/utils/cash_forecast_engine.py @@ -221,8 +221,33 @@ def calc_trend(values: list) -> float: return slope / max(abs(avg_y), 1.0) * 100 # 趋势百分比 +# ── B1 现金基线拆账套（2026-09-22 dama-workbench-entity-isolation-fullstack-20260922）── +# 背景：`SystemConfig.config_key` 为 unique 单行键 → `cash.current_balance` 全库仅一行， +# 所有账套共用；任一账套导入网银流水（`cash.py::_sync_cash_balance_from_ledger` 按 entity +# 算完余额后丢账套写回）即覆盖其他账套的预测基线。实测事故值 -9.3 被全账套共用。 +# +# 治本（方案 A，零 DDL）：键名带账套后缀 → `cash.current_balance.`。 +# · 读：只读本账套键；本账套未设值 → None（由 KPI 实际值兜底，**不跨账套借用**）。 +# 不做「回退旧全局键」——那正是本次污染通路（其余账套会读到别家的基线）。 +# · 写：只写本账套键（调用方必须传会话 entity）。 +# · 旧值搬运：一次性落到「当前生产账套」名下（脚本 +# `scripts/migrate_cash_balance_key_entity_20260922.py`，--check/--apply/--down）。 +# 回退路径：--down 删新键（旧全局键原样保留，回退老码即可继续读）。 +CASH_BALANCE_KEY_PREFIX = "cash.current_balance" + + +def cash_balance_key(entity_id: int) -> str: + """账套维度的现金基线键名：cash.current_balance.""" + return f"{CASH_BALANCE_KEY_PREFIX}.{int(entity_id)}" + + def get_current_cash_balance(db: Session, entity_id: int) -> Optional[float]: - """获取当前现金余额（万元）— 优先级：KPI实际值 > SystemConfig > None""" + """获取当前现金余额（万元）— 优先级：KPI实际值 > 本账套SystemConfig键 > None + + 账套隔离（B1，2026-09-22）：只读 `cash.current_balance.`。 + 本账套未设基线 → 返回 None（由 KPI 实际值兜底 / 预测引擎默认值）， + **不回退旧全局键 `cash.current_balance`**——回退即恢复跨账套借用。 + """ from app.models import SystemConfig, KPIValue # C3(2026-09-19 dama-20260919)：补传实体前缀后缀候选（与 F2 的应收/应付路径一致） # 修复前漏传 → 博海/演示企业一条候选都不命中 → 回退 SystemConfig.cash.current_balance(全库唯一键 -9.3) @@ -236,7 +261,7 @@ def get_current_cash_balance(db: Session, entity_id: int) -> Optional[float]: if latest_cash and latest_cash.actual_value is not None: return float(latest_cash.actual_value) cfg = db.query(SystemConfig).filter( - SystemConfig.config_key == "cash.current_balance" + SystemConfig.config_key == cash_balance_key(entity_id) ).first() if cfg and cfg.config_value: try: @@ -246,19 +271,27 @@ def get_current_cash_balance(db: Session, entity_id: int) -> Optional[float]: return None -def set_current_cash_balance(db: Session, value: float) -> float: - """设置当前现金余额（写入SystemConfig，供预测引擎使用）""" +def set_current_cash_balance(db: Session, value: float, entity_id: int) -> float: + """设置当前现金余额（写入本账套键 `cash.current_balance.`，供预测引擎使用） + + entity_id 为**必传**（2026-09-22 B1）：签名中不设默认值，强制调用方带上会话账套， + 防「漏传 → 落默认账套」重演本次污染。全部调用点见 cash.py（POST /balance、 + 网银流水导入联动）。 + """ from app.models import SystemConfig + if entity_id is None: + raise ValueError("set_current_cash_balance 必须传 entity_id（账套隔离，2026-09-22 B1）") + key = cash_balance_key(entity_id) cfg = db.query(SystemConfig).filter( - SystemConfig.config_key == "cash.current_balance" + SystemConfig.config_key == key ).first() if cfg: cfg.config_value = str(value) else: cfg = SystemConfig( - config_key="cash.current_balance", + config_key=key, config_value=str(value), - description="当前现金余额（万元），资金预测基线", + description=f"当前现金余额（万元），资金预测基线 — 账套 {int(entity_id)}", ) db.add(cfg) db.commit() diff --git a/backend/scripts/migrate_cash_balance_key_entity_20260922.py b/backend/scripts/migrate_cash_balance_key_entity_20260922.py new file mode 100644 index 00000000..ee708f39 --- /dev/null +++ b/backend/scripts/migrate_cash_balance_key_entity_20260922.py @@ -0,0 +1,329 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""全局键拆账套 — 一次性搬运（数据迁移，非 DDL） + +任务：`dama-workbench-entity-isolation-fullstack-20260922` §三 B1/B3(a) +（DAMA 诊断 / 项目Bot batch5 §B 方案段 / 任总 2026-09-22「都批」） + +背景 +---- +`system_configs.config_key` 与 `role_permissions.key` 均为 **unique 单行键** ⇒ +下列两个「全局键」全库只有一行，被所有账套共用（本次「酣客工作台显示博海数据」 +污染的同源结构性缺陷）： + + 键 处置 + ------------------------------- ------------------------------------------ + system_configs.cash.current_balance → cash.current_balance. + role_permissions.alignment_config → alignment_config. + +代码侧（读写）已改为**只认带账套后缀的新键**（方案 A，零 DDL）： + backend/app/utils/cash_forecast_engine.py （get_/set_current_cash_balance） + backend/app/api/cash.py （POST /balance、网银流水联动） + backend/app/api/alignment.py （GET/POST /config、GET /tree） + +本脚本 = **旧值一次性搬运**（把原来唯一的那一行值落到「当前生产账套」名下）。 +「当前生产账套」判定：默认 `--target-entity 1`（entity 1 = 陕西酣客，生产账套）， +`--check` 会打印**逐账套凭证推导的货币资金期末余额**与旧键值的比对，供人工确认 +（哪个账套的 1001/1002 期末余额 == 旧键值 ⇒ 该值本来就是它写的）。 + +口径（须与代码注释一致） +------------------------ +· 新键值 = 旧键值（一次性搬运）**只落到 target-entity**；**其余账套留空** + ⇒ 留空语义 = 该账套现金基线走 KPI 实际值兜底 / 预测引擎默认值， + **不跨账套借用**（正是本次污染的成因）。 +· 旧键**原样保留不删**（回退杠杆：回退老代码即可继续读旧键）。 +· alignment_config 同理：旧全局值落到 target-entity；其余账套 configured=false。 +· 幂等：新键已存在 → 不覆盖（除非 `--force`）；二跑 `--apply` 零副作用。 + +用法 +---- + # 只读预检（默认动作，零写入） + python3 scripts/migrate_cash_balance_key_entity_20260922.py --check + + # 写库（**需审批 / 生产写**；--apply 才写） + python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply + python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply --target-entity 2 + python3 scripts/migrate_cash_balance_key_entity_20260922.py --apply --force # 覆盖已存在的新键 + + # 回退（删新键 + 把旧全局键写回原值） + python3 scripts/migrate_cash_balance_key_entity_20260922.py --down + +选项 +---- + --target-entity N 旧值落哪个账套（默认 1；--check 会给出证据建议） + --force 新键已存在时允许覆盖（默认拒绝，保留既有值） + --json 机器可读输出 + --allow-db NAME 生产库名保护豁免（默认只允许库名 `cma`） + +退出码 +------ + 0 成功 ｜ 1 执行失败 ｜ 2 用法/保护拒绝（库名不符、目标账套不存在） + +⚠️ 本脚本写的是**生产业务库**：不得在开发级会话直接 `--apply`。 + 依据《生产资产派单纪律》（skills-shared/operations/prod-asset-dispatch-discipline）： + 碰生产资产一律走派单 + 审批；`--check` 同样是生产读，须留单痕。 +""" +import argparse +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sqlalchemy import text # noqa: E402 + +CASH_KEY = "cash.current_balance" # 旧全局键（保留不删） +ALIGN_KEY = "alignment_config" # 旧全局键（保留不删） +PROD_DB_NAME = "cma" # 生产库名（保护口径） + +CASH_DESC = "当前现金余额（万元），资金预测基线 — 账套 {eid}" +ALIGN_DESC = "对齐模式配置 — 账套 {eid}（B3(a) 拆账套 2026-09-22）" + + +def _engine(): + from app.database import get_engine + return get_engine() + + +def _scalar(conn, sql, **params): + row = conn.execute(text(sql), params).fetchone() + return row[0] if row else None + + +def _detect_db_name(conn) -> str: + try: + return _scalar(conn, "SELECT DATABASE()") or "" + except Exception: + return "" + + +def _cash_ledger_by_entity(conn): + """逐账套：货币资金(1001/1002, carry_forward=0) 期末净额（元 → 万元） + + 与 `cash.py::_sync_cash_balance_from_ledger` 同口径（该函数正是写旧键的调用方） + ——用于回答「-9.3 这个值原本是哪家写的」。 + """ + out = {} + try: + rows = conn.execute(text( + "SELECT entity_id, " + " ROUND(SUM(COALESCE(debit_amount,0) - COALESCE(credit_amount,0))/10000, 4) AS wan, " + " COUNT(*) AS n " + "FROM voucher_details " + "WHERE carry_forward = 0 AND subject_code IS NOT NULL " + " AND (subject_code LIKE '1001%' OR subject_code LIKE '1002%') " + "GROUP BY entity_id ORDER BY entity_id" + )).fetchall() + for r in rows: + out[int(r[0])] = {"cash_wan": float(r[1] or 0), "voucher_rows": int(r[2] or 0)} + except Exception as e: # 表缺失/权限 → 只报告，不阻断 + out["_error"] = str(e)[:300] + return out + + +def _snapshot(conn): + """只读快照：新旧键现状 + 逐账套凭证推导余额""" + snap = { + "db": _detect_db_name(conn), + "legacy": { + "cash.current_balance": None, + "alignment_config": None, + }, + "new_keys": {"cash": {}, "alignment": {}}, + "entities": [], + "cash_ledger_by_entity": _cash_ledger_by_entity(conn), + } + + row = conn.execute(text( + "SELECT config_value, description FROM system_configs WHERE config_key = :k" + ), {"k": CASH_KEY}).fetchone() + if row: + snap["legacy"]["cash.current_balance"] = {"config_value": row[0], "description": row[1]} + + row = conn.execute(text( + "SELECT value FROM role_permissions WHERE `key` = :k" + ), {"k": ALIGN_KEY}).fetchone() + if row: + snap["legacy"]["alignment_config"] = {"value": row[0]} + + for r in conn.execute(text( + "SELECT id, name, short_name, status FROM entities ORDER BY id" + )).fetchall(): + snap["entities"].append({"id": int(r[0]), "name": r[1], "short_name": r[2], "status": r[3]}) + + for r in conn.execute(text( + "SELECT config_key, config_value FROM system_configs " + "WHERE config_key LIKE :p ORDER BY config_key" + ), {"p": CASH_KEY + ".%"}).fetchall(): + snap["new_keys"]["cash"][r[0]] = r[1] + + for r in conn.execute(text( + "SELECT `key`, value FROM role_permissions WHERE `key` LIKE :p ORDER BY `key`" + ), {"p": ALIGN_KEY + ".%"}).fetchall(): + snap["new_keys"]["alignment"][r[0]] = r[1] + + # 建议目标账套（只读推断）：旧值 == 某账套凭证推导余额 + suggestion = None + legacy_cash = (snap["legacy"]["cash.current_balance"] or {}).get("config_value") + if legacy_cash is not None: + try: + want = float(legacy_cash) + for eid, info in snap["cash_ledger_by_entity"].items(): + if isinstance(eid, int) and abs(float(info["cash_wan"]) - want) < 1e-6: + suggestion = eid + break + except Exception: + suggestion = None + snap["suggested_target_entity"] = suggestion + return snap + + +def cmd_check(args): + eng = _engine() + with eng.connect() as conn: + snap = _snapshot(conn) + if args.json: + print(json.dumps(snap, ensure_ascii=False, indent=2)) + return 0 + print(f"目标库: {snap['db']}") + print(f"旧全局键 cash.current_balance : {snap['legacy']['cash.current_balance']}") + print(f"旧全局键 alignment_config : {snap['legacy']['alignment_config']}") + print(f"已存在新键(cash) : {snap['new_keys']['cash']}") + print(f"已存在新键(alignment) : {snap['new_keys']['alignment']}") + print("逐账套凭证推导货币资金期末余额（万元）:") + for eid, info in snap["cash_ledger_by_entity"].items(): + print(f" entity {eid}: {info}") + print(f"建议 --target-entity : {snap['suggested_target_entity']}" + f"（旧值与他家凭证推导余额一致时给出；None = 无匹配，按默认 1 或人工裁定）") + print("【只读预检，零写入】") + return 0 + + +def _apply_cash(conn, target: int, force: bool, dry: bool): + new_key = f"{CASH_KEY}.{target}" + row = conn.execute(text( + "SELECT config_value FROM system_configs WHERE config_key = :k" + ), {"k": CASH_KEY}).fetchone() + if not row: + return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"} + legacy_value = row[0] + + exist = conn.execute(text( + "SELECT config_value FROM system_configs WHERE config_key = :k" + ), {"k": new_key}).fetchone() + if exist and not force: + return {"key": new_key, "action": "noop", + "reason": "新键已存在且未 --force（保留既有值，幂等）", + "existing_value": exist[0]} + if dry: + return {"key": new_key, "action": "would-insert-or-update", + "value": legacy_value, "existing_value": exist[0] if exist else None} + + if exist: + conn.execute(text( + "UPDATE system_configs SET config_value = :v, description = :d WHERE config_key = :k" + ), {"v": legacy_value, "d": CASH_DESC.format(eid=target), "k": new_key}) + action = "updated" + else: + conn.execute(text( + "INSERT INTO system_configs (config_key, config_value, description) " + "VALUES (:k, :v, :d)" + ), {"k": new_key, "v": legacy_value, "d": CASH_DESC.format(eid=target)}) + action = "inserted" + return {"key": new_key, "action": action, "value": legacy_value} + + +def _apply_alignment(conn, target: int, force: bool, dry: bool): + new_key = f"{ALIGN_KEY}.{target}" + row = conn.execute(text( + "SELECT value FROM role_permissions WHERE `key` = :k" + ), {"k": ALIGN_KEY}).fetchone() + if not row: + return {"key": new_key, "action": "skip", "reason": "旧全局键不存在（无可搬运值）"} + + exist = conn.execute(text( + "SELECT value FROM role_permissions WHERE `key` = :k" + ), {"k": new_key}).fetchone() + if exist and not force: + return {"key": new_key, "action": "noop", + "reason": "新键已存在且未 --force（保留既有值，幂等）", + "existing_value": exist[0]} + if dry: + return {"key": new_key, "action": "would-insert", "value": row[0]} + + conn.execute(text( + "INSERT INTO role_permissions (`key`, value) VALUES (:k, :v)" + ), {"k": new_key, "v": row[0]}) + return {"key": new_key, "action": "inserted", "value": row[0]} + + +def _guard_db(args, conn): + db = _detect_db_name(conn) + if args.allow_db: + return db + if db != PROD_DB_NAME: + print(f"[拒绝] 库名 {db!r} ≠ 生产库 {PROD_DB_NAME!r}；" + f"如确为预期，请显式加 --allow-db {db}", file=sys.stderr) + sys.exit(2) + return db + + +def cmd_apply(args): + eng = _engine() + results = [] + with eng.begin() as conn: + db = _guard_db(args, conn) + # 目标账套存在性校验（防写出孤儿键） + ent = conn.execute(text("SELECT id FROM entities WHERE id = :i"), + {"i": args.target_entity}).fetchone() + if not ent: + print(f"[拒绝] 目标账套 entity_id={args.target_entity} 不存在", file=sys.stderr) + sys.exit(2) + results.append(_apply_cash(conn, args.target_entity, args.force, dry=False)) + results.append(_apply_alignment(conn, args.target_entity, args.force, dry=False)) + out = {"mode": "apply", "db": db, "target_entity": args.target_entity, "results": results} + print(json.dumps(out, ensure_ascii=False, indent=2) if args.json else + "\n".join(f" {r}" for r in results)) + return 0 + + +def cmd_down(args): + """回退：删除新键（旧全局键原样保留，老代码即可继续读）""" + eng = _engine() + removed = [] + with eng.begin() as conn: + db = _guard_db(args, conn) + r1 = conn.execute(text( + "DELETE FROM system_configs WHERE config_key LIKE :p"), {"p": CASH_KEY + ".%"}) + r2 = conn.execute(text( + "DELETE FROM role_permissions WHERE `key` LIKE :p"), {"p": ALIGN_KEY + ".%"}) + removed = [f"system_configs(+{CASH_KEY}.*) deleted={r1.rowcount}", + f"role_permissions({ALIGN_KEY}.*) deleted={r2.rowcount}"] + print(json.dumps({"mode": "down", "db": db, "removed": removed}, + ensure_ascii=False, indent=2) if args.json else "\n".join(removed)) + print("旧全局键未改动 → 代码回退到老版本即可继续读旧键。") + return 0 + + +def main(): + p = argparse.ArgumentParser(description="全局键拆账套 · 一次性搬运（cash.current_balance / alignment_config）") + g = p.add_mutually_exclusive_group() + g.add_argument("--check", action="store_true", help="只读预检（默认）") + g.add_argument("--apply", action="store_true", help="写库（需审批）") + g.add_argument("--down", action="store_true", help="回退：删除新键") + p.add_argument("--target-entity", type=int, default=1, + help="旧值落到哪个账套（默认 1 = 陕西酣客，生产账套）") + p.add_argument("--force", action="store_true", help="允许覆盖已存在的新键") + p.add_argument("--json", action="store_true", help="机器可读输出") + p.add_argument("--allow-db", default=None, help="生产库名保护豁免（默认仅允许 cma）") + args = p.parse_args() + + if args.apply: + return cmd_apply(args) + if args.down: + return cmd_down(args) + return cmd_check(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_alignment_entity_isolation_20260918.py b/backend/tests/test_alignment_entity_isolation_20260918.py index ee1bee77..da73adad 100644 --- a/backend/tests/test_alignment_entity_isolation_20260918.py +++ b/backend/tests/test_alignment_entity_isolation_20260918.py @@ -41,13 +41,19 @@ def _ensure_entity(db: Session, eid: int, name: str) -> Entity: 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})) +def _set_mode(db: Session, mode: str, entity_ids=(1, 2)) -> None: + """写入对齐模式配置 —— 2026-09-22 B3(a) 拆账套后为 `alignment_config.` + + 本文件断言的语义 =「两账套处在同一对齐模式时，各自只返回本账套 KPI」， + 故对 e1/e2 各写一份（修复前为全局唯一键，一份两账套共用）。 + """ + for eid in entity_ids: + key = f"alignment_config.{eid}" + perm = db.query(RolePermission).filter(RolePermission.key == key).first() + if perm: + perm.value = {"mode": mode} + else: + db.add(RolePermission(key=key, value={"mode": mode})) db.commit() diff --git a/backend/tests/test_notifications.py b/backend/tests/test_notifications.py index 6e5ec46a..722f5388 100644 --- a/backend/tests/test_notifications.py +++ b/backend/tests/test_notifications.py @@ -2,7 +2,7 @@ import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session -from tests.conftest import create_test_user, get_token_for_user, auth_header +from tests.conftest import create_test_user, create_test_kpi, get_token_for_user, auth_header class TestNotificationChannels: @@ -149,16 +149,32 @@ class TestNotificationLogs: assert resp.json()["total"] == 0 def test_logs_with_data(self, client: TestClient, db: Session): - """有了日志记录后可查询""" - from app.models import NotificationLog + """有了日志记录后可查询 + + 2026-09-22（dama-workbench-entity-isolation-fullstack-20260922 §四 B2）行为变更： + `notification_logs` 无 entity_id 列 → 改为**经父键 join 归属** + （`alert_id → kpi_alerts.id → kpi_alerts.entity_id`）。 + 故日志必须挂到**本账套真实存在**的 KPIAlert 上才可见； + `alert_id IS NULL` 或指向不存在父键的孤儿行**不返回**（宁少勿错）。 + """ + from app.models import NotificationLog, KPIAlert from datetime import datetime - # 直接插入日志 - log = NotificationLog( - alert_id=1, channel="wecom", recipient="测试群", - title="测试通知", content="测试内容", status="sent", sent_at=datetime.now(), - ) - db.add(log) + kpi = create_test_kpi(db, kpi_code="NOTIF_LOG_KPI", entity_id=1) + alert = KPIAlert(entity_id=1, kpi_id=kpi.id, alert_level="yellow", + alert_message="通知日志用例预警", status="pending") + db.add(alert) + db.commit() + db.refresh(alert) + + db.add_all([ + NotificationLog( + alert_id=alert.id, channel="wecom", recipient="测试群", + title="测试通知", content="测试内容", status="sent", sent_at=datetime.now(), + ), + # 孤儿行：无父键可归属 → 一律不返回 + NotificationLog(alert_id=None, channel="wecom", title="无归属通知", status="sent"), + ]) db.commit() create_test_user(db) @@ -166,8 +182,9 @@ class TestNotificationLogs: resp = client.get("/api/cma/notifications/logs", headers=auth_header(token)) assert resp.status_code == 200 - assert resp.json()["total"] >= 1 - assert len(resp.json()["data"]) >= 1 + assert resp.json()["total"] == 1, resp.json() + assert len(resp.json()["data"]) == 1 + assert resp.json()["data"][0]["title"] == "测试通知" class TestAlertPush: diff --git a/backend/tests/test_workbench_entity_isolation_20260922.py b/backend/tests/test_workbench_entity_isolation_20260922.py new file mode 100644 index 00000000..6f0bceeb --- /dev/null +++ b/backend/tests/test_workbench_entity_isolation_20260922.py @@ -0,0 +1,624 @@ +"""工作台跨实体污染 · 治本回归测试 — 2026-09-22 + +任务：`dama-workbench-entity-isolation-fullstack-20260922` +（DAMA 诊断 / 项目Bot batch5 §B 方案段 / 任总 2026-09-22「都批」） + +事故：任总 2026-09-22 报「陕西酣客工作台大量显示陕西博海数据」。 +根因族 = `system_configs` / `role_permissions` 的**全局唯一键**没有账套维度 +（`config_key` / `key` 均 unique）→ 一次写入全平台生效。 + +本单五项（逐条对应一个测试类）： + + P0 前端取数修正 — 工作台取数按**会话 entity**，不得用 `current_tenant` 全局键 + （含 current_tenant 收敛为「仅 it/ceo 可读」的项目Bot A2A 状态记录） + B1 `cash.current_balance` → `cash.current_balance.`（零 DDL） + B3a `alignment_config` → `alignment_config.`（零 DDL） + B3b `bot_bridge_v2` 预警 KPI 归属：删 `kpi_id or 1` 兜底，改显式校验（跳过错挂 + 留痕） + B2 `notification_logs` 经父键 join 归属（→ kpi_alerts.entity_id）；零迁移 + +外加：迁移脚本 `scripts/migrate_cash_balance_key_entity_20260922.py` 的 +check/apply/down 幂等与「原值只落目标账套、其余留空」口径。 + +全部用例只跑内存 SQLite / 临时文件 SQLite（tests/conftest.py 机制）， +**零生产库 / 零生产 Redis 连接**（conftest 已硬阻断）。 +""" +import os +import sys +import json + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, text +from sqlalchemy.orm import Session + +from app.models import ( + Entity, + KPIDefinition, + KPIAlert, + KPIValue, + NotificationLog, + OperationLog, + RolePermission, + SystemConfig, +) +from app.database import Base +from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user + +DASH = "/api/cma/dashboard" +CASH = "/api/cma/cash" +ALIGN = "/api/cma/alignment" +NOTIF = "/api/cma/notifications" +TENANT = "/api/cma/tenant" + +# 两账套的资源编码（kpi_code 库内唯一） +E1_KPI = "ZZ_WB_E1_KPI" +E2_KPI = "ZZ_WB_E2_KPI" + + +# ────────────────────────────────────────────── +# 夹具 +# ────────────────────────────────────────────── + +def _ensure_entity(db: Session, eid: int, name: str, status: str = "active") -> Entity: + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent: + ent = Entity(id=eid, name=name, short_name=name, status=status) + db.add(ent) + db.commit() + db.refresh(ent) + return ent + + +def _mk_kpi(db: Session, code: str, eid: int, **kw) -> 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="元", + status="active", + ) + kpi.__dict__.update(kw) + db.add(kpi) + db.commit() + db.refresh(kpi) + return kpi + + +def _login(client: TestClient, entity_id: int) -> str: + return _login_as(client, "testadmin", entity_id) + + +def _login_as(client: TestClient, username: str, entity_id: int) -> str: + resp = client.post("/api/cma/auth/login", json={ + "username": username, + "password": TEST_USER_PASSWORD, + "entity_id": entity_id, + }) + assert resp.status_code == 200, f"登录 {username}@entity{entity_id} 失败: {resp.text[:300]}" + return resp.json()["token"] + + +def _seed_two_entities(db: Session): + """两账套各一个 active KPI（双会话双实体互不串的最小底座）""" + _ensure_entity(db, 1, "陕西酣客文化传媒") + _ensure_entity(db, 2, "陕西博海网络科技") + k1 = _mk_kpi(db, E1_KPI, 1) + k2 = _mk_kpi(db, E2_KPI, 2) + # 各写一个最新值（便于断言「实际值」口径也隔离） + for k, v in ((k1, 111.0), (k2, 222.0)): + db.add(KPIValue(kpi_id=k.id, period="2026-09", actual_value=v, + source_type="test", data_status="verified")) + db.commit() + return k1, k2 + + +def _write_current_tenant(db: Session, entity_id: int, name: str) -> None: + """模拟事故现场：把全局键 current_tenant 写成某个账套""" + cfg = db.query(SystemConfig).filter(SystemConfig.config_key == "current_tenant").first() + payload = json.dumps({"tenant_id": f"company_{entity_id}", "entity_id": entity_id, + "name": name, "short_name": name, + "source": "test-polluter", + "switched_at": "2026-09-04T09:28:38"}, ensure_ascii=False) + if cfg: + cfg.config_value = payload + else: + db.add(SystemConfig(config_key="current_tenant", config_value=payload, + description="当前租户状态记录")) + db.commit() + + +# ══════════════════════════════════════════════ +# P0 前端取数修正：工作台按会话 entity，全局键不得当数据维度 +# ══════════════════════════════════════════════ + +class TestP0WorkbenchSessionEntity: + """事故直接触发点：工作台取数必须用**会话 entity**，与 current_tenant 全局键无关 + + 「双会话双实体互不串」= 同一进程内两个独立 Bearer token（e1 / e2）， + 即使全局键被写成对方账套，各自仍只看到本账套数据。 + """ + + def test_my_kpis_isolated_while_global_tenant_key_points_elsewhere( + self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) # role=ceo + _write_current_tenant(db, 2, "陕西博海网络科技") # 事故现场：全局键=博海 + + t1 = _login(client, 1) + t2 = _login(client, 2) + + r1 = client.get(f"{DASH}/my-kpis", headers=auth_header(t1)) + assert r1.status_code == 200, r1.text + codes1 = {k["kpi_code"] for k in r1.json()["data"]} + assert E1_KPI in codes1 and E2_KPI not in codes1, \ + f"酣客(e1)会话不应看到博海 KPI（全局键指向 e2 时也不许）：{sorted(codes1)}" + + r2 = client.get(f"{DASH}/my-kpis", headers=auth_header(t2)) + assert r2.status_code == 200, r2.text + codes2 = {k["kpi_code"] for k in r2.json()["data"]} + assert E2_KPI in codes2 and E1_KPI not in codes2, \ + f"博海(e2)会话不应看到酣客 KPI：{sorted(codes2)}" + + # 反转全局键 → 两会话结果必须**不变**（证明取数不依赖全局键） + _write_current_tenant(db, 1, "陕西酣客文化传媒") + codes1b = {k["kpi_code"] for k in + client.get(f"{DASH}/my-kpis", headers=auth_header(t1)).json()["data"]} + codes2b = {k["kpi_code"] for k in + client.get(f"{DASH}/my-kpis", headers=auth_header(t2)).json()["data"]} + assert codes1b == codes1, "全局键翻转后 e1 工作台数据变了 ⇒ 取数依赖了全局键" + assert codes2b == codes2, "全局键翻转后 e2 工作台数据变了 ⇒ 取数依赖了全局键" + + def test_my_dashboard_and_summary_scoped_by_session( + self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) + _write_current_tenant(db, 2, "陕西博海网络科技") + + t1 = _login(client, 1) + t2 = _login(client, 2) + + d1 = client.get(f"{DASH}/my-dashboard", headers=auth_header(t1)) + d2 = client.get(f"{DASH}/my-dashboard", headers=auth_header(t2)) + assert d1.status_code == 200 and d2.status_code == 200 + codes1 = {k["kpi_code"] for k in d1.json()["kpis"]} + codes2 = {k["kpi_code"] for k in d2.json()["kpis"]} + assert not (codes1 & codes2), f"两会话工作台 KPI 出现交集：{codes1 & codes2}" + + s1 = client.get(f"{DASH}/summary", headers=auth_header(t1)) + s2 = client.get(f"{DASH}/summary", headers=auth_header(t2)) + assert s1.status_code == 200 and s2.status_code == 200 + + def test_workbench_endpoints_all_use_session_entity(self, client: TestClient, db: Session): + """工作台类端点逐个复核：带 token 时按 token 账套取数（不看全局键）""" + _seed_two_entities(db) + create_test_user(db) + _write_current_tenant(db, 2, "陕西博海网络科技") + t1 = _login(client, 1) + for path in (f"{DASH}/summary", f"{DASH}/kpis", f"{DASH}/my-kpis", + f"{DASH}/my-dashboard"): + r = client.get(path, headers=auth_header(t1)) + assert r.status_code == 200, f"{path} 期望 200，实际 {r.status_code}: {r.text[:200]}" + # 显式传他账套 entity_id → 403（会话账套为准） + r = client.get(f"{DASH}/my-kpis?entity_id=2", headers=auth_header(t1)) + assert r.status_code == 403, r.text + assert "无权访问" in r.json()["detail"] + + +#: 工作台类端点矩阵（§二.3「前端/接口两侧都要复核」的接口侧证据） +WORKBENCH_ENDPOINTS = [ + "/api/cma/dashboard/summary", + "/api/cma/dashboard/kpis", + "/api/cma/dashboard/my-kpis", + "/api/cma/dashboard/my-dashboard", + "/api/cma/dashboard/finance-analysis", + "/api/cma/dashboard/predict", + "/api/cma/kpis?page_size=5", + "/api/cma/alerts", + "/api/cma/cash/dashboard", + "/api/cma/cash/balance", + "/api/cma/budget/plans", + "/api/cma/budget/config", + "/api/cma/expenses/reimbursements", + "/api/cma/expenses/stats", + "/api/cma/alignment/config", + "/api/cma/alignment/tree", + "/api/cma/notifications/logs", + "/api/cma/data-quality/logs", +] + + +class TestP0WorkbenchEndpointMatrix: + """工作台类端点「双会话双实体」矩阵复核：任一端点都不得把对方账套数据带出来 + + 方法论（可复算）：两个独立 Bearer token（e1 / e2）逐个打同一端点， + 断言 ①两会话状态码一致（不得出现「一方 200 一方 403」的不对称） + ②任一响应体里**不含**对方账套的独有标记（kpi_code）。 + 同时把三个全局键都写成 e2（模拟事故现场），证明取数与全局键无关。 + """ + + def test_no_cross_entity_marker_in_any_workbench_endpoint( + self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) + _write_current_tenant(db, 2, "陕西博海网络科技") + db.add(SystemConfig(config_key="cash.current_balance", config_value="-9.3", + description="历史全局键（事故值）")) + db.commit() + + t1 = _login(client, 1) + t2 = _login(client, 2) + report = [] + violations = [] + for path in WORKBENCH_ENDPOINTS: + r1 = client.get(path, headers=auth_header(t1)) + r2 = client.get(path, headers=auth_header(t2)) + report.append(f"{path} → e1:{r1.status_code} e2:{r2.status_code}") + if r1.status_code != r2.status_code: + violations.append(f"{path} 状态码不对称 e1={r1.status_code} e2={r2.status_code}") + if E2_KPI in r1.text: + violations.append(f"{path} e1 会话响应含 e2 标记 {E2_KPI}") + if E1_KPI in r2.text: + violations.append(f"{path} e2 会话响应含 e1 标记 {E1_KPI}") + print("\n[工作台端点矩阵]\n" + "\n".join(report)) + assert not violations, "跨实体泄漏/不对称:\n" + "\n".join(violations) + + +class TestP0CurrentTenantConvergence: + + def test_no_token_401(self, client: TestClient, db: Session): + assert client.get(f"{TENANT}/current").status_code == 401 + + def test_non_admin_role_403(self, client: TestClient, db: Session): + _ensure_entity(db, 1, "测试企业") + create_test_user(db) # testadmin / ceo + create_test_user(db, username="fin_user", role="finance") + tk_fin = _login_as(client, "fin_user", 1) + tk_ceo = _login(client, 1) + r = client.get(f"{TENANT}/current", headers=auth_header(tk_fin)) + assert r.status_code == 403, f"finance 会话读 current_tenant 应 403，实际 {r.status_code}" + # ceo（业务负责人）与 it（admin 动作权限持有者）放行 + assert client.get(f"{TENANT}/current", headers=auth_header(tk_ceo)).status_code == 200 + + def test_switch_writes_audit_log_and_is_not_a_data_dimension( + self, client: TestClient, db: Session): + """POST /switch 落 operation_logs（修复前无审计）；写入后不影响他账套取数""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + r = client.post(f"{TENANT}/switch", headers=auth_header(t1), + json={"entity_id": 2, "source": "test"}) + assert r.status_code == 200, r.text + db.expire_all() + n = db.query(OperationLog).filter(OperationLog.action == "switch_tenant_state").count() + assert n == 1, f"switch 应落 1 条审计，实际 {n}" + # 全局键虽被改成 e2，e1 会话工作台仍只见 e1 数据 + codes1 = {k["kpi_code"] for k in + client.get(f"{DASH}/my-kpis", headers=auth_header(t1)).json()["data"]} + assert E2_KPI not in codes1 + + +# ══════════════════════════════════════════════ +# B1 现金基线拆账套 +# ══════════════════════════════════════════════ + +class TestB1CashBalancePerEntity: + + def test_two_entities_do_not_overwrite_each_other(self, client: TestClient, db: Session): + """e1 / e2 分别设基线后各自读回本账套值（修复前共用一个唯一键）""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + t2 = _login(client, 2) + + assert client.post(f"{CASH}/balance", headers=auth_header(t1), + json={"current_cash": 11.5}).status_code == 200 + assert client.post(f"{CASH}/balance", headers=auth_header(t2), + json={"current_cash": 22.5}).status_code == 200 + + v1 = client.get(f"{CASH}/balance", headers=auth_header(t1)).json() + v2 = client.get(f"{CASH}/balance", headers=auth_header(t2)).json() + assert v1["current_cash"] == 11.5, f"e1 基线被覆盖：{v1}" + assert v2["current_cash"] == 22.5, f"e2 基线被覆盖：{v2}" + assert v1["entity_id"] == 1 and v2["entity_id"] == 2 + + db.expire_all() + keys = {r[0] for r in db.query(SystemConfig.config_key) + .filter(SystemConfig.config_key.like("cash.current_balance%")).all()} + assert keys == {"cash.current_balance.1", "cash.current_balance.2"}, keys + # 不再写全局唯一键 + assert "cash.current_balance" not in keys + + def test_no_cross_entity_fallback_to_legacy_key(self, client: TestClient, db: Session): + """旧全局键有值（事故值 -9.3），本账套未设基线 → 返回 None，**不借用**他账套值""" + _seed_two_entities(db) + create_test_user(db) + db.add(SystemConfig(config_key="cash.current_balance", config_value="-9.3", + description="历史全局键（事故值）")) + db.commit() + t2 = _login(client, 2) + body = client.get(f"{CASH}/balance", headers=auth_header(t2)).json() + assert body["current_cash"] is None, \ + f"e2 未设基线却读到旧全局键（跨账套借用）：{body}" + + def test_body_entity_id_not_trusted(self, client: TestClient, db: Session): + """带 token 时 body 里的 entity_id 不可信（仍写会话账套）""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + r = client.post(f"{CASH}/balance", headers=auth_header(t1), + json={"current_cash": 33.0, "entity_id": 2}) + assert r.status_code == 200, r.text + assert r.json()["entity_id"] == 1, r.json() + db.expire_all() + assert db.query(SystemConfig).filter( + SystemConfig.config_key == "cash.current_balance.2").first() is None + assert db.query(SystemConfig).filter( + SystemConfig.config_key == "cash.current_balance.1").first() is not None + + def test_set_helper_requires_entity(self, db: Session): + """set_current_cash_balance 必须传 entity_id（防漏传落默认账套）""" + from app.utils.cash_forecast_engine import set_current_cash_balance + with pytest.raises(TypeError): + set_current_cash_balance(db, 1.0) # 缺 entity_id + with pytest.raises(ValueError): + set_current_cash_balance(db, 1.0, None) + + +# ══════════════════════════════════════════════ +# B3(a) 对齐模式拆账套 +# ══════════════════════════════════════════════ + +class TestB3aAlignmentPerEntity: + + def test_config_isolated_per_entity(self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + t2 = _login(client, 2) + + assert client.post(f"{ALIGN}/config", headers=auth_header(t1), + json={"mode": "vertical_decomposition"}).status_code == 200 + assert client.post(f"{ALIGN}/config", headers=auth_header(t2), + json={"mode": "bsc_chain"}).status_code == 200 + + c1 = client.get(f"{ALIGN}/config", headers=auth_header(t1)).json() + c2 = client.get(f"{ALIGN}/config", headers=auth_header(t2)).json() + assert c1["mode"]["mode"] == "vertical_decomposition", c1 + assert c2["mode"]["mode"] == "bsc_chain", c2 + assert c1["entity_id"] == 1 and c2["entity_id"] == 2 + + db.expire_all() + keys = {r[0] for r in db.query(RolePermission.key) + .filter(RolePermission.key.like("alignment_config%")).all()} + assert keys == {"alignment_config.1", "alignment_config.2"}, keys + assert "alignment_config" not in keys + + def test_unconfigured_entity_not_inheriting_other_entity(self, client: TestClient, db: Session): + """e1 已配置、e2 未配置 → e2 不得继承 e1 的模式（修复前全局键必然继承）""" + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + t2 = _login(client, 2) + assert client.post(f"{ALIGN}/config", headers=auth_header(t1), + json={"mode": "bsc_chain"}).status_code == 200 + + c2 = client.get(f"{ALIGN}/config", headers=auth_header(t2)).json() + assert c2["configured"] is False and c2["mode"] is None, \ + f"e2 未配置却继承了 e1 的对齐模式：{c2}" + # e2 的 /tree 应提示先初始化（400），而不是拿到 e1 的模式 + r = client.get(f"{ALIGN}/tree", headers=auth_header(t2)) + assert r.status_code == 400, f"e2 未配置应 400，实际 {r.status_code}: {r.text[:200]}" + # e1 正常 + assert client.get(f"{ALIGN}/tree", headers=auth_header(t1)).status_code == 200 + + def test_tree_uses_own_entity_mode(self, client: TestClient, db: Session): + _seed_two_entities(db) + create_test_user(db) + t1 = _login(client, 1) + t2 = _login(client, 2) + client.post(f"{ALIGN}/config", headers=auth_header(t1), + json={"mode": "vertical_decomposition"}) + client.post(f"{ALIGN}/config", headers=auth_header(t2), json={"mode": "bsc_chain"}) + b1 = client.get(f"{ALIGN}/tree", headers=auth_header(t1)).json() + b2 = client.get(f"{ALIGN}/tree", headers=auth_header(t2)).json() + assert b1["mode"] == "vertical_decomposition" + assert b2["mode"] == "bsc_chain" + + +# ══════════════════════════════════════════════ +# B3(b) 预警 KPI 归属：删 kpi_id or 1 兜底 +# ══════════════════════════════════════════════ + +class TestB3bAlertKpiAttribution: + + def test_missing_kpi_definition_skips_alert_instead_of_kpi_id_1(self, db: Session): + """e2 无 C_REBATE_RATE 定义时：不建预警（不挂 kpi_id=1）+ 落 OperationLog 留痕""" + from app.api.bot_bridge_v2 import receive_mpm_result + from app.models import MpmResult + + _ensure_entity(db, 1, "陕西酣客文化传媒") + _ensure_entity(db, 2, "陕西博海网络科技") + first = _mk_kpi(db, "ZZ_WB_FIRST_KPI", 1) # 全库第 1 个 KPI（旧兜底目标） + assert first.id == 1, f"夹具前提：第一个 KPI id 应为 1，实际 {first.id}" + MpmResult.__table__.create(bind=db.get_bind(), checkfirst=True) + before_alerts = db.query(KPIAlert).count() + before_logs = db.query(OperationLog).count() + + r = receive_mpm_result( + data={"entity_id": 2, "period": "2026-09", + "results": {"channel_rebate_rate": 80.0}}, + bridge_bot="test-bridge", db=db) + db.expire_all() + + assert r["alerts_triggered"] == 0, r + assert r["alerts_skipped_missing_kpi"] == 1, r + assert db.query(KPIAlert).count() == before_alerts, "KPI 定义缺失不应建预警" + assert db.query(KPIAlert).filter( + KPIAlert.entity_id == 2, KPIAlert.kpi_id == 1).first() is None, \ + "不得把 e2 的预警错挂到 kpi_id=1（可能是他账套的 KPI）" + logs = db.query(OperationLog).filter(OperationLog.action == "skip_alert").all() + assert len(logs) == 1 and db.query(OperationLog).count() == before_logs + 1 + assert logs[0].detail["reason"] == "kpi_definition_not_found_for_entity" + assert logs[0].detail["entity_id"] == 2 + + def test_present_kpi_definition_still_creates_alert(self, db: Session): + """正对照：本账套有 KPI 定义 → 照常建预警且归属正确""" + from app.api.bot_bridge_v2 import receive_mpm_result + from app.models import MpmResult + + _ensure_entity(db, 1, "陕西酣客文化传媒") + k = _mk_kpi(db, "C_REBATE_RATE", 1) + MpmResult.__table__.create(bind=db.get_bind(), checkfirst=True) + r = receive_mpm_result( + data={"entity_id": 1, "period": "2026-09", + "results": {"channel_rebate_rate": 80.0}}, + bridge_bot="test-bridge", db=db) + db.expire_all() + assert r["alerts_triggered"] == 1, r + assert r["alerts_skipped_missing_kpi"] == 0 + alert = db.query(KPIAlert).filter(KPIAlert.entity_id == 1).first() + assert alert is not None and alert.kpi_id == k.id + + +# ══════════════════════════════════════════════ +# B2 缺列五表：Log 类经父键 join 归属（零迁移） +# ══════════════════════════════════════════════ + +class TestB2NotificationLogScoping: + + def test_logs_scoped_by_alert_entity(self, client: TestClient, db: Session): + k1, k2 = _seed_two_entities(db) + create_test_user(db) + a1 = KPIAlert(entity_id=1, kpi_id=k1.id, alert_level="red", + alert_message="e1 预警", status="pending") + a2 = KPIAlert(entity_id=2, kpi_id=k2.id, alert_level="red", + alert_message="e2 预警", status="pending") + db.add_all([a1, a2]) + db.commit() + db.refresh(a1) + db.refresh(a2) + db.add_all([ + NotificationLog(alert_id=a1.id, channel="wecom", title="e1 通知", status="sent"), + NotificationLog(alert_id=a2.id, channel="wecom", title="e2 通知", status="sent"), + NotificationLog(alert_id=None, channel="wecom", title="无归属通知", status="sent"), + ]) + db.commit() + + t1 = _login(client, 1) + t2 = _login(client, 2) + l1 = client.get(f"{NOTIF}/logs", headers=auth_header(t1)).json() + l2 = client.get(f"{NOTIF}/logs", headers=auth_header(t2)).json() + + titles1 = {x["title"] for x in l1["data"]} + titles2 = {x["title"] for x in l2["data"]} + assert titles1 == {"e1 通知"}, f"e1 会话通知日志混入他账套/无归属行：{titles1}" + assert titles2 == {"e2 通知"}, f"e2 会话通知日志混入他账套/无归属行：{titles2}" + assert l1["total"] == 1 and l2["total"] == 1 + assert l1["entity_id"] == 1 and l2["entity_id"] == 2 + + +# ══════════════════════════════════════════════ +# 迁移脚本：原值搬运口径（目标账套落值 / 其余留空 / 幂等 / 可回退） +# ══════════════════════════════════════════════ + +MIGRATION = "scripts.migrate_cash_balance_key_entity_20260922" + + +class TestMigrationScript: + """在**临时文件 SQLite**上跑迁移脚本（monkeypatch 引擎，绝不碰 app.database → 生产库）""" + + @pytest.fixture + def mig_env(self, tmp_path, monkeypatch): + import importlib + sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + mod = importlib.import_module(MIGRATION) + eng = create_engine(f"sqlite:///{tmp_path/'mig.db'}") + Base.metadata.create_all(bind=eng) + with eng.begin() as conn: + conn.execute(text("INSERT INTO entities (id, name, short_name, status) " + "VALUES (1, '酣客', '酣客', 'active')")) + conn.execute(text("INSERT INTO entities (id, name, short_name, status) " + "VALUES (2, '博海', '博海', 'active')")) + conn.execute(text("INSERT INTO system_configs (config_key, config_value, description) " + "VALUES ('cash.current_balance', '-9.3', '历史全局键')")) + conn.execute(text("INSERT INTO role_permissions (`key`, value) " + "VALUES ('alignment_config', '{\"mode\": \"bsc_chain\"}')")) + monkeypatch.setattr(mod, "_engine", lambda: eng) + return mod, eng + + def _run(self, mod, argv): + import sys as _sys + old = _sys.argv + _sys.argv = ["migrate"] + argv + try: + return mod.main() + finally: + _sys.argv = old + + def test_check_is_readonly(self, mig_env): + mod, eng = mig_env + assert self._run(mod, ["--check", "--allow-db", "sqlite"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'cash.current_balance.%'")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM role_permissions " + "WHERE `key` LIKE 'alignment_config.%'")).scalar() == 0 + + def test_apply_moves_value_to_target_only_and_is_idempotent(self, mig_env): + mod, eng = mig_env + assert self._run(mod, ["--apply", "--allow-db", "sqlite", + "--target-entity", "1", "--json"]) == 0 + with eng.connect() as conn: + v = conn.execute(text("SELECT config_value FROM system_configs " + "WHERE config_key = 'cash.current_balance.1'")).scalar() + assert v == "-9.3" + # 其余账套留空（不跨账套借用） + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key = 'cash.current_balance.2'")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM role_permissions " + "WHERE `key` = 'alignment_config.1'")).scalar() == 1 + assert conn.execute(text("SELECT COUNT(*) FROM role_permissions " + "WHERE `key` = 'alignment_config.2'")).scalar() == 0 + # 旧键保留（回退杠杆） + assert conn.execute(text("SELECT config_value FROM system_configs " + "WHERE config_key = 'cash.current_balance'")).scalar() == "-9.3" + # 二跑零副作用（幂等） + assert self._run(mod, ["--apply", "--allow-db", "sqlite", + "--target-entity", "1", "--json"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'cash.current_balance.%'")).scalar() == 1 + assert conn.execute(text("SELECT config_value FROM system_configs " + "WHERE config_key = 'cash.current_balance.1'")).scalar() == "-9.3" + + def test_down_removes_new_keys_keeps_legacy(self, mig_env): + mod, eng = mig_env + self._run(mod, ["--apply", "--allow-db", "sqlite", "--json"]) + assert self._run(mod, ["--down", "--allow-db", "sqlite", "--json"]) == 0 + with eng.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key LIKE 'cash.current_balance.%'")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM role_permissions " + "WHERE `key` LIKE 'alignment_config.%'")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM system_configs " + "WHERE config_key = 'cash.current_balance'")).scalar() == 1 + + def test_prod_db_name_guard(self, mig_env): + """库名保护：非 cma 库未显式豁免 → 拒绝（exit 2）""" + mod, eng = mig_env + with pytest.raises(SystemExit) as e: + self._run(mod, ["--apply", "--json"]) # sqlite → DATABASE() 不可用 → db='' + assert e.value.code == 2 + + def test_target_entity_must_exist(self, mig_env): + mod, eng = mig_env + with pytest.raises(SystemExit) as e: + self._run(mod, ["--apply", "--allow-db", "sqlite", "--target-entity", "99", "--json"]) + assert e.value.code == 2