Hermes CI Fix
ac645a56ee
- ai_analysis: generate_rule_suggestions pending预警按entity过滤; kpi_analysis归属校验404; dashboard-analysis-stream/ask/review-plans 补 get_entity_id + KPI/ActionPlan/KPIAlert 全按账套过滤; stream cache key 含 entity_id (原跨企业串缓存) - dashboard: summary alert_count 补 KPIAlert.entity_id 过滤 (kpi_total 已有) - bot_kpis: update_bot_kpi_value 归属校验 404 (原任意登录可改他企 bot KPI) - cost: ActualCost/StandardCost 确认多企业共用 → cost_overview/variance/breakdown/dashboard 补 entity 过滤(engine 函数加 entity_id 透传); dashboard total_cost 既有 KeyError 修复 - kpis: create_kpi 字段白名单 (仿 UPDATE_KPI_WHITELIST+kpi_code), 拦截 status/important_flag 注入 - tests: 新增 test_multitenant_ai_20260907.py 11条越权用例 (跨entity 404/空返回断言) - 全量 pytest: 745 passed / 10 failed(与基线 HEAD 一致, 存量问题非本次引入) / 1 xfailed
275 lines
13 KiB
Python
275 lines
13 KiB
Python
"""OpenCode 审查 2026-09-07 修复验证 — ai_analysis/dashboard/bot_kpis/cost/kpis 多租户隔离越权用例
|
||
|
||
覆盖：
|
||
- ai_analysis kpi_analysis 跨企业 kpi_id → 404
|
||
- ai_analysis generate_rule_suggestions 跨企业预警不进建议
|
||
- ai_analysis review-plans / ask 按账套过滤（统计口径 + 不 500）
|
||
- dashboard summary alert_count 按账套
|
||
- bot_kpis update 跨企业写 → 404 / 同企业写 → 200
|
||
- cost dashboard/overview 按账套（ActualCost 多企业共用确认）
|
||
- kpis create 白名单：status/important_flag 注入被忽略
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan, AISuggestion, Entity, UserEntity, ActualCost
|
||
from app.api import ai_analysis
|
||
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
|
||
|
||
|
||
def _ensure_entity(db: Session, eid: int) -> Entity:
|
||
ent = db.query(Entity).filter(Entity.id == eid).first()
|
||
if not ent:
|
||
ent = Entity(id=eid, name=f"企业{eid}", short_name=f"E{eid}", status="active")
|
||
db.add(ent)
|
||
db.commit()
|
||
db.refresh(ent)
|
||
return ent
|
||
|
||
|
||
def _seed_kpi(db: Session, entity_id: int, code: str) -> KPIDefinition:
|
||
return create_test_kpi(db, kpi_code=code, entity_id=entity_id,
|
||
dimension="finance", status="active", target_value=100.0)
|
||
|
||
|
||
def _seed_user(db: Session, entity_id: int):
|
||
"""默认测试用户，登录指定账套"""
|
||
create_test_user(db)
|
||
_ensure_entity(db, entity_id)
|
||
# 无授权记录 → login 会默认授予全部 active 企业（_ensure_default_grants）
|
||
db.commit()
|
||
|
||
|
||
def _token(client: TestClient, entity_id: int) -> str:
|
||
"""登录并返回指定账套的 token（get_token_for_user 固定 entity1，这里手动登录）"""
|
||
resp = client.post("/api/cma/auth/login", json={
|
||
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
|
||
})
|
||
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
|
||
return resp.json().get("token")
|
||
|
||
|
||
# ============================================================
|
||
# ai_analysis
|
||
# ============================================================
|
||
class TestAiAnalysisIsolation:
|
||
def test_kpi_analysis_cross_entity_404(self, client: TestClient, db: Session):
|
||
"""跨企业 kpi_id 调 kpi-analysis → 404（不泄漏他企 KPI 数据）"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
k_e2 = _seed_kpi(db, entity_id=2, code="MT_AI_E2")
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.get(f"/api/cma/ai/kpi-analysis/{k_e2.id}", headers=auth_header(token1))
|
||
assert r.status_code == 404, f"跨企业 kpi 应 404: {r.text[:200]}"
|
||
|
||
def test_kpi_analysis_own_entity_ok(self, client: TestClient, db: Session):
|
||
"""本企业 kpi_id → 200（回归保护）"""
|
||
_seed_user(db, 1)
|
||
k_e1 = _seed_kpi(db, entity_id=1, code="MT_AI_E1")
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.get(f"/api/cma/ai/kpi-analysis/{k_e1.id}", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:300]
|
||
|
||
def test_generate_suggestions_cross_entity_alert_excluded(self, db: Session):
|
||
"""generate_rule_suggestions(entity=1) 不消费 entity2 的 pending 预警"""
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_kpi(db, entity_id=1, code="MT_SUG_E1")
|
||
k2 = _seed_kpi(db, entity_id=2, code="MT_SUG_E2")
|
||
db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1预警", status="pending", entity_id=1))
|
||
a2 = KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2预警", status="pending", entity_id=2)
|
||
db.add(a2)
|
||
db.commit()
|
||
|
||
ai_analysis.generate_rule_suggestions(db, entity_id=1, source="test", user_id=None)
|
||
db.commit()
|
||
|
||
# entity1 的预警 → 生成 alert 类建议
|
||
sugs = db.query(AISuggestion).filter(AISuggestion.source == "test").all()
|
||
alert_sugs = [s for s in sugs if s.suggestion_type == "action_plan" and s.target_type == "alert"]
|
||
assert len(alert_sugs) >= 1, "本账套 pending 预警应生成建议"
|
||
# 无一建议引用 entity2 的预警 id / KPI
|
||
sug_alert_ids = {s.suggestion_data.get("alert_id") for s in alert_sugs}
|
||
sug_kpi_ids = {s.suggestion_data.get("kpi_id") for s in alert_sugs}
|
||
assert a2.id not in sug_alert_ids, "跨企业预警不应进入建议"
|
||
assert k2.id not in sug_kpi_ids, "跨企业 KPI 不应进入建议"
|
||
|
||
def test_review_plans_entity_scoped(self, client: TestClient, db: Session):
|
||
"""review-plans 只复盘本账套 ActionPlan（stats.total 口径隔离）"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_kpi(db, entity_id=1, code="MT_PL_E1")
|
||
k2 = _seed_kpi(db, entity_id=2, code="MT_PL_E2")
|
||
db.add(ActionPlan(entity_id=1, alert_id=None, kpi_id=k1.id, title="e1计划", assignee="张三",
|
||
priority="high", status="in_progress", progress=50, created_by="系统"))
|
||
db.add(ActionPlan(entity_id=2, alert_id=None, kpi_id=k2.id, title="e2计划", assignee="李四",
|
||
priority="high", status="pending", progress=0, created_by="系统"))
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.post("/api/cma/ai/review-plans", headers=auth_header(token1))
|
||
assert r.status_code == 200, r.text[:300]
|
||
stats = r.json()["stats"]
|
||
assert stats["total"] == 1, f"应只统计本账套1条，实际 {stats}"
|
||
|
||
def test_ask_question_no_crash_cross_entity(self, client: TestClient, db: Session):
|
||
"""ask 上下文含他企 KPI 时不 500 且正常返回（entity 过滤后无泄漏）"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
k2 = _seed_kpi(db, entity_id=2, code="MT_ASK_E2")
|
||
db.add(KPIValue(kpi_id=k2.id, period="2026-06", actual_value=88.0, entity_id=2))
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.post("/api/cma/ai/ask", headers=auth_header(token1),
|
||
json={"question": "企业营收情况如何"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
assert "answer" in r.json()
|
||
|
||
|
||
# ============================================================
|
||
# dashboard summary alert_count
|
||
# ============================================================
|
||
class TestDashboardAlertCountIsolation:
|
||
def test_alert_count_per_entity(self, client: TestClient, db: Session):
|
||
"""summary alert_count 按账套：entity1 只数自己的 pending 预警"""
|
||
from app.utils import cache as cache_util
|
||
cache_util.delete("dashboard") # 防跨测试缓存污染
|
||
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
k1 = _seed_kpi(db, entity_id=1, code="MT_DB_E1")
|
||
k2 = _seed_kpi(db, entity_id=2, code="MT_DB_E2")
|
||
db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="e1", status="pending", entity_id=1))
|
||
db.add(KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="e2a", status="pending", entity_id=2))
|
||
db.add(KPIAlert(kpi_id=k2.id, alert_level="yellow", alert_message="e2b", status="resolved", entity_id=2))
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
token2 = _token(client, 2)
|
||
|
||
r1 = client.get("/api/cma/dashboard/summary", headers=auth_header(token1))
|
||
assert r1.status_code == 200, r1.text[:200]
|
||
assert r1.json()["alert_count"] == 1, f"entity1 应只数自己的1条: {r1.json()['alert_count']}"
|
||
|
||
cache_util.delete("dashboard")
|
||
r2 = client.get("/api/cma/dashboard/summary", headers=auth_header(token2))
|
||
assert r2.status_code == 200, r2.text[:200]
|
||
assert r2.json()["alert_count"] == 1, f"entity2 应只数自己的 pending 1条(不含resolved): {r2.json()['alert_count']}"
|
||
|
||
|
||
# ============================================================
|
||
# bot_kpis 归属校验
|
||
# ============================================================
|
||
class TestBotKpisOwnership:
|
||
def _bot_kpi(self, db: Session, entity_id: int, code: str) -> KPIDefinition:
|
||
return create_test_kpi(db, kpi_code=code, entity_id=entity_id, status="active",
|
||
bot_source="finance-bot", kpi_name=code, target_value=100.0)
|
||
|
||
def test_cross_entity_update_404(self, client: TestClient, db: Session):
|
||
"""entity1 用户改 entity2 的 Bot KPI → 404"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
k2 = self._bot_kpi(db, entity_id=2, code="MT_BOT_E2")
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.post(f"/api/cma/bot-kpis/{k2.id}/value", headers=auth_header(token1),
|
||
json={"actual_value": 55, "period": "2026-09"})
|
||
assert r.status_code == 404, f"跨企业改 Bot KPI 应 404: {r.status_code} {r.text[:200]}"
|
||
|
||
def test_same_entity_update_ok(self, client: TestClient, db: Session):
|
||
"""本企业改自己的 Bot KPI → 200（回归保护）"""
|
||
_seed_user(db, 1)
|
||
k1 = self._bot_kpi(db, entity_id=1, code="MT_BOT_E1")
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
|
||
r = client.post(f"/api/cma/bot-kpis/{k1.id}/value", headers=auth_header(token1),
|
||
json={"actual_value": 66, "period": "2026-09"})
|
||
assert r.status_code == 200, r.text[:300]
|
||
|
||
|
||
# ============================================================
|
||
# cost 多账套（ActualCost 共用表确认 → 过滤）
|
||
# ============================================================
|
||
class TestCostIsolation:
|
||
def _seed_actual(self, db: Session, entity_id: int, code: str, amount: float, period: str = "2026-06"):
|
||
ac = ActualCost(entity_id=entity_id, period=period, product_code=code, product_name=code,
|
||
cost_type="material", item_name="材料", actual_quantity=1.0,
|
||
actual_price=amount, actual_cost=amount, source="manual")
|
||
db.add(ac)
|
||
|
||
def test_cost_dashboard_entity_filtered(self, client: TestClient, db: Session):
|
||
"""cost/dashboard 产品列表/总额按账套过滤（ActualCost 多企业共用）"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
self._seed_actual(db, 1, "PROD_E1", 100.0)
|
||
self._seed_actual(db, 2, "PROD_E2", 200.0)
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
token2 = _token(client, 2)
|
||
|
||
r1 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token1))
|
||
assert r1.status_code == 200, r1.text[:300]
|
||
codes1 = {p["code"] for p in r1.json()["products"]}
|
||
assert codes1 == {"PROD_E1"}, f"entity1 不应看到 entity2 产品: {codes1}"
|
||
|
||
r2 = client.get("/api/cma/cost/dashboard?period=2026-06", headers=auth_header(token2))
|
||
assert r2.status_code == 200, r2.text[:300]
|
||
codes2 = {p["code"] for p in r2.json()["products"]}
|
||
assert codes2 == {"PROD_E2"}, f"entity2 不应看到 entity1 产品: {codes2}"
|
||
|
||
def test_cost_overview_entity_filtered(self, client: TestClient, db: Session):
|
||
"""cost/overview manual_cost 按账套过滤"""
|
||
_seed_user(db, 1)
|
||
_ensure_entity(db, 2)
|
||
self._seed_actual(db, 1, "OV_E1", 100.0)
|
||
self._seed_actual(db, 2, "OV_E2", 300.0)
|
||
db.commit()
|
||
token1 = _token(client, 1)
|
||
token2 = _token(client, 2)
|
||
|
||
r1 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token1))
|
||
assert r1.status_code == 200, r1.text[:300]
|
||
assert r1.json()["manual_cost"] == 100.0, f"entity1 manual_cost 应100: {r1.json()}"
|
||
|
||
r2 = client.get("/api/cma/cost/overview?period=2026-06", headers=auth_header(token2))
|
||
assert r2.status_code == 200, r2.text[:300]
|
||
assert r2.json()["manual_cost"] == 300.0, f"entity2 manual_cost 应300: {r2.json()}"
|
||
|
||
|
||
# ============================================================
|
||
# kpis create 白名单（mass-assignment 残留项）
|
||
# ============================================================
|
||
class TestKpiCreateWhitelist:
|
||
def test_create_injected_sensitive_fields_ignored(self, client: TestClient, db: Session):
|
||
"""create 带 status/important_flag 注入 → 被忽略，落库为默认值"""
|
||
_seed_user(db, 1)
|
||
token1 = _token(client, 1)
|
||
|
||
resp = client.post("/api/cma/kpis", headers=auth_header(token1), json={
|
||
"kpi_code": "F_MT_INJ_001",
|
||
"kpi_name": "注入测试指标",
|
||
"dimension": "finance",
|
||
"target_value": 100.0,
|
||
"unit": "%",
|
||
"formula": "测试公式",
|
||
"data_source": "测试系统",
|
||
"data_owner": "测试管理员",
|
||
"status": "disabled", # 敏感字段：注入应被忽略 → 默认 active
|
||
"important_flag": 1, # 敏感字段：注入应被忽略 → 默认 0
|
||
"entity_id": 999, # 归属字段：强制覆盖为 token 企业 1
|
||
})
|
||
assert resp.status_code == 200, resp.text[:300]
|
||
data = resp.json()
|
||
assert data["status"] == "active", f"status 注入应被忽略: {data.get('status')}"
|
||
assert data.get("important_flag", 0) == 0, f"important_flag 注入应被忽略: {data.get('important_flag')}"
|
||
|
||
row = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_MT_INJ_001").first()
|
||
assert row is not None
|
||
assert row.entity_id == 1, f"entity_id 应强制为 token 企业1: {row.entity_id}"
|