Hermes CI Fix
80fbc11cba
- 新增 POST /api/cma/bot/ai/suggestions (bot_suggestions.py, X-BOT-KEY鉴权, L2) 高危建议(kpi_target/budget_adjust/action_plan) status恒unapplied 等人审批 source白名单(bot/rule)防冒充manual; entity多租户归属校验; 幂等防重 - risk_levels.py: API_RISK_MAP加新端点 + 文件头补写操作语义分级low/high说明 - 收口: bot/kpis/create-with-links 拒绝直写非零target_value(显式置0, 考核基准=人拍板) - test_risk_levels: L2计数5→6; 新增test_bot_suggestion_channel 12条(鉴权/语义/多租户/幂等/审批闭环/审计)
197 lines
8.3 KiB
Python
197 lines
8.3 KiB
Python
"""Bot侧高危建议创建通道 + 分层授权 测试 (P0-② 2026-09-03)
|
||
|
||
覆盖:
|
||
1. 鉴权: 无 X-BOT-KEY → 401; X-BOT-KEY → 200
|
||
2. 高危语义: status 恒 unapplied; source 白名单(bot/rule); 类型白名单
|
||
3. 多租户: 建议只能指向当前账套KPI（跨账套 target → 404）; entity 不存在/未激活 → 403
|
||
4. 幂等: 同 entity+type+target_id+title 未应用已存在 → 不重复创建
|
||
5. 审批闭环: 人 apply → KPI目标变更 + apply_detail before/after + applied_by/applied_at + 重复apply拒绝
|
||
6. audit 留痕: bot 创建建议写入 audit（monkeypatch捕获 /api/cma/bot/* 中间件）
|
||
7. create-with-links 收口: Bot 直写非零 target_value → 400
|
||
"""
|
||
import json
|
||
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
from app.models import AISuggestion, KPIDefinition, Entity
|
||
|
||
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
|
||
|
||
BOT_PATH = "/api/cma/bot/ai/suggestions"
|
||
|
||
|
||
def _bot_create(client, kpi_id, **kw):
|
||
body = {
|
||
"suggestion_type": "kpi_target",
|
||
"target_id": kpi_id,
|
||
"title": "Bot建议：上调目标（E2E-验证）",
|
||
"content": "规则引擎/财务Bot分析建议",
|
||
"source": "bot",
|
||
"suggestion_data": {"kpi_id": kpi_id, "target_value": 150.0},
|
||
}
|
||
body.update(kw)
|
||
return client.post(BOT_PATH, json=body, headers=BOT_KEY)
|
||
|
||
|
||
class TestBotAuth:
|
||
def test_requires_bot_key(self, client, db):
|
||
kpi = create_test_kpi(db)
|
||
r = client.post(BOT_PATH, json={"suggestion_type": "kpi_target", "title": "x",
|
||
"target_id": kpi.id})
|
||
assert r.status_code == 401
|
||
|
||
def test_create_with_bot_key(self, client, db):
|
||
kpi = create_test_kpi(db)
|
||
r = _bot_create(client, kpi.id)
|
||
assert r.status_code == 200, r.text
|
||
data = r.json()["data"]
|
||
assert data["status"] == "unapplied"
|
||
assert data["source"] == "bot"
|
||
assert data["suggestion_type"] == "kpi_target"
|
||
assert data["user_id"] is None
|
||
# 落库
|
||
sug = db.query(AISuggestion).filter(AISuggestion.id == data["id"]).first()
|
||
assert sug is not None and sug.status == "unapplied"
|
||
|
||
def test_human_token_rejected_on_bot_path(self, client, db):
|
||
"""人token不能走Bot通道（Bot端点只认X-BOT-KEY）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db)
|
||
r = client.post(BOT_PATH, json={"suggestion_type": "kpi_target", "title": "x",
|
||
"target_id": kpi.id, "target_value": 150},
|
||
headers=auth_header(token))
|
||
assert r.status_code == 401
|
||
|
||
|
||
class TestBotSemantics:
|
||
def test_type_whitelist(self, client, db):
|
||
kpi = create_test_kpi(db)
|
||
r = _bot_create(client, kpi.id, suggestion_type="delete_all")
|
||
assert r.status_code == 400
|
||
assert "仅支持" in r.text
|
||
|
||
def test_source_whitelist_no_spoof(self, client, db):
|
||
kpi = create_test_kpi(db)
|
||
r = _bot_create(client, kpi.id, source="manual")
|
||
assert r.status_code == 400
|
||
assert "source" in r.text
|
||
|
||
def test_kpi_target_requires_kpi_and_value(self, client, db):
|
||
# 无 kpi/target
|
||
r = _bot_create(client, None, suggestion_data={})
|
||
assert r.status_code == 400
|
||
# kpi_target 无 target_value
|
||
kpi = create_test_kpi(db)
|
||
r2 = _bot_create(client, kpi.id, suggestion_data={"kpi_id": kpi.id})
|
||
assert r2.status_code == 400
|
||
|
||
def test_cross_entity_target_rejected(self, client, db):
|
||
"""多租户铁律: 建议只能指向当前账套KPI"""
|
||
db.add(Entity(id=2, name="另一企业", short_name="B", status="active"))
|
||
db.commit()
|
||
kpi_e1 = create_test_kpi(db, entity_id=1)
|
||
# 以 entity_id=2 身份建议 entity1 的KPI → 404
|
||
r = _bot_create(client, kpi_e1.id, entity_id=2)
|
||
assert r.status_code == 404
|
||
assert "不存在于当前账套" in r.text
|
||
# entity 不存在/未激活 → 403
|
||
r2 = _bot_create(client, kpi_e1.id, entity_id=99)
|
||
assert r2.status_code == 403
|
||
|
||
def test_idempotent(self, client, db):
|
||
kpi = create_test_kpi(db)
|
||
r1 = _bot_create(client, kpi.id)
|
||
r2 = _bot_create(client, kpi.id)
|
||
assert r1.status_code == 200
|
||
assert r2.status_code == 200
|
||
assert r2.json()["duplicated"] is True
|
||
assert r2.json()["data"]["id"] == r1.json()["data"]["id"]
|
||
assert db.query(AISuggestion).count() == 1
|
||
|
||
|
||
class TestApproveClosedLoop:
|
||
def test_bot_suggestion_apply_chain(self, client, db):
|
||
"""bot创建kpi_target建议 → unapplied → 人apply → 目标值生效 + 留痕完整"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, target_value=100.0)
|
||
|
||
r = _bot_create(client, kpi.id, title="Bot建议：上调目标（E2E-验证）")
|
||
sug_id = r.json()["data"]["id"]
|
||
assert r.json()["data"]["status"] == "unapplied"
|
||
|
||
# 待批列表可见（status=unapplied）
|
||
lst = client.get("/api/cma/ai/suggestions?status=unapplied", headers=auth_header(token)).json()
|
||
assert any(s["id"] == sug_id for s in lst["data"])
|
||
|
||
# 人 apply
|
||
app = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "kpi_target", "target_value": 150.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app.status_code == 200, app.text
|
||
ad = app.json()["data"]
|
||
assert ad["status"] == "applied"
|
||
assert ad["applied_by"] == "测试管理员"
|
||
assert ad["applied_at"]
|
||
assert ad["apply_detail"][0]["before"] == 100.0
|
||
assert ad["apply_detail"][0]["after"] == 150.0
|
||
assert ad["apply_detail"][0]["action"] == "update_target_value"
|
||
|
||
db.refresh(kpi)
|
||
assert kpi.target_value == 150.0
|
||
|
||
# 重复apply拒绝
|
||
app2 = client.post(
|
||
f"/api/cma/ai/suggestions/{sug_id}/apply",
|
||
json={"action": "kpi_target", "target_value": 200.0},
|
||
headers=auth_header(token),
|
||
)
|
||
assert app2.status_code == 400
|
||
|
||
|
||
class TestAuditCoverage:
|
||
def test_bot_suggestion_audited(self, client, db, monkeypatch):
|
||
"""/api/cma/bot/ai/suggestions 被 bot_audit_middleware 留痕（risk L2）"""
|
||
import app.api.audit_log as audit_log_module
|
||
records = []
|
||
monkeypatch.setattr(audit_log_module, "write_audit_line",
|
||
lambda rec: records.append(rec))
|
||
kpi = create_test_kpi(db)
|
||
r = _bot_create(client, kpi.id)
|
||
assert r.status_code == 200
|
||
assert records, "应有审计记录"
|
||
rec = records[-1]
|
||
assert rec["endpoint"] == BOT_PATH
|
||
assert rec["method"] == "POST"
|
||
assert rec["risk_level"] == "L2"
|
||
assert rec["bot_name"] == "财务BOT"
|
||
assert rec["status"] == 200
|
||
|
||
|
||
class TestCreateWithLinksClosure:
|
||
def test_nonzero_target_value_rejected(self, client, db):
|
||
"""收口: create-with-links 直写非零 target_value → 400"""
|
||
r = client.post("/api/cma/bot/kpis/create-with-links",
|
||
json={"kpi_code": "E2E_TGT", "kpi_name": "收口验证KPI",
|
||
"dimension": "finance", "target_value": 500000,
|
||
"entity_id": 1},
|
||
headers=BOT_KEY)
|
||
assert r.status_code == 400
|
||
assert "不允许直写目标值" in r.text
|
||
|
||
def test_without_target_value_ok(self, client, db):
|
||
"""省略/置0 target_value 仍可创建KPI（收口自动置0，不阻断KPI创建）"""
|
||
r = client.post("/api/cma/bot/kpis/create-with-links",
|
||
json={"kpi_code": "F_E2E_TGT2", "kpi_name": "收口放行KPI",
|
||
"dimension": "finance", "unit": "万元", "formula": "营业收入",
|
||
"data_source": "Bot创建验证", "data_owner": "财务部",
|
||
"frequency": "monthly", "kpi_level": "operational",
|
||
"entity_id": 1},
|
||
headers=BOT_KEY)
|
||
assert r.status_code == 200, r.text
|
||
kpi = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_E2E_TGT2").first()
|
||
assert kpi is not None
|
||
assert kpi.target_value == 0
|