Hermes CI Fix
a0762b367f
P0-A cash.py: scope api_update_plan / api_delete_plan / api_complete_plan /
api_register_payment to session entity (cross-entity id -> 404, aligning with
the file's existing get_entity_id endpoints).
P0-B budget.py: write entity_id on BudgetPlan creation + require the KPI to
belong to the session entity; scope /comparison, /comparison/kpi/{id} and
/deviation-report reads to the session entity.
P1 notifications.py: explicit field whitelist (name/config/enabled) on
update_channel to close mass-assignment.
P1 read leaks scoped to session entity: bot_kpis list, data_quality check/stats
(KPI calibre only; quality-log table has no entity_id column), kpis causality
chain + associate-map (KPI and map ownership), dashboard my-kpis,
customer_dashboard list, kpi_governance audit, budget_generate candidates,
driver_budget history (+ entity_id on its write path), expenses stats
(+ entity_id on its write path).
P2 okr_templates.py: write entity_id on the StrategicMap created by apply.
tests: tests/test_cross_entity_isolation_batch4_20260921.py (30 cases;
RED before fix = 22 failed, GREEN after = 30 passed; full suite 1302 passed /
1 xfailed / 0 failed).
616 lines
24 KiB
Python
616 lines
24 KiB
Python
"""CMA 跨实体隔离收口（第四批）账套隔离回归测试 — 2026-09-21
|
||
|
||
任务：project-20260921-cma-entity-isolation-batch4（项目Bot 派单；来源 OpenCode 周审 20260921）
|
||
缺陷（本单范围）：
|
||
P0-A cash.py 4 端点无账套校验（跨租户改/删/完成/回款）
|
||
PUT /api/cma/cash/plans/{plan_id}
|
||
DELETE /api/cma/cash/plans/{plan_id}
|
||
POST /api/cma/cash/plans/{plan_id}/complete
|
||
POST /api/cma/cash/receivables/{plan_id}/payment
|
||
P0-B budget.py create_budget_plan 建行未写 entity_id（落默认 1，错归属）
|
||
P1 notifications.py update_channel 无字段白名单（mass-assignment）
|
||
P1 跨实体只读泄漏点（bot_kpis / data_quality / kpis / dashboard /
|
||
customer_dashboard / kpi_governance / budget_generate / budget 读 /
|
||
driver_budget / expenses）
|
||
P2 okr_templates.py apply_okr_template 建 StrategicMap 未写 entity_id
|
||
|
||
每条均「跨账套负例 + 本账套正对照」；改前 RED 原文见回报 §三。
|
||
本文件只跑内存 SQLite（tests/conftest.py 机制），零生产库 / 零生产 Redis 连接。
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import (
|
||
BudgetPlan,
|
||
CashPlan,
|
||
Entity,
|
||
ExpenseReimbursement,
|
||
KPIDefinition,
|
||
KpiDataQualityLog,
|
||
NotificationChannel,
|
||
OKRTemplate,
|
||
StrategicMap,
|
||
DriverFactorBudget,
|
||
)
|
||
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
|
||
|
||
CASH = "/api/cma/cash"
|
||
BUDGET = "/api/cma/budget"
|
||
NOTIF = "/api/cma/notifications"
|
||
KPIS = "/api/cma/kpis"
|
||
DQ = "/api/cma/data-quality"
|
||
DASH = "/api/cma/dashboard"
|
||
CUST = "/api/cma/customer-dashboard"
|
||
GOV = "/api/cma/kpi"
|
||
OKRT = "/api/cma/okr-templates"
|
||
EXPENSES = "/api/cma/expenses"
|
||
|
||
BOT_SRC = "batch4-bot"
|
||
|
||
# 两账套资源编码前缀（kpi_code 库内唯一）
|
||
E1_KPI = "ZZ_B4_E1_KPI_001"
|
||
E2_KPI = "ZZ_B4_E2_KPI_001"
|
||
E1_CUST = "ZZ_B4_E1_CUST_001"
|
||
E2_CUST = "ZZ_B4_E2_CUST_001"
|
||
E1_BOT = "ZZ_B4_E1_BOT_001"
|
||
E2_BOT = "ZZ_B4_E2_BOT_001"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 夹具
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _ensure_entity(db: Session, eid: int, name: str) -> Entity:
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=name, short_name=name, status="active")
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _mk_kpi(db: Session, code: str, eid: int, dimension: str = "finance", **kw) -> KPIDefinition:
|
||
kpi = KPIDefinition(
|
||
entity_id=eid,
|
||
kpi_code=code,
|
||
kpi_name=f"{code}-名称",
|
||
dimension=dimension,
|
||
category="revenue_growth",
|
||
formula="-",
|
||
data_source_type="excel",
|
||
data_source="Excel导入",
|
||
data_owner="财务部",
|
||
frequency="monthly",
|
||
unit="元",
|
||
target_value=0,
|
||
kpi_level="operational",
|
||
status="active",
|
||
**kw,
|
||
)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
def _mk_plan(db: Session, eid: int, amount: float = 10.0, plan_type: str = "receive",
|
||
status: str = "pending", paid_amount: float = 0.0, **kw) -> CashPlan:
|
||
from datetime import datetime
|
||
plan = CashPlan(
|
||
entity_id=eid, plan_type=plan_type, amount=amount,
|
||
plan_date=datetime(2026, 9, 30), counterparty=f"e{eid}客户",
|
||
description=f"e{eid}计划", status=status, source="manual",
|
||
paid_amount=paid_amount, **kw,
|
||
)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
def _mk_map(db: Session, eid: int, title: str) -> StrategicMap:
|
||
m = StrategicMap(entity_id=eid, title=title, status="draft",
|
||
dimensions=[], canvas_data={"connections": []})
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
return m
|
||
|
||
|
||
def _mk_channel(db: Session, name: str) -> NotificationChannel:
|
||
ch = NotificationChannel(name=name, channel_type="wecom", config={"webhook": "x"}, enabled=True)
|
||
db.add(ch)
|
||
db.commit()
|
||
db.refresh(ch)
|
||
return ch
|
||
|
||
|
||
def _mk_budget(db: Session, kpi_id: int, eid: int, period: str = "2026-09", value: float = 100.0):
|
||
bp = BudgetPlan(entity_id=eid, kpi_id=kpi_id, period=period, budget_value=value,
|
||
budget_year=2026, budget_month=9, version="v1.0", status="active")
|
||
db.add(bp)
|
||
db.commit()
|
||
db.refresh(bp)
|
||
return bp
|
||
|
||
|
||
def _mk_template(db: Session, name: str) -> OKRTemplate:
|
||
t = OKRTemplate(name=name, description="批4测试模板", dimension="finance",
|
||
layer="level1", preset_krs=[{"name": "KR测试", "target_value": 100, "weight": 50}],
|
||
source="system", is_active=1)
|
||
db.add(t)
|
||
db.commit()
|
||
db.refresh(t)
|
||
return t
|
||
|
||
|
||
def _mk_reimb(db: Session, eid: int, no: str, amount: float = 100.0):
|
||
from datetime import datetime
|
||
r = ExpenseReimbursement(entity_id=eid, reimb_no=no, applicant="张三", department="财务部",
|
||
expense_type="office", title=f"{no}-报销", amount=amount,
|
||
expense_date=datetime(2026, 9, 10), status="pending",
|
||
check_result="pass")
|
||
db.add(r)
|
||
db.commit()
|
||
db.refresh(r)
|
||
return r
|
||
|
||
|
||
def _login(client: TestClient, entity_id: int) -> str:
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin",
|
||
"password": TEST_USER_PASSWORD,
|
||
"entity_id": entity_id,
|
||
})
|
||
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
|
||
return resp.json()["token"]
|
||
|
||
|
||
def _seed_two_entities(db: Session):
|
||
_ensure_entity(db, 1, "测试企业1")
|
||
_ensure_entity(db, 2, "测试企业2")
|
||
create_test_user(db)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P0-A. cash.py 4 端点按账套（跨账套 404）
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_cash_update_plan_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
other = _mk_plan(db, 2, amount=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{CASH}/plans/{other.id}", headers=auth_header(token1), json={"amount": 999.0})
|
||
assert r.status_code == 404, f"跨账套改计划应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
|
||
assert row is not None, "对端计划被删除"
|
||
assert row.amount == 50.0, f"跨账套改计划生效了：amount={row.amount}"
|
||
|
||
|
||
def test_cash_update_plan_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
mine = _mk_plan(db, 1, amount=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{CASH}/plans/{mine.id}", headers=auth_header(token1), json={"amount": 88.0})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first().amount == 88.0
|
||
|
||
|
||
def test_cash_delete_plan_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
other = _mk_plan(db, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{CASH}/plans/{other.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套删计划应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
assert db.query(CashPlan).filter(CashPlan.id == other.id).first() is not None, \
|
||
"跨账套删除竟生效（对端计划已消失）"
|
||
|
||
|
||
def test_cash_delete_plan_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
mine = _mk_plan(db, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{CASH}/plans/{mine.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first() is None
|
||
|
||
|
||
def test_cash_complete_plan_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
other = _mk_plan(db, 2, amount=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{CASH}/plans/{other.id}/complete", headers=auth_header(token1), json={})
|
||
assert r.status_code == 404, f"跨账套完成计划应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
|
||
assert row.status == "pending", f"跨账套完成生效了：status={row.status}"
|
||
assert (row.paid_amount or 0) == 0, f"跨账套完成写入了 paid_amount={row.paid_amount}"
|
||
|
||
|
||
def test_cash_complete_plan_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
mine = _mk_plan(db, 1, amount=50.0)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{CASH}/plans/{mine.id}/complete", headers=auth_header(token1), json={})
|
||
assert r.status_code == 200, r.text[:300]
|
||
db.expire_all()
|
||
row = db.query(CashPlan).filter(CashPlan.id == mine.id).first()
|
||
assert row.status == "completed" and row.paid_amount == 50.0
|
||
|
||
|
||
def test_cash_register_payment_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
other = _mk_plan(db, 2, amount=50.0, plan_type="receive")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{CASH}/receivables/{other.id}/payment", headers=auth_header(token1),
|
||
json={"amount": 5.0, "paid_date": "2026-09-20"})
|
||
assert r.status_code == 404, f"跨账套回款登记应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
db.expire_all()
|
||
row = db.query(CashPlan).filter(CashPlan.id == other.id).first()
|
||
assert (row.paid_amount or 0) == 0, f"跨账套回款登记生效了：paid_amount={row.paid_amount}"
|
||
assert row.status == "pending", f"跨账套回款登记改了 status={row.status}"
|
||
|
||
|
||
def test_cash_register_payment_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
mine = _mk_plan(db, 1, amount=50.0, plan_type="receive")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{CASH}/receivables/{mine.id}/payment", headers=auth_header(token1),
|
||
json={"amount": 5.0, "paid_date": "2026-09-20"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
db.expire_all()
|
||
assert db.query(CashPlan).filter(CashPlan.id == mine.id).first().paid_amount == 5.0
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P0-B. budget.py 建预算写会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_create_budget_plan_writes_session_entity(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.post(f"{BUDGET}/plans", headers=auth_header(token2),
|
||
json={"kpi_id": k2.id, "period": "2026-09", "budget_value": 123.0})
|
||
assert r.status_code == 200, r.text[:300]
|
||
new_id = r.json()["id"]
|
||
|
||
db.expire_all()
|
||
row = db.query(BudgetPlan).filter(BudgetPlan.id == new_id).first()
|
||
assert row is not None, "预算未落库"
|
||
assert row.entity_id == 2, (
|
||
f"会话账套=2 新建预算落库归属 entity_id={row.entity_id}（缺陷：未写归属→模型默认 1）"
|
||
)
|
||
|
||
|
||
def test_create_budget_plan_e1_control(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{BUDGET}/plans", headers=auth_header(token1),
|
||
json={"kpi_id": k1.id, "period": "2026-09", "budget_value": 55.0})
|
||
assert r.status_code == 200, r.text[:300]
|
||
db.expire_all()
|
||
row = db.query(BudgetPlan).filter(BudgetPlan.id == r.json()["id"]).first()
|
||
assert row.entity_id == 1, f"e1 会话建预算 entity_id={row.entity_id}"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P1. notifications.py 字段白名单
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_update_channel_rejects_non_whitelist_fields(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
ch = _mk_channel(db, "批4渠道")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{NOTIF}/channels/{ch.id}", headers=auth_header(token1), json={
|
||
"name": "批4渠道-改名",
|
||
"channel_type": "email",
|
||
"id": 999999,
|
||
"created_at": None,
|
||
})
|
||
assert r.status_code in (200, 400), f"非白名单字段应被忽略或 400，实际 {r.status_code} {r.text[:200]}"
|
||
db.expire_all()
|
||
row = db.query(NotificationChannel).filter(NotificationChannel.id == ch.id).first()
|
||
assert row.name == "批4渠道-改名", "白名单字段 name 未更新"
|
||
assert row.channel_type == "wecom", f"非白名单字段 channel_type 被篡改 → {row.channel_type}"
|
||
assert row.id == ch.id, "主键被篡改"
|
||
|
||
|
||
def test_update_channel_allows_business_fields(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
ch = _mk_channel(db, "批4渠道2")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{NOTIF}/channels/{ch.id}", headers=auth_header(token1), json={
|
||
"name": "批4渠道2-新", "config": {"webhook": "y"}, "enabled": False,
|
||
})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
row = db.query(NotificationChannel).filter(NotificationChannel.id == ch.id).first()
|
||
assert row.name == "批4渠道2-新"
|
||
assert row.enabled in (False, 0)
|
||
assert (row.config or {}).get("webhook") == "y"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P1. 跨实体只读泄漏点
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_bot_kpis_list_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_BOT, 1, bot_source=BOT_SRC)
|
||
_mk_kpi(db, E2_BOT, 2, bot_source=BOT_SRC)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"/api/cma/bot-kpis?source={BOT_SRC}", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
codes = {k["kpi_code"] for k in r.json()["kpis"]}
|
||
assert E1_BOT not in codes, f"e2 会话读到 e1 Bot KPI: {sorted(codes)}"
|
||
assert E2_BOT in codes, f"e2 会话读不到自己的 Bot KPI: {sorted(codes)}"
|
||
|
||
|
||
def test_data_quality_check_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
_mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{DQ}/check", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
body = r.json()
|
||
assert body["total_kpis"] == 1, f"e2 会话扫描了 {body['total_kpis']} 个 KPI（应只扫本账套 1 个）"
|
||
|
||
db.expire_all()
|
||
leaked = db.query(KpiDataQualityLog).filter(KpiDataQualityLog.kpi_id == k1.id).count()
|
||
assert leaked == 0, f"e2 会话的质量检查为 e1 的 KPI 写了 {leaked} 条日志"
|
||
|
||
|
||
def test_data_quality_stats_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_KPI, 1)
|
||
_mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{DQ}/stats", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["total_kpis"] == 1, f"e2 统计口径含他账套：total_kpis={r.json()['total_kpis']}"
|
||
|
||
|
||
def test_kpis_causality_chain_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k2.id}/causality-chain", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套读因果链应 404，实际 {r.status_code} {r.text[:200]}"
|
||
assert E2_KPI not in r.text, "跨账套响应体泄漏了对端 kpi_code"
|
||
|
||
|
||
def test_kpis_causality_chain_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{KPIS}/{k1.id}/causality-chain", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["kpi"]["kpi_code"] == E1_KPI
|
||
|
||
|
||
def test_kpis_associate_map_cross_entity_kpi_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
m1 = _mk_map(db, 1, "e1地图")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{KPIS}/{k2.id}/associate-map", headers=auth_header(token1), json={"map_id": m1.id})
|
||
assert r.status_code == 404, f"跨账套改 KPI 归属应 404，实际 {r.status_code} {r.text[:200]}"
|
||
db.expire_all()
|
||
assert db.query(KPIDefinition).filter(KPIDefinition.id == k2.id).first().map_id is None, \
|
||
"跨账套 associate-map 竟改动了对端 KPI"
|
||
|
||
|
||
def test_kpis_associate_map_cross_entity_map_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
m2 = _mk_map(db, 2, "e2地图")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{KPIS}/{k1.id}/associate-map", headers=auth_header(token1), json={"map_id": m2.id})
|
||
assert r.status_code == 404, f"本账套 KPI 关联对端地图应 404，实际 {r.status_code} {r.text[:200]}"
|
||
db.expire_all()
|
||
assert db.query(KPIDefinition).filter(KPIDefinition.id == k1.id).first().map_id is None
|
||
|
||
|
||
def test_kpis_associate_map_own_entity_ok(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
m1 = _mk_map(db, 1, "e1地图")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{KPIS}/{k1.id}/associate-map", headers=auth_header(token1), json={"map_id": m1.id})
|
||
assert r.status_code == 200, r.text[:200]
|
||
db.expire_all()
|
||
assert db.query(KPIDefinition).filter(KPIDefinition.id == k1.id).first().map_id == m1.id
|
||
|
||
|
||
def test_dashboard_my_kpis_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_KPI, 1)
|
||
_mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{DASH}/my-kpis", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert E1_KPI not in r.text, "e2 会话在 /dashboard/my-kpis 看到 e1 KPI 数据"
|
||
assert E2_KPI in r.text, "e2 会话看不到自己的 KPI"
|
||
|
||
|
||
def test_customer_dashboard_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_CUST, 1, dimension="customer")
|
||
_mk_kpi(db, E2_CUST, 2, dimension="customer")
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(CUST, headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert E1_CUST not in r.text, "e2 会话在客户看板看到 e1 的客户维度 KPI"
|
||
assert E2_CUST in r.text, "e2 会话看不到自己的客户维度 KPI"
|
||
|
||
|
||
def test_kpi_governance_audit_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_KPI, 1)
|
||
_mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{GOV}/governance/audit", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
body = r.json()
|
||
assert body["total"] == 1, f"e2 治理审计扫了 {body['total']} 个 KPI（应只含本账套 1 个）"
|
||
assert E1_KPI not in r.text, "e2 治理审计泄漏 e1 KPI"
|
||
|
||
|
||
def test_budget_generate_candidates_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_KPI, 1)
|
||
_mk_kpi(db, E2_KPI, 2)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{BUDGET}/kpi-budget-candidates?year=2026", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert E1_KPI not in r.text, "e2 会话在预算候选里看到 e1 KPI"
|
||
assert E2_KPI in r.text, "e2 会话看不到自己的预算候选 KPI"
|
||
|
||
|
||
def test_budget_comparison_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
_mk_budget(db, k1.id, 1)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{BUDGET}/comparison?year=2026", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert r.json()["total_kpis"] == 0, (
|
||
f"e2 会话的预算对比含他账套预算：total_kpis={r.json()['total_kpis']}"
|
||
)
|
||
|
||
|
||
def test_budget_kpi_comparison_cross_entity_404(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.get(f"{BUDGET}/comparison/kpi/{k2.id}?year=2026", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套读预算对比应 404，实际 {r.status_code} {r.text[:200]}"
|
||
|
||
|
||
def test_budget_deviation_report_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_kpi(db, E1_KPI, 1)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.get(f"{BUDGET}/deviation-report?period=2026-09", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text[:200]
|
||
assert E1_KPI not in r.text, "e2 会话在预算差异报告看到 e1 KPI"
|
||
assert r.json()["summary"]["total_kpis"] == 0, (
|
||
f"e2 差异报告含他账套 KPI：total_kpis={r.json()['summary']['total_kpis']}"
|
||
)
|
||
|
||
|
||
def test_driver_budget_write_and_history_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
token2 = _login(client, 2)
|
||
payload = {
|
||
"name": "批4驱动预算",
|
||
"formula_text": "a * b",
|
||
"factor_defs": [{"key": "a", "label": "A", "default": 2}, {"key": "b", "label": "B", "default": 3}],
|
||
"factors": {},
|
||
}
|
||
r = client.post(f"{BUDGET}/driver/calculate", headers=auth_header(token2), json=payload)
|
||
assert r.status_code == 200, r.text[:300]
|
||
rec_id = r.json()["id"]
|
||
|
||
db.expire_all()
|
||
rec = db.query(DriverFactorBudget).filter(DriverFactorBudget.id == rec_id).first()
|
||
assert rec is not None and rec.entity_id == 2, (
|
||
f"e2 会话的驱动因子预算落库归属 entity_id={getattr(rec, 'entity_id', None)}（应为 2）"
|
||
)
|
||
|
||
# 历史列表只含本账套
|
||
r1 = client.get(f"{BUDGET}/driver/history", headers=auth_header(_login(client, 1)))
|
||
assert r1.status_code == 200, r1.text[:200]
|
||
assert "批4驱动预算" not in r1.text, "e1 会话在驱动因子历史里看到 e2 记录"
|
||
r2 = client.get(f"{BUDGET}/driver/history", headers=auth_header(token2))
|
||
assert "批4驱动预算" in r2.text, "e2 会话看不到自己的驱动因子历史"
|
||
|
||
|
||
def test_expenses_write_and_stats_scoped(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
_mk_reimb(db, 1, "ZZ_B4_E1_BX001", amount=111.0)
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.post(f"{EXPENSES}/reimbursements", headers=auth_header(token2), json={
|
||
"applicant": "李四", "department": "财务部", "expense_type": "office",
|
||
"title": "批4 e2 报销", "amount": 222.0, "expense_date": "2026-09-15",
|
||
})
|
||
assert r.status_code == 200, r.text[:300]
|
||
new_id = r.json()["data"]["id"]
|
||
|
||
db.expire_all()
|
||
row = db.query(ExpenseReimbursement).filter(ExpenseReimbursement.id == new_id).first()
|
||
assert row.entity_id == 2, f"e2 会话报销单落库归属 entity_id={row.entity_id}（应为 2）"
|
||
|
||
r2 = client.get(f"{EXPENSES}/stats", headers=auth_header(token2))
|
||
assert r2.status_code == 200, r2.text[:200]
|
||
counts = r2.json()["counts"]
|
||
assert counts["total"] == 1, (
|
||
f"e2 费用看板统计含他账套：total={counts['total']}（应只统计本账套 1 条）"
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# P2. okr_templates 应用模板写会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_apply_okr_template_writes_session_entity(client: TestClient, db: Session):
|
||
_seed_two_entities(db)
|
||
t = _mk_template(db, "批4模板")
|
||
token2 = _login(client, 2)
|
||
|
||
r = client.post(f"{OKRT}/{t.id}/apply", headers=auth_header(token2), json={})
|
||
assert r.status_code == 200, r.text[:300]
|
||
map_id = r.json().get("map_id") or (r.json().get("data") or {}).get("id")
|
||
|
||
db.expire_all()
|
||
m = None
|
||
if map_id:
|
||
m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
|
||
if m is None:
|
||
m = db.query(StrategicMap).order_by(StrategicMap.id.desc()).first()
|
||
assert m is not None, f"应用模板未创建地图：{r.text[:300]}"
|
||
assert m.entity_id == 2, (
|
||
f"会话账套=2 应用 OKR 模板生成的地图归属 entity_id={m.entity_id}（缺陷：未传 entity_id→默认 1）"
|
||
)
|