Hermes CI Fix
ad953ac759
ci/woodpecker/push/main Pipeline failed
Scope: backend/app/api/{alerts,action_plans,budget,cost}.py + tests
Task: 0-P1-project-20260918-entity-isolation-batch2
425 lines
18 KiB
Python
425 lines
18 KiB
Python
"""CMA 跨实体隔离收口（第二批）账套隔离回归测试 — 2026-09-18
|
||
|
||
任务：0-P1-project-20260918-entity-isolation-batch2（项目Bot 派单，任总 2026-09-18 17:59 放行）
|
||
缺陷（本单范围）：
|
||
1. 写型·列表批处理
|
||
- POST /api/cma/budget/roll-forward 全账套 active 预算一起删最早月/加新月
|
||
- GET /api/cma/alerts/check-timeout 全账套 pending 超时预警一起升级为 red
|
||
2. 写型/读型·按 id 直取（无归属校验 → 跨账套可达）
|
||
- POST /api/cma/alerts/{alert_id}/process | /escalate | /create-action-plan
|
||
- PUT /api/cma/action-plans/{plan_id} | DELETE /api/cma/action-plans/{plan_id}
|
||
3. 读型·已在发生
|
||
- GET /api/cma/cost/abc/activities 全表取（生产 10 行全属 entity2）→ 酣客会话可见博海 ABC 作业中心
|
||
|
||
本用例覆盖（每条均双向 + 同账套正对照）：
|
||
- 写型批处理：会话账套触发只影响本账套行，对端账套行前后逐字段不变；
|
||
- 按 id 直取：跨账套 id → 404（对齐同文件既有语义），本账套 id → 正常；且跨账套请求后对端行未被修改/删除；
|
||
- 读型：会话账套只返回本账套行，不含对端行。
|
||
|
||
修复前（旧码）本文件在下列断言上必失败 —— 即「改前/改后对照」中的改前证据（见回报 §三）。
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from datetime import datetime, timedelta
|
||
import json
|
||
|
||
from app.models import (
|
||
AbcActivity,
|
||
ActionPlan,
|
||
BudgetPlan,
|
||
Entity,
|
||
KPIAlert,
|
||
KPIDefinition,
|
||
SystemConfig,
|
||
)
|
||
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
|
||
|
||
ALERTS = "/api/cma/alerts"
|
||
BUDGET = "/api/cma/budget"
|
||
COST = "/api/cma/cost"
|
||
PLANS = "/api/cma/action-plans"
|
||
|
||
# 账套 1/2 各自的资源编码前缀（kpi_code / activity_code 库内唯一）
|
||
E1_KPI = "ZZ_B2_E1_KPI_001"
|
||
E2_KPI = "ZZ_B2_E2_KPI_001"
|
||
E1_ACT = "ZZ_B2_E1_ACT_001"
|
||
E2_ACT = "ZZ_B2_E2_ACT_001"
|
||
E2_ACT2 = "ZZ_B2_E2_ACT_002"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 夹具
|
||
# ──────────────────────────────────────────────
|
||
|
||
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) -> 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="元",
|
||
target_value=0,
|
||
kpi_level="operational",
|
||
status="active",
|
||
)
|
||
db.add(kpi)
|
||
db.commit()
|
||
db.refresh(kpi)
|
||
return kpi
|
||
|
||
|
||
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 _set_rolling(db: Session) -> None:
|
||
"""滚动预算模式（全局 SystemConfig，配置值为 JSON 文本，与 /budget/config 写入口径一致）"""
|
||
cfg = db.query(SystemConfig).filter(SystemConfig.config_key == "budget_mode").first()
|
||
val = json.dumps({"mode": "rolling", "rolling_months": 12}, ensure_ascii=False)
|
||
if cfg:
|
||
cfg.config_value = val
|
||
else:
|
||
db.add(SystemConfig(config_key="budget_mode", config_value=val,
|
||
description="预算模式: fixed=固定预算, rolling=滚动预算"))
|
||
db.commit()
|
||
|
||
|
||
def _seed_plans(db: Session, kpi_id: int, eid: int, months: int, base: float) -> None:
|
||
for m in range(1, months + 1):
|
||
db.add(BudgetPlan(
|
||
entity_id=eid, kpi_id=kpi_id, period=f"2026-{m:02d}",
|
||
budget_value=base + m, budget_year=2026, budget_month=m,
|
||
version="v1.0", status="active",
|
||
))
|
||
db.commit()
|
||
|
||
|
||
def _plan_snapshot(db: Session, eid: int) -> list:
|
||
"""某账套全部预算行（按 period 排序，逐字段快照）"""
|
||
rows = (db.query(BudgetPlan)
|
||
.filter(BudgetPlan.entity_id == eid)
|
||
.order_by(BudgetPlan.period)
|
||
.all())
|
||
return [
|
||
(r.id, r.kpi_id, r.period, round(r.budget_value, 4), r.budget_year,
|
||
r.budget_month, r.status, r.version)
|
||
for r in rows
|
||
]
|
||
|
||
|
||
def _mk_alert(db: Session, kpi_id: int, eid: int, **kw) -> KPIAlert:
|
||
alert = KPIAlert(
|
||
entity_id=eid,
|
||
kpi_id=kpi_id,
|
||
alert_level=kw.get("alert_level", "yellow"),
|
||
alert_message=kw.get("alert_message", f"e{eid} 测试预警"),
|
||
status=kw.get("status", "pending"),
|
||
created_at=kw.get("created_at", datetime.now()),
|
||
)
|
||
db.add(alert)
|
||
db.commit()
|
||
db.refresh(alert)
|
||
return alert
|
||
|
||
|
||
def _mk_plan(db: Session, kpi_id: int, eid: int, title: str) -> ActionPlan:
|
||
plan = ActionPlan(entity_id=eid, kpi_id=kpi_id, title=title,
|
||
assignee="张三", priority="medium", status="pending", progress=0)
|
||
db.add(plan)
|
||
db.commit()
|
||
db.refresh(plan)
|
||
return plan
|
||
|
||
|
||
def _mk_activity(db: Session, code: str, eid: int) -> AbcActivity:
|
||
act = AbcActivity(
|
||
entity_id=eid, activity_code=code, activity_name=f"{code}-作业",
|
||
cost_driver="动因", driver_unit="次", total_cost=1000.0, driver_volume=10,
|
||
driver_rate=100.0, status="active",
|
||
)
|
||
db.add(act)
|
||
db.commit()
|
||
db.refresh(act)
|
||
return act
|
||
|
||
|
||
def _seed_two_entities(db: Session):
|
||
_ensure_entity(db, 1, "酣客")
|
||
_ensure_entity(db, 2, "博海")
|
||
create_test_user(db)
|
||
k1 = _mk_kpi(db, E1_KPI, 1)
|
||
k2 = _mk_kpi(db, E2_KPI, 2)
|
||
return k1, k2
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 1. 写型·列表批处理：budget/roll-forward 按会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_roll_forward_only_touches_session_entity(client: TestClient, db: Session):
|
||
"""酣客会话滚动延展 → 只动 e1 预算，e2 预算前后逐字段不变（含行数/最早月/新月份）"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
_set_rolling(db)
|
||
_seed_plans(db, k1.id, 1, months=12, base=1000.0) # e1: 12 个月
|
||
_seed_plans(db, k2.id, 2, months=3, base=2000.0) # e2: 3 个月
|
||
|
||
before_e1 = _plan_snapshot(db, 1)
|
||
before_e2 = _plan_snapshot(db, 2)
|
||
assert len(before_e1) == 12 and len(before_e2) == 3
|
||
|
||
token1 = _login(client, 1)
|
||
r = client.post(f"{BUDGET}/roll-forward", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text
|
||
data = r.json()
|
||
|
||
after_e1 = _plan_snapshot(db, 1)
|
||
after_e2 = _plan_snapshot(db, 2)
|
||
|
||
# 响应只含本账套 KPI
|
||
assert len(data["rolled_kpis"]) == 1, f"应只延展 1 个 KPI，实际 {data['rolled_kpis']}"
|
||
assert data["rolled_kpis"][0]["kpi_id"] == k1.id
|
||
|
||
# e1：删 1 行加 1 行（行数不变，最早月被删）
|
||
assert len(after_e1) == 12
|
||
e1_periods = [p for _, _, p, *_ in after_e1]
|
||
assert "2026-01" not in e1_periods
|
||
assert data["rolled_kpis"][0]["added_period"] in e1_periods
|
||
|
||
# e2：逐字段完全未变（这是缺陷核心：跨账套批量写）
|
||
assert after_e2 == before_e2, (
|
||
f"entity2 预算被跨账套修改\nbefore={before_e2}\nafter={after_e2}"
|
||
)
|
||
|
||
|
||
def test_roll_forward_reverse_direction(client: TestClient, db: Session):
|
||
"""博海会话滚动延展 → 只动 e2 预算，e1 预算前后不变"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
_set_rolling(db)
|
||
_seed_plans(db, k1.id, 1, months=12, base=1000.0)
|
||
_seed_plans(db, k2.id, 2, months=3, base=2000.0)
|
||
|
||
before_e1 = _plan_snapshot(db, 1)
|
||
|
||
token2 = _login(client, 2)
|
||
r = client.post(f"{BUDGET}/roll-forward", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text
|
||
data = r.json()
|
||
|
||
assert len(data["rolled_kpis"]) == 1
|
||
assert data["rolled_kpis"][0]["kpi_id"] == k2.id
|
||
assert _plan_snapshot(db, 1) == before_e1, "entity1 预算被 e2 会话跨账套修改"
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 2. 写型·列表批处理：alerts/check-timeout 按会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _alert_state(db: Session, eid: int):
|
||
rows = db.query(KPIAlert).filter(KPIAlert.entity_id == eid).order_by(KPIAlert.id).all()
|
||
return [(a.id, a.status, a.alert_level) for a in rows]
|
||
|
||
|
||
def test_check_timeout_only_upgrades_session_entity(client: TestClient, db: Session):
|
||
"""酣客会话超时检测 → 只升级 e1 的 pending 预警，e2 保持 pending/yellow"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
old = datetime.now() - timedelta(hours=25)
|
||
_mk_alert(db, k1.id, 1, created_at=old)
|
||
_mk_alert(db, k1.id, 1, created_at=old)
|
||
_mk_alert(db, k2.id, 2, created_at=old)
|
||
_mk_alert(db, k2.id, 2, created_at=old)
|
||
|
||
token1 = _login(client, 1)
|
||
r = client.get(f"{ALERTS}/check-timeout", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text
|
||
body = r.json()
|
||
assert body["total_timeout"] == 2, f"应只命中本账套 2 条，实际 {body}"
|
||
assert body["upgraded_count"] == 2
|
||
|
||
# e1 升级为 red/processing
|
||
for _id, status, level in _alert_state(db, 1):
|
||
assert (status, level) == ("processing", "red"), f"e1 预警 {_id} 未升级: {status}/{level}"
|
||
# e2 完全未动（缺陷核心：一次调用把所有账套升级）
|
||
for _id, status, level in _alert_state(db, 2):
|
||
assert (status, level) == ("pending", "yellow"), f"e2 预警 {_id} 被跨账套升级: {status}/{level}"
|
||
|
||
|
||
def test_check_timeout_reverse_direction(client: TestClient, db: Session):
|
||
"""博海会话超时检测 → 只升级 e2，e1 保持 pending/yellow"""
|
||
k1, k2 = _seed_two_entities(db)
|
||
old = datetime.now() - timedelta(hours=25)
|
||
_mk_alert(db, k1.id, 1, created_at=old)
|
||
_mk_alert(db, k2.id, 2, created_at=old)
|
||
|
||
token2 = _login(client, 2)
|
||
r = client.get(f"{ALERTS}/check-timeout", headers=auth_header(token2))
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["upgraded_count"] == 1
|
||
|
||
for _id, status, level in _alert_state(db, 1):
|
||
assert (status, level) == ("pending", "yellow"), f"e1 预警 {_id} 被 e2 会话跨账套升级"
|
||
for _id, status, level in _alert_state(db, 2):
|
||
assert (status, level) == ("processing", "red")
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 3. 按 id 直取：alerts 三条（跨账套 → 404，同账套 → 正常）
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_alert_process_entity_scoped(client: TestClient, db: Session):
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_alert(db, k1.id, 1)
|
||
other = _mk_alert(db, k2.id, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
# 跨账套 → 404（对齐 /resolve 既有语义：查无此记录）
|
||
r = client.post(f"{ALERTS}/{other.id}/process", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 404, f"跨账套 process 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
db.refresh(other)
|
||
assert other.status == "pending", "跨账套 process 竟改动了对端账套记录"
|
||
|
||
# 同账套 → 正常
|
||
r = client.post(f"{ALERTS}/{mine.id}/process", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 200, r.text
|
||
db.refresh(mine)
|
||
assert mine.status == "processing" and mine.assignee == "李四"
|
||
|
||
|
||
def test_alert_escalate_entity_scoped(client: TestClient, db: Session):
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_alert(db, k1.id, 1)
|
||
other = _mk_alert(db, k2.id, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{ALERTS}/{other.id}/escalate", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 404, f"跨账套 escalate 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
db.refresh(other)
|
||
assert other.alert_level == "yellow", "跨账套 escalate 竟升级了对端账套预警"
|
||
|
||
r = client.post(f"{ALERTS}/{mine.id}/escalate", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 200, r.text
|
||
db.refresh(mine)
|
||
assert mine.alert_level == "red"
|
||
|
||
|
||
def test_alert_create_action_plan_entity_scoped(client: TestClient, db: Session):
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_alert(db, k1.id, 1)
|
||
other = _mk_alert(db, k2.id, 2)
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.post(f"{ALERTS}/{other.id}/create-action-plan", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 404, f"跨账套 create-action-plan 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
assert db.query(ActionPlan).filter(ActionPlan.alert_id == other.id).count() == 0, \
|
||
"跨账套 create-action-plan 竟为对端账套预警建了计划"
|
||
db.refresh(other)
|
||
assert other.action_plan_linked_id is None
|
||
|
||
r = client.post(f"{ALERTS}/{mine.id}/create-action-plan", headers=auth_header(token1),
|
||
json={"assignee": "李四"})
|
||
assert r.status_code == 200, r.text
|
||
db.refresh(mine)
|
||
assert mine.action_plan_linked_id is not None
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 4. 按 id 直取：action-plans 改/删（跨账套 → 404，同账套 → 正常）
|
||
# ──────────────────────────────────────────────
|
||
|
||
def test_update_plan_entity_scoped(client: TestClient, db: Session):
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_plan(db, k1.id, 1, "e1 计划")
|
||
other = _mk_plan(db, k2.id, 2, "e2 计划")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.put(f"{PLANS}/{other.id}", headers=auth_header(token1),
|
||
json={"title": "被跨账套改名", "status": "completed"})
|
||
assert r.status_code == 404, f"跨账套 update 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
db.refresh(other)
|
||
assert other.title == "e2 计划" and other.status == "pending", "跨账套 update 竟改动了对端计划"
|
||
|
||
r = client.put(f"{PLANS}/{mine.id}", headers=auth_header(token1),
|
||
json={"title": "本账套改名", "status": "in_progress"})
|
||
assert r.status_code == 200, r.text
|
||
db.refresh(mine)
|
||
assert mine.title == "本账套改名" and mine.status == "in_progress"
|
||
|
||
|
||
def test_delete_plan_entity_scoped(client: TestClient, db: Session):
|
||
k1, k2 = _seed_two_entities(db)
|
||
mine = _mk_plan(db, k1.id, 1, "e1 计划")
|
||
other = _mk_plan(db, k2.id, 2, "e2 计划")
|
||
token1 = _login(client, 1)
|
||
|
||
r = client.delete(f"{PLANS}/{other.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨账套 delete 应 404，实际 {r.status_code} {r.text[:200]}"
|
||
assert db.query(ActionPlan).filter(ActionPlan.id == other.id).count() == 1, \
|
||
"跨账套 delete 竟删除了对端账套计划"
|
||
|
||
r = client.delete(f"{PLANS}/{mine.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text
|
||
assert db.query(ActionPlan).filter(ActionPlan.id == mine.id).count() == 0
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 5. 读型：cost/abc/activities 按会话账套
|
||
# ──────────────────────────────────────────────
|
||
|
||
def _abc_codes(client: TestClient, token: str) -> set:
|
||
r = client.get(f"{COST}/abc/activities", headers=auth_header(token))
|
||
assert r.status_code == 200, r.text
|
||
return {row["activity_code"] for row in r.json()["data"]}
|
||
|
||
|
||
def test_abc_activities_isolated_by_entity(client: TestClient, db: Session):
|
||
"""酣客会话不应看到博海 ABC 作业中心（生产：10 行全属 entity2）"""
|
||
_seed_two_entities(db)
|
||
_mk_activity(db, E1_ACT, 1)
|
||
_mk_activity(db, E2_ACT, 2)
|
||
_mk_activity(db, E2_ACT2, 2)
|
||
|
||
token1 = _login(client, 1)
|
||
token2 = _login(client, 2)
|
||
|
||
codes1 = _abc_codes(client, token1)
|
||
codes2 = _abc_codes(client, token2)
|
||
|
||
assert codes1 == {E1_ACT}, f"entity1 会话应只见本账套作业中心，实际 {sorted(codes1)}"
|
||
assert not any(c.startswith("ZZ_B2_E2_") for c in codes1), "酣客会话混入博海 ABC 作业中心"
|
||
assert codes2 == {E2_ACT, E2_ACT2}, f"entity2 会话应只见本账套，实际 {sorted(codes2)}"
|
||
|
||
|
||
def test_abc_activities_without_token_401(client: TestClient, db: Session):
|
||
"""鉴权底线不变：无 Bearer token → 401"""
|
||
_seed_two_entities(db)
|
||
_mk_activity(db, E1_ACT, 1)
|
||
r = client.get(f"{COST}/abc/activities")
|
||
assert r.status_code == 401, f"无 token 应 401，实际 {r.status_code}"
|