"""工作台跨实体污染 · 治本回归测试 — 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}" # 审计行必须可归属（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