fullstack
30fa745135
DAMA 独立回查 dama-recheck-workbench-entity-isolation-20260922 的 §4b/§5 两处待办：
1. POST /api/cma/tenant/switch 审计行原来不可归属：
- user_id=None → 改为注入会话用户（user=Depends(require_auth_401)）
- target_type/target_id 从 system_config/None 对齐为 entity/<entity_id>，
与 /auth/switch-entity 的 switch_entity 共用目标维度 ⇒ 按账套可过滤，零 DDL
- action 字面 switch_tenant_state 保留不合并（与会话级 switch_entity 语义不同），
口径登记 docs/audit-and-notification-caliber-20260922.md
2. /api/cma/notifications/logs 增加 policy/note 口径字段，前端发送记录表
empty-text 提示同一口径 —— 避免「B2 上线后 e1/e2 均 0 条」被当成页面故障
（现库 8 行历史记录全为 alert_id IS NULL 的 OKR 复盘提醒，无账套归属）
回归：tests/test_workbench_entity_isolation_20260922.py 22/22；全量 1370 passed；前端 vite build 通过
637 lines
31 KiB
Python
637 lines
31 KiB
Python
"""工作台跨实体污染 · 治本回归测试 — 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.<entity_id>`（零 DDL）
|
||
B3a `alignment_config` → `alignment_config.<entity_id>`（零 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}"
|
||
# 审计行必须可归属（2026-09-22 DAMA 独立回查 §4b）：操作人 + 账套目标维度
|
||
log = (db.query(OperationLog)
|
||
.filter(OperationLog.action == "switch_tenant_state")
|
||
.order_by(OperationLog.id.desc()).first())
|
||
assert log.user_id is not None, "审计行 user_id 不得为 None（事后须能追操作人）"
|
||
assert log.target_type == "entity" and log.target_id == 2, (
|
||
f"审计须按账套可过滤（target_type/target_id），实际 {log.target_type}/{log.target_id}")
|
||
assert (log.detail or {}).get("entity_id") == 2, f"detail 须含 entity_id：{log.detail}"
|
||
# 全局键虽被改成 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
|
||
# 口径字段（2026-09-22 DAMA 回查 §5）：空页要能解释成口径而非故障
|
||
for l in (l1, l2):
|
||
assert l["policy"] == "alert-linked-only", l
|
||
assert "无父键" in l["note"] and "okr_review_reminder" in l["note"], l
|
||
|
||
|
||
# ══════════════════════════════════════════════
|
||
# 迁移脚本：原值搬运口径（目标账套落值 / 其余留空 / 幂等 / 可回退）
|
||
# ══════════════════════════════════════════════
|
||
|
||
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
|