wecom-fullstack
467472311d
task dama-entity-isolation-sweep-20260923 (sections 2-2 4/5, merged by projectBot 2026-09-23 18:2x)
4) operation_logs write side: every audit write carries the account (no DDL, no column added)
- new app/utils/oplog.py::log_op() single wrapper: sets detail.entity_id; scope=global for
by-design non-entity writes (platform dict / role-level); scope=batch + OPLOG_BATCH_NO_ENTITY
sentinel for cross-account batches; no context -> OPLOG_NO_ENTITY sentinel (never guess)
- 16 remaining construction points converted (data/alert_rules/bi_reports/kpi_causality/
bot_bridge x2/budget_generate/driver_budget/kpis _log x3/templates _log x2/erp_sync);
pre-4 state had 20 sites without entity_id, now 0 outside the wrapper
- read side untouched (knowledge_service still user_id fallback); history not backfilled
5) docs/entity-dimension-data-dictionary.md: 8-table three-state registry
(kpi_templates/driver_factor_templates/scenario_suggestions/bot_bridge_config = non-entity;
kpi_subject_map = inherited via kpi_id; analysis_results/subjects = account dimension,
legacy rows still NULL -> other task; operation_logs = non-entity column + detail.entity_id)
Evidence: RED 2 failed/7 passed (reverted call sites) -> GREEN full 1428 passed; 9 new tests
including two read-back rows; app md5 restore verified identical; no cross-account writes
LOCAL commit only - push is gated (ahead contains the P1 entity-dim batch commit fb4609b)
539 lines
28 KiB
Python
539 lines
28 KiB
Python
"""隔离扫尾 · 同源全局键补漏 + 审计缺口收口 — 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.<entity_id>`（零 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
|