"""隔离扫尾 · 同源全局键补漏 + 审计缺口收口 — 2026-09-23 任务：`dama-entity-isolation-sweep-20260923` （DAMA 独立回查新增漏项 / 任总 2026-09-23「按照建议执行」；父单 `dama-workbench-entity-isolation-fullstack-20260922`） 三块（逐条对应一个测试类）： ① `kpi_alert_higher_better` 拆账套 → `kpi_alert_higher_better.`（零 DDL） （消费点：`utils/deviation_engine.py::get_higher_better_codes`； `api/budget.py` GET/PUT `/alert-direction`） ② `kpi_visibility`（`role_permissions`）**口径裁定**：角色级天然全局、**非账套维度**， 不拆；口径写进 `auth_middleware` 注释 + `docs/global-config-keys-caliber-20260923.md` ③ `POST /tenant/switch` 审计收口：`detail` 显式 from→to/source/operator； **审计写失败不静默**（同事务回滚 + `AUDIT_WRITE_FAILED` ERROR + HTTP 500） 外加：搬运脚本 `scripts/migrate_higher_better_key_entity_20260923.py` 的 check/apply/down 幂等、「原值只落目标账套」与「库内无旧值 ⇒ skip 零写入」。 全部用例只跑内存 SQLite / 临时文件 SQLite（tests/conftest.py 机制）， **零生产库 / 零生产 Redis 连接**（conftest 已硬阻断）。 """ import importlib import json import os import re import sys import ast import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine, inspect, text from sqlalchemy.orm import Session from app.models import Entity, OperationLog, RolePermission, SystemConfig from app.database import Base from app.utils.oplog import (BATCH_NO_ENTITY_TAG, NO_ENTITY_TAG, SCOPE_BATCH, SCOPE_GLOBAL, log_op) from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user BUDGET = "/api/cma/budget" TENANT = "/api/cma/tenant" LEGACY_KEY = "kpi_alert_higher_better" SENTINEL_CODE = "ZZ_LEGACY_ONLY_CODE" # 只存在于「旧全局键」的哨兵，用于证伪回退 # ────────────────────────────────────────────── # 夹具 # ────────────────────────────────────────────── 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 _seed_two_entities(db: Session): _ensure_entity(db, 1, "陕西酣客文化传媒") _ensure_entity(db, 2, "陕西博海网络科技") 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 _set_legacy_global(db: Session, codes: list) -> None: """模拟事故现场：旧全局唯一键有值（拆键前所有账套共用的那一行）""" cfg = db.query(SystemConfig).filter(SystemConfig.config_key == LEGACY_KEY).first() val = json.dumps(codes, ensure_ascii=False) if cfg: cfg.config_value = val else: db.add(SystemConfig(config_key=LEGACY_KEY, config_value=val, description="历史全局键（事故值）")) db.commit() # ══════════════════════════════════════════════ # ① kpi_alert_higher_better 拆账套 # ══════════════════════════════════════════════ class TestHigherBetterPerEntity: 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_as(client, "testadmin", 1) t2 = _login_as(client, "testadmin", 2) r1 = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t1), json={"codes": ["SALES_TOTAL", "CUSTOMER_COUNT"]}) assert r1.status_code == 200, r1.text r2 = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t2), json={"codes": ["COST_TOTAL"]}) assert r2.status_code == 200, r2.text v1 = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t1)).json() v2 = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t2)).json() assert v1["codes"] == ["SALES_TOTAL", "CUSTOMER_COUNT"], f"e1 配置被覆盖：{v1}" assert v2["codes"] == ["COST_TOTAL"], f"e2 配置被覆盖：{v2}" assert v1["entity_id"] == 1 and v2["entity_id"] == 2 assert v1["config_key"] == f"{LEGACY_KEY}.1" assert v2["config_key"] == f"{LEGACY_KEY}.2" db.expire_all() keys = {r[0] for r in db.query(SystemConfig.config_key) .filter(SystemConfig.config_key.like(f"{LEGACY_KEY}%")).all()} assert keys == {f"{LEGACY_KEY}.1", f"{LEGACY_KEY}.2"}, keys assert LEGACY_KEY not in keys, "不得再写全局唯一键" def test_no_cross_entity_fallback_to_legacy_key(self, client: TestClient, db: Session): """旧全局键有值（事故值），本账套未配置 → 走内置默认，**不借用**旧全局值""" _seed_two_entities(db) create_test_user(db) _set_legacy_global(db, [SENTINEL_CODE]) # 事故现场 t2 = _login_as(client, "testadmin", 2) body = client.get(f"{BUDGET}/alert-direction", headers=auth_header(t2)).json() assert SENTINEL_CODE not in body["codes"], \ f"e2 未配置却读到旧全局键（跨账套借用）：{body}" assert body["codes"], "未配置须回落内置默认表（兜底语义）" assert body["is_configured"] is False def test_helper_requires_entity_id(self): """读/写辅助函数的账套参数必传：漏传 → TypeError / ValueError（防落默认账套）""" from app.utils.deviation_engine import get_higher_better_codes, higher_better_key with pytest.raises(TypeError): get_higher_better_codes(None) # 缺 entity_id（位置参数必传） with pytest.raises(ValueError): higher_better_key(None) assert higher_better_key(2) == f"{LEGACY_KEY}.2" 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_as(client, "testadmin", 1) r = client.put(f"{BUDGET}/alert-direction", headers=auth_header(t1), json={"codes": ["SALES_TOTAL"], "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 == f"{LEGACY_KEY}.2").first() is None def test_patch_migrate_scope_unchanged(self): """读侧不得再出现「回退旧全局键」分支（源级断言，防回归）""" src = open(os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "app", "utils", "deviation_engine.py"), encoding="utf-8").read() fn = src[src.index("def get_higher_better_codes"):src.index("# ====", src.index("def get_higher_better_codes"))] assert "higher_better_key(entity_id)" in fn assert re.search(r"config_key\s*==\s*CONFIG_KEY_HIGHER_BETTER\b", fn) is None, \ "读路径不得回退旧全局键（回退即跨账套借用）" # ══════════════════════════════════════════════ # ② kpi_visibility 口径裁定：非账套维度 # ══════════════════════════════════════════════ class TestKpiVisibilityNonEntityDimension: """裁定：角色级天然全局（不拆账套）——用结构 + 行为 + 登记三面钉住""" def test_role_permissions_has_no_entity_column(self, db: Session): """结构面：role_permissions 无 entity_id 列（key 唯一）⇒ 天然无账套维度""" cols = {c["name"] for c in inspect(db.get_bind()).get_columns("role_permissions")} assert "entity_id" not in cols, cols def test_visibility_is_role_keyed_not_entity_keyed(self, db: Session): """行为面：读的是裸键 `kpi_visibility`（值 = {role: [规则]}），不受账套影响""" from app import auth_middleware as am db.add(RolePermission(key="kpi_visibility", value={"finance": ["finance_*"], "ceo": ["*"]})) db.commit() am._load_kpi_visibility._cache = {"data": None, "ts": 0} # 清缓存 vis = am._load_kpi_visibility(db) assert vis["finance"] == ["finance_*"] and vis["ceo"] == ["*"] am._load_kpi_visibility._cache = {"data": None, "ts": 0} # 不带账套参数也照读（角色维度，无账套语义） assert am._load_kpi_visibility(db)["finance"] == ["finance_*"] def test_caliber_documented_in_code(self): """登记面：「非账套维度」口径必须写在代码里（否则下轮审计重复报缺口）""" p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "app", "auth_middleware.py") src = open(p, encoding="utf-8").read() assert "非账套维度" in src and "kpi_visibility" in src # ══════════════════════════════════════════════ # ③ tenant/switch 审计收口 # ══════════════════════════════════════════════ class TestTenantSwitchAudit: def _seed_tenant_key(self, db: Session, entity_id: int, name: str): db.add(SystemConfig( config_key="current_tenant", config_value=json.dumps({"tenant_id": "company_a", "entity_id": entity_id, "name": name, "short_name": name, "source": "seed", "switched_at": "2026-09-22T10:36:01"}, ensure_ascii=False), description="当前租户状态记录")) db.commit() def test_audit_detail_carries_from_to_source_operator(self, client: TestClient, db: Session): _seed_two_entities(db) create_test_user(db) self._seed_tenant_key(db, 1, "陕西酣客文化传媒") t1 = _login_as(client, "testadmin", 1) r = client.post(f"{TENANT}/switch", headers=auth_header(t1), json={"entity_id": 2, "source": "sweep-test"}) assert r.status_code == 200, r.text db.expire_all() log = (db.query(OperationLog) .filter(OperationLog.action == "switch_tenant_state") .order_by(OperationLog.id.desc()).first()) assert log is not None, "写路径必须落审计（修复前无 operation_logs 行）" assert log.user_id == 1 assert log.target_type == "entity" and log.target_id == 2 d = log.detail or {} assert d.get("from_entity_id") == 1, f"detail 须含 from→to：{d}" assert d.get("to_entity_id") == 2, d assert d.get("entity_id") == 2, d # 兼容既有消费方 assert d.get("source") == "sweep-test", d op = d.get("operator") or {} assert op.get("user_id") == 1 and op.get("username") == "testadmin", op assert op.get("role") == "ceo", op def test_audit_write_failure_is_not_silent(self, client: TestClient, db: Session, monkeypatch, caplog): """审计写不进去 → 状态变更一并回滚 + ERROR 留痕 + HTTP 500（不静默）""" import logging from app.api import tenant as tenant_mod _seed_two_entities(db) create_test_user(db) self._seed_tenant_key(db, 1, "陕西酣客文化传媒") t1 = _login_as(client, "testadmin", 1) class _Boom: def __init__(self, *a, **kw): raise RuntimeError("simulated audit table failure") monkeypatch.setattr(tenant_mod, "OperationLog", _Boom) with caplog.at_level(logging.ERROR): r = client.post(f"{TENANT}/switch", headers=auth_header(t1), json={"entity_id": 2, "source": "sweep-test"}) assert r.status_code == 500, f"审计失败必须显式失败，实际 {r.status_code}: {r.text[:200]}" assert any("AUDIT_WRITE_FAILED" in rec.getMessage() for rec in caplog.records), \ [rec.getMessage() for rec in caplog.records] db.expire_all() cfg = db.query(SystemConfig).filter(SystemConfig.config_key == "current_tenant").first() assert json.loads(cfg.config_value)["entity_id"] == 1, \ "审计失败时状态变更必须回滚（不得留「改了但没审计」）" assert db.query(OperationLog).filter( OperationLog.action == "switch_tenant_state").count() == 0 def test_get_current_has_no_write_path(self, client: TestClient, db: Session): """GET /current 纯只读：读一次不产生任何审计行（不给读操作写审计表）""" _ensure_entity(db, 1, "陕西酣客文化传媒") create_test_user(db) t1 = _login_as(client, "testadmin", 1) assert client.get(f"{TENANT}/current", headers=auth_header(t1)).status_code == 200 db.expire_all() assert db.query(OperationLog).count() == 0 # ══════════════════════════════════════════════ # 搬运脚本（临时文件 SQLite，绝不碰 app.database → 生产库） # ══════════════════════════════════════════════ MIGRATION = "scripts.migrate_higher_better_key_entity_20260923" class TestMigrationScript: pytestmark = [] @pytest.fixture def mig_env(self, tmp_path, monkeypatch): 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 ('kpi_alert_higher_better', '[\"SALES_TOTAL\"]', '历史全局键')")) monkeypatch.setattr(mod, "_engine", lambda: eng) return mod, eng def _run(self, mod, argv): 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 'kpi_alert_higher_better.%'")).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 = 'kpi_alert_higher_better.1'")).scalar() assert json.loads(v) == ["SALES_TOTAL"] assert conn.execute(text("SELECT COUNT(*) FROM system_configs " "WHERE config_key = 'kpi_alert_higher_better.2'")).scalar() == 0 assert conn.execute(text("SELECT config_value FROM system_configs " "WHERE config_key = 'kpi_alert_higher_better'")).scalar() == \ '["SALES_TOTAL"]' # 旧键保留（回退杠杆） 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 'kpi_alert_higher_better.%'")).scalar() == 1 def test_apply_without_legacy_row_is_skip_zero_write(self, tmp_path, monkeypatch): """生产实况：库内无旧值 ⇒ --apply 走 skip、零写入（拆键后行为与拆键前逐字一致）""" 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/'empty.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')")) monkeypatch.setattr(mod, "_engine", lambda: eng) assert self._run(mod, ["--apply", "--allow-db", "sqlite", "--json"]) == 0 with eng.connect() as conn: assert conn.execute(text("SELECT COUNT(*) FROM system_configs " "WHERE config_key LIKE 'kpi_alert_higher_better%'")).scalar() == 0 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 'kpi_alert_higher_better.%'")).scalar() == 0 assert conn.execute(text("SELECT COUNT(*) FROM system_configs " "WHERE config_key = 'kpi_alert_higher_better'")).scalar() == 1 def test_prod_db_name_guard(self, mig_env): 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 # ══════════════════════════════════════════════ # ④ operation_logs 写侧统一带账套（`detail.entity_id`，零 DDL） # 任务单 §二之二④（项目Bot 2026-09-23 18:2x 并入） # ══════════════════════════════════════════════ BACKEND_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OPLOG_REL = os.path.join("app", "utils", "oplog.py") def _operation_log_construction_sites(): """全仓 `OperationLog(` 构造点 → [(相对路径, 行号, 所在语句源码)] 语句源码取「包含该构造调用的完整 Assign/Expr 语句」（多行 dict 字面量整段在内）， 用于断言「写入时带账套」这件事落在**语句级**，而不是碰巧命中了别处的 entity_id。 """ out = [] for rel in ("app", "scripts"): for dp, _dn, fns in os.walk(os.path.join(BACKEND_ROOT, rel)): if "__pycache__" in dp: continue for fn in fns: if not fn.endswith(".py"): continue path = os.path.join(dp, fn) src = open(path, encoding="utf-8").read() tree = ast.parse(src) lines = src.splitlines() stmts = [(n.lineno, n.end_lineno) for n in ast.walk(tree) if isinstance(n, (ast.Assign, ast.Expr, ast.Return)) and getattr(n, "lineno", None) and getattr(n, "end_lineno", None)] for node in ast.walk(tree): if not isinstance(node, ast.Call): continue fn_node = node.func name = getattr(fn_node, "id", None) or getattr(fn_node, "attr", None) if name != "OperationLog": continue cand = [s for s in stmts if s[0] <= node.lineno and s[1] >= node.end_lineno] s, e = min(cand, key=lambda x: x[1] - x[0]) if cand else (node.lineno, node.end_lineno) out.append((os.path.relpath(path, BACKEND_ROOT).replace("\\", "/"), node.lineno, "\n".join(lines[s - 1:e]))) return out class TestAuditWriteSideEntityId: """④ 写侧收口：所有审计行带账套，且集中封装（不加列、不回填历史）""" def test_no_new_raw_OperationLog_sites_lack_entity(self): """源级断言：除集中封装件外，每个构造点的语句里必须出现 entity_id""" sites = _operation_log_construction_sites() assert len(sites) >= 13, f"构造点数量异常（疑似收集失败）: {len(sites)}" missing = [(p, l) for p, l, seg in sites if p != OPLOG_REL and "entity_id" not in seg] assert missing == [], f"以下构造点未带账套（禁止猜账套，必须显式）: {missing}" def test_wrapper_is_single_place_constructing_rows(self): """集中封装：裸构造 OperationLog 只允许出现在 app/utils/oplog.py（防第 28 处再漏）""" wrappers = [(p, l) for p, l, _seg in _operation_log_construction_sites() if p == OPLOG_REL] assert len(wrappers) == 1, f"oplog.py 应恰好 1 处裸构造（log_op 本体）: {wrappers}" def test_log_op_callers_pass_entity_or_scope(self): """所有 `log_op(` 调用点必须传 entity_id= 或 scope=（否则等于留覆盖缺口）""" bad = [] for rel in ("app", "scripts"): for dp, _dn, fns in os.walk(os.path.join(BACKEND_ROOT, rel)): if "__pycache__" in dp: continue for fn in fns: if not fn.endswith(".py"): continue path = os.path.join(dp, fn) tree = ast.parse(open(path, encoding="utf-8").read()) for node in ast.walk(tree): if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "log_op": if not ({"entity_id", "scope"} & {k.arg for k in node.keywords}): bad.append((os.path.relpath(path, BACKEND_ROOT), node.lineno)) assert bad == [], f"log_op 调用点缺账套参数: {bad}" # ── 落库实证 ──────────────────────────────── def test_readback_entity_id_persisted(self, db: Session): """落库实证 ①：log_op 写入 → **raw SELECT 回查数据库**，detail.entity_id 在表里""" log_op(db, "unit_probe_entity", target_type="kpi", target_id=123, detail={"kpi_code": "PROBE_KPI"}, entity_id=2, commit=True) row = db.execute(text("SELECT action, detail FROM operation_logs " "WHERE action = 'unit_probe_entity'")).fetchone() assert row is not None, "审计行未落库" detail = json.loads(row[1]) if isinstance(row[1], str) else row[1] assert detail.get("entity_id") == 2, f"落库 detail 未带账套: {detail}" assert detail.get("kpi_code") == "PROBE_KPI", f"原 detail 内容被破坏: {detail}" def test_readback_kpi_create_endpoint_audit_entity(self, client: TestClient, db: Session): """落库实证 ②：真实写端点（POST /kpis，会话账套 e2）→ 回查审计行 detail.entity_id == 2""" _seed_two_entities(db) create_test_user(db) token = _login_as(client, "testadmin", 2) # 博海 e2 resp = client.post("/api/cma/kpis", headers=auth_header(token), json={ "kpi_code": "F_AUDIT_PROBE_E2", "kpi_name": "审计落库探针", "dimension": "finance", "target_value": 100.0, "unit": "%", "formula": "测试公式", "data_source": "测试系统", "data_owner": "测试管理员", }) assert resp.status_code == 200, resp.text[:300] row = db.execute(text("SELECT detail FROM operation_logs WHERE action = 'create' " "AND target_type = 'kpi' ORDER BY id DESC LIMIT 1")).scalar() assert row is not None, "端点未写审计行" detail = json.loads(row) if isinstance(row, str) else row assert detail.get("entity_id") == 2, f"审计 detail 未带会话账套: {detail}" # ── 形态与哨兵 ────────────────────────────── def test_legacy_json_string_detail_is_not_nested(self, db: Session): """历史写法（detail=json.dumps({...})）归一后不嵌套：entity_id 与原键同一层""" log_op(db, "unit_probe_legacy", target_type="budget", target_id=1, detail=json.dumps({"year": 2026, "mode": "rolling"}, ensure_ascii=False), entity_id=1, commit=True) detail = db.execute(text("SELECT detail FROM operation_logs " "WHERE action = 'unit_probe_legacy'")).scalar() detail = json.loads(detail) if isinstance(detail, str) else detail assert detail == {"year": 2026, "mode": "rolling", "entity_id": 1}, detail def test_global_scope_no_gap_warning(self, db: Session, caplog): """非账套维度（平台/角色级）：写 entity_scope=global，且不产生缺口告警""" with caplog.at_level("WARNING", logger="cma.oplog"): log_op(db, "unit_probe_global", target_type="template", target_id=1, detail={"kpi_code": "T1"}, scope=SCOPE_GLOBAL, commit=True) detail = db.execute(text("SELECT detail FROM operation_logs " "WHERE action = 'unit_probe_global'")).scalar() detail = json.loads(detail) if isinstance(detail, str) else detail assert detail.get("entity_scope") == "global" and "entity_id" not in detail, detail assert NO_ENTITY_TAG not in caplog.text def test_missing_context_warns_sentinel_never_guesses(self, db: Session, caplog): """无账套上下文：留空 + 可 grep 哨兵（绝不猜账套）""" with caplog.at_level("WARNING", logger="cma.oplog"): log_op(db, "unit_probe_unknown", target_type="kpi", target_id=1, detail={"note": "no context"}, commit=True) assert NO_ENTITY_TAG in caplog.text detail = db.execute(text("SELECT detail FROM operation_logs " "WHERE action = 'unit_probe_unknown'")).scalar() detail = json.loads(detail) if isinstance(detail, str) else detail assert "entity_id" not in detail, detail def test_batch_scope_warns_own_sentinel(self, db: Session, caplog): """跨账套批量：entity_scope=batch + 独立哨兵（不猜单账套）""" with caplog.at_level("WARNING", logger="cma.oplog"): log_op(db, "unit_probe_batch", target_type="kpi", detail={"total": 3}, scope=SCOPE_BATCH, commit=True) assert BATCH_NO_ENTITY_TAG in caplog.text detail = db.execute(text("SELECT detail FROM operation_logs " "WHERE action = 'unit_probe_batch'")).scalar() detail = json.loads(detail) if isinstance(detail, str) else detail assert detail.get("entity_scope") == "batch", detail