From 80fbc11cba30823f19b9f4fc91d008b95cdca689 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Thu, 3 Sep 2026 18:08:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(bot-tier):=20M1=20Bot=E4=BE=A7=E9=AB=98?= =?UTF-8?q?=E5=8D=B1=E5=BB=BA=E8=AE=AE=E5=88=9B=E5=BB=BA=E9=80=9A=E9=81=93?= =?UTF-8?q?+=E7=9B=AE=E6=A0=87=E5=80=BC=E7=9B=B4=E5=86=99=E6=94=B6?= =?UTF-8?q?=E5=8F=A3=20(P0-=E2=91=A1=E5=88=86=E5=B1=82=E6=8E=88=E6=9D=83)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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条(鉴权/语义/多租户/幂等/审批闭环/审计) --- backend/app/api/bot_bridge.py | 17 +- backend/app/api/bot_suggestions.py | 175 +++++++++++++++++ backend/app/main.py | 3 +- backend/app/risk_levels.py | 31 ++- backend/tests/test_bot_suggestion_channel.py | 196 +++++++++++++++++++ backend/tests/test_risk_levels.py | 6 +- 6 files changed, 422 insertions(+), 6 deletions(-) create mode 100644 backend/app/api/bot_suggestions.py create mode 100644 backend/tests/test_bot_suggestion_channel.py diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index 3d5e3a1c..2282d50c 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -728,11 +728,26 @@ def bot_kpi_value_with_check(data: dict, db: Session = Depends(get_db), bot: dic @risk_level("L3") def bot_kpi_create_with_links(data: dict, db: Session = Depends(get_db), bot: dict = Depends(verify_bot_key)): """聚合B: 创建KPI + 关联战略地图 + 批量因果链（Agent建KPI标准动作） - body: {kpi_code, kpi_name, dimension, entity_id?, target_value?, unit?, link_map_id?, link_causality?}""" + body: {kpi_code, kpi_name, dimension, entity_id?, unit?, link_map_id?, link_causality?} + + 分层授权收口（2026-09-03 P0-②）: Bot 创建KPI不得携带自定目标值—— + 考核基准类（目标值）= 人拍板；target_value 一律显式置 0（满足治理必填， + 后续由人工设定或走 POST /api/cma/bot/ai/suggestions 建议通道待审批）。 + """ entity_id = int(data.get("entity_id") or 1) from app.api.kpis import _validate_kpi_data, apply_calc_type_inference kpi_data = {k: v for k, v in data.items() if k not in ("entity_id", "link_map_id", "link_causality")} + # 收口: 非零 target_value 直接拒绝（不静默丢弃，让调用方明确知晓规则） + tv_raw = data.get("target_value") + if tv_raw not in (None, 0, "0", 0.0): + raise HTTPException( + 400, + "Bot创建KPI不允许直写目标值（考核基准类=人拍板，2026-09-03分层授权）。" + "请省略 target_value（自动置0，人工后续设定）或改走 POST /api/cma/bot/ai/suggestions 建议通道待审批。", + ) + kpi_data["target_value"] = 0 # 显式置0（覆盖 0/"0"/None 三种传入） + # ① 创建KPI（编码唯一 + 治理校验 + 强制企业） code = kpi_data.get("kpi_code", "") if not code: diff --git a/backend/app/api/bot_suggestions.py b/backend/app/api/bot_suggestions.py new file mode 100644 index 00000000..3f655a02 --- /dev/null +++ b/backend/app/api/bot_suggestions.py @@ -0,0 +1,175 @@ +"""Bot侧AI高危建议创建通道 — 写操作分层授权落地 (P0-② 2026-09-03) + +背景（CMA交互架构规范 v1 5.1 分层授权模型，任总拍板）: + 考核基准类（KPI目标值/预算调整）→ 🔒 Bot只能"建议"，人审批后生效 + Bot不得直写 KPI target_value / budget 审批类字段； + Bot产生的高危变更一律走本通道建 ai_suggestions（status=unapplied） + → UI AI建议中心(SuggestionCenter.vue)待批列表 → 人 apply 后生效+留痕。 + +通道关系（两套分级勿混，见 risk_levels.py 顶部注释）: + L1-L4（技术风险分级）: 本端点挂 L2（业务写单条） + low/high（业务写语义分级）: 本端点是 high 类的"建议载体"—— + 端点本身不直写考核基准，只建建议行，属 high 流程的入口。 + +鉴权/审计: + X-BOT-KEY（verify_bot_key, bot_bridge 共享密钥）+ @risk_level("L2") + 路径 /api/cma/bot/* → bot_audit_middleware 自动留痕（bot_name/risk/status） + entity_id 多租户: X-Entity-Id header → body.entity_id → 默认1，校验 Entity.active +""" +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Request +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models import AISuggestion, KPIDefinition, Entity +from app.api.bot_bridge import verify_bot_key +from app.api.ai_suggestions import _sug_dict +from app.risk_levels import risk_level + +router = APIRouter(prefix="/api/cma/bot/ai/suggestions", tags=["AI建议Bot通道"]) + +# 高危建议类型（Bot仅允许创建这三类建议，禁止其他） +_SUPPORTED_TYPES = ("kpi_target", "budget_adjust", "action_plan") +# source 白名单：Bot 不得冒充 manual/dashboard/kpi/budget（影响UI语义与推送判断） +_ALLOWED_SOURCES = ("rule", "bot") +_TARGET_TYPE_MAP = { + "kpi_target": "kpi", + "budget_adjust": "budget", + "action_plan": "kpi", +} + + +def _resolve_bot_entity(request: Request, data: dict, db: Session) -> int: + """账套解析（Bot无token通道）: body.entity_id → X-Entity-Id header → 默认1，校验active。 + + 与 deps.get_entity_id 的 Bot 白名单语义一致（query/header/body 任一来源）， + 本端点不接收登录token，故不在此处理 Bearer。 + """ + raw = None + if data.get("entity_id") not in (None, ""): + raw = data["entity_id"] + else: + xh = request.headers.get("X-Entity-Id") + if xh and str(xh).isdigit(): + raw = int(xh) + try: + eid = int(raw) if raw is not None else 1 + except (TypeError, ValueError): + raise HTTPException(422, f"entity_id 无效: {raw!r}") + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent or ent.status != "active": + raise HTTPException(403, f"企业 entity_id={eid} 不存在或未激活") + return eid + + +@router.post("") +@risk_level("L2") +def bot_create_suggestion( + request: Request, + data: dict, + bot: dict = Depends(verify_bot_key), + db: Session = Depends(get_db), +): + """Bot创建高危建议（kpi_target/budget_adjust/action_plan），status 固定 unapplied 等人审批。 + + 入参（字段与人工创建端点 /api/cma/ai/suggestions 一致）: + suggestion_type (str, 必填): kpi_target / budget_adjust / action_plan + title (str, 必填): 建议标题（幂等键之一） + entity_id (int, 可选, 默认1): 账套；也支持 X-Entity-Id header + target_id (int, 可选): 目标ID；kpi_target/budget_adjust 必填(kpi_id 亦可) + suggestion_data(dict, 可选): 应用参数 + kpi_target: {kpi_id, target_value} + budget_adjust:{kpi_id, period:YYYY-MM, budget_value} + action_plan: {kpi_id, title?, assignee?, priority?, due_date?} + content (str, 可选): 建议理由/依据（财务Bot三情景测算content携带） + source (str, 默认"bot"): 仅允许 bot/rule，禁止冒充 manual/dashboard + target_type (str, 可选): 缺省按 suggestion_type 映射 kpi/budget + + 约束: + - status 恒为 unapplied（Bot 不能直接生效） + - kpi_target/budget_adjust 必须带有效 kpi_id/target_id 且属于当前账套 + - 幂等: 同 entity+suggestion_type+target_id+title 且 unapplied 已存在 → 返回既有建议 + """ + suggestion_type = data.get("suggestion_type") + title = str(data.get("title") or "").strip() + if suggestion_type not in _SUPPORTED_TYPES: + raise HTTPException( + 400, + f"Bot建议仅支持 suggestion_type: {'/'.join(_SUPPORTED_TYPES)}（高危变更=建议通道，不提供直写）", + ) + if not title: + raise HTTPException(400, "缺少 title") + + source = data.get("source", "bot") + if source not in _ALLOWED_SOURCES: + raise HTTPException(400, f"source 仅允许 {'/'.join(_ALLOWED_SOURCES)}（禁止冒充 manual/dashboard/kpi/budget）") + + entity_id = _resolve_bot_entity(request, data, db) + sd = data.get("suggestion_data") or {} + target_type = data.get("target_type") or _TARGET_TYPE_MAP.get(suggestion_type) + + # 目标归属校验（多租户铁律: 只能建议当前账套的KPI） + kpi_id = None + raw_kpi = data.get("kpi_id") or data.get("target_id") or sd.get("kpi_id") + if raw_kpi is not None: + try: + kpi_id = int(raw_kpi) + except (TypeError, ValueError): + kpi_id = None + if suggestion_type in ("kpi_target", "budget_adjust"): + if kpi_id is None: + raise HTTPException(400, f"suggestion_type={suggestion_type} 必须带 kpi_id/target_id") + kpi = db.query(KPIDefinition).filter( + KPIDefinition.id == kpi_id, + KPIDefinition.entity_id == entity_id, + ).first() + if not kpi: + raise HTTPException(404, f"KPI {kpi_id} 不存在于当前账套 entity_id={entity_id}") + if suggestion_type == "kpi_target" and sd.get("target_value") is None: + raise HTTPException(400, "kpi_target 建议的 suggestion_data 必须带 target_value") + if suggestion_type == "budget_adjust" and ( + not sd.get("period") or sd.get("budget_value") is None + ): + raise HTTPException(400, "budget_adjust 建议的 suggestion_data 必须带 period(YYYY-MM) + budget_value") + + # 幂等: 同 entity+type+target_id+title 的 unapplied 建议存在则不重复创建 + target_id = kpi_id if kpi_id is not None else data.get("target_id") + dup = db.query(AISuggestion).filter( + AISuggestion.entity_id == entity_id, + AISuggestion.suggestion_type == suggestion_type, + AISuggestion.target_id == target_id, + AISuggestion.title == title, + AISuggestion.status == "unapplied", + ).first() + if dup: + return { + "success": True, + "message": "建议已存在（幂等，未重复创建）", + "data": _sug_dict(dup), + "duplicated": True, + } + + category = "alert" if target_type == "alert" else data.get("category", "decision") + sug = AISuggestion( + entity_id=entity_id, + user_id=None, # Bot 创建无人工账号；bot 名与调用留痕在 bot_audit.log（X-BOT-KEY） + source=source, + suggestion_type=suggestion_type, + target_type=target_type, + category=category, + target_id=target_id, + title=title, + content=data.get("content"), + suggestion_data=sd, + status="unapplied", # 铁律: Bot 只能建议，生效必须人 apply + ) + db.add(sug) + db.commit() + db.refresh(sug) + return { + "success": True, + "message": "高危建议已创建，等待人工审批（AI建议中心 → 应用）", + "data": _sug_dict(sug), + "duplicated": False, + } diff --git a/backend/app/main.py b/backend/app/main.py index 8aaa8399..51dbd373 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -5,7 +5,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from dotenv import load_dotenv from app.database import init_db -from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure +from app.api import auth, kpis, kpi_governance, templates, maps, dashboard, data, alerts, ai_analysis, ai_suggestions, alert_rules, users, thresholds, notifications, permissions, action_plans, alignment, org, objectives, versions, budget, cost, predict, reports, security, knowledge, bot_bridge, bot_bridge_v2, bot_suggestions, lead, tenant, customer_dashboard, deviation_push, budget_generate, knowledge_articles, kpi_causality, data_quality, bi_reports, entities, bsc_layers, okr, okr_templates, subjects, driver_budget, bot_kpis, ontology, bot_iron_law, analysis_results, expenses, cash, tax_compliance, verify, growth_quality, products, data_classification, value_sources, zero_based, derivation_rules, cash_classify, cashflow_structure from app.utils.cache import clear_all as clear_cache, delete as delete_cache from scripts.erp_sync import run_sync as run_erp_sync from app.auth_middleware import require_auth @@ -44,6 +44,7 @@ app.include_router(data.router) app.include_router(alerts.router) app.include_router(ai_analysis.router) app.include_router(ai_suggestions.router) +app.include_router(bot_suggestions.router) app.include_router(alert_rules.router) app.include_router(users.router) app.include_router(thresholds.router) diff --git a/backend/app/risk_levels.py b/backend/app/risk_levels.py index b454feab..1ebc7fa4 100644 --- a/backend/app/risk_levels.py +++ b/backend/app/risk_levels.py @@ -9,6 +9,34 @@ CMA Bot API 风险分级（L1-L4）与路由标注装饰器 L4 危险 — 不向Bot API开放（DROP/TRUNCATE/批量DELETE/生产结构修改） 安全底线: Bot API面不存在L4端点（0项开放 = 天然隔离）。 + +写操作语义分级（业务语义 low/high，2026-09-03 分层授权落地 P0-②） +============================================================ +依据: CMA交互架构规范 v1 第5.1节 分层授权模型（任总拍板）—— + 考核基准类（KPI目标值/预算调整）→ Bot只能"建议"，人拍板生效； + 监测参数类（红黄绿阈值）→ Bot建议+人批量确认（P1待办，未开放Bot直写）； + 预算 = 三阶段分权（编制人定框架/执行Bot自动/调整人审批）。 + +low/high 与 L1-L4 是两套正交分级，勿混: + L1-L4 = 技术风险分级（操作面/写前校验/批量约束，本文件API_RISK_MAP） + low/high = 业务写语义分级（值是不是"考核基准/资源承诺"） + 关系: low 类 = 现有 L2/L3 直写端点行为不变（规则型数据写）； + high 类 = Bot API面不提供直写端点，只提供建议通道端点（L2）。 + +端点归类清单（low=Bot直写+留痕 / high=建议通道人审批）: + [low · 直写 KPI实际值/流水/验证记录]（写库即生效，靠 audit_log/source_batch 留痕） + POST /api/cma/bot/kpi-value-with-check — 写KPI实际值+联动预警 (L2) + POST /api/cma/bot-bridge/kpi-result — 写KPI实际值 (source_batch) (L2) + POST /api/cma/bot-kpis/{kpi_id}/value — 写KPI当前值 (L2) + POST /api/cma/bot/import — Excel批量导入KPI实际值 (L3) + POST /api/cma/bot-bridge/mpm-result — MPM分析结果→KPI实际值 (L3) + POST /api/cma/bot-bridge/verify/{action_plan_id} — 验证ActionPlan执行 (L2, 轻量verify_log) + POST /api/cma/bot/okr/create — 创建OKR Objective (L2, 管理类业务写，保留评审: 2026-09-03 P1待办) + POST /api/cma/bot/kpis/create-with-links — 创建新KPI定义+因果链 (L3; 2026-09-03收口: 拒绝直写非零 target_value) + [high · 建议通道（端点只建 ai_suggestions, status=unapplied, 人 apply 后生效+apply_detail留痕）] + POST /api/cma/bot/ai/suggestions — Bot创建高危建议 (L2, 本任务新增) + 说明: KPI目标值修改/预算调整 无任何Bot直写端点（grep实证）—— + kpi_target/budget_adjust 的生效只经 /api/cma/ai/suggestions/{id}/apply（人token）。 """ import functools @@ -54,12 +82,13 @@ API_RISK_MAP = { "GET /api/cma/bot-kpis": "L1", # 本任务新增的只读端点 "GET /api/cma/bot/risk-levels": "L1", - # ── L2 业务写（4项方案清单 + okr/create单条业务写） ── + # ── L2 业务写（4项方案清单 + okr/create单条业务写 + bot建议通道） ── "POST /api/cma/bot/kpi-value-with-check": "L2", "POST /api/cma/bot-bridge/kpi-result": "L2", "POST /api/cma/bot-kpis/{kpi_id}/value": "L2", "POST /api/cma/bot-bridge/verify/{action_plan_id}": "L2", "POST /api/cma/bot/okr/create": "L2", # 单条OKR创建（方案清单未列出，按单条业务写归类） + "POST /api/cma/bot/ai/suggestions": "L2", # Bot高危建议创建通道（写操作语义 high，见文件头注释；本身只建建议不直写考核基准） # ── L3 批量写/创建（3项） ── "POST /api/cma/bot/import": "L3", "POST /api/cma/bot/kpis/create-with-links": "L3", diff --git a/backend/tests/test_bot_suggestion_channel.py b/backend/tests/test_bot_suggestion_channel.py new file mode 100644 index 00000000..be749400 --- /dev/null +++ b/backend/tests/test_bot_suggestion_channel.py @@ -0,0 +1,196 @@ +"""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 diff --git a/backend/tests/test_risk_levels.py b/backend/tests/test_risk_levels.py index 97340ba0..badea720 100644 --- a/backend/tests/test_risk_levels.py +++ b/backend/tests/test_risk_levels.py @@ -49,11 +49,11 @@ class TestRiskMapCoverage: assert not unlabeled, f"以下路由函数缺少 @risk_level 装饰器: {unlabeled}" def test_risk_level_counts(self): - """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=5(4项清单+okr/create)、L3=3、L4=0""" + """分级统计与方案一致：L1=21(20项清单+risk-levels端点)、L2=6(4项清单+okr/create+bot建议通道)、L3=3、L4=0""" from collections import Counter counts = Counter(API_RISK_MAP.values()) assert counts["L1"] == 21, counts - assert counts["L2"] == 5, counts + assert counts["L2"] == 6, counts assert counts["L3"] == 3, counts assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）" @@ -94,7 +94,7 @@ class TestRiskLevelsEndpoint: # 分级统计 assert data["summary"]["L1"] == 21 - assert data["summary"]["L2"] == 5 + assert data["summary"]["L2"] == 6 assert data["summary"]["L3"] == 3 assert data["summary"]["L4"] == 0