"""Bot API 风险分级（L1-L4）标注 + 操作审计日志 测试 覆盖: 1. API_RISK_MAP 覆盖所有 /api/cma/bot* 路由（app.routes 遍历核对） 2. GET /api/cma/bot/risk-levels 返回200且含L1-L4定义 3. Bot API面不存在L4端点（无 drop/truncate/delete 批量端点，安全底线） 4. 审计日志在调用Bot API后写入（monkeypatch + 真实日志文件双验证） """ import io import json import pytest from openpyxl import Workbook from app.main import app from app.models import KPIDefinition from app.risk_levels import API_RISK_MAP, RISK_LEVELS import app.api.audit_log as audit_log_module BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"} _HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH") def _bot_routes(): """遍历 app.routes，返回所有 /api/cma/bot* 路由 (route, methods列表)""" routes = [] for r in app.routes: path = getattr(r, "path", "") if path.startswith("/api/cma/bot"): methods = sorted(m for m in (getattr(r, "methods", set()) or set()) if m in _HTTP_METHODS) routes.append((r, methods)) return routes class TestRiskMapCoverage: def test_api_risk_map_covers_all_bot_routes(self): """每个 /api/cma/bot* 路由都在 API_RISK_MAP 有级别标注，且函数有@risk_level装饰器""" missing = [] unlabeled = [] for r, methods in _bot_routes(): for m in methods: key = f"{m} {r.path}" if key not in API_RISK_MAP: missing.append(key) if not getattr(r.endpoint, "risk_level", None): unlabeled.append(f"{sorted(methods)} {r.path}") assert not missing, f"API_RISK_MAP 缺少以下路由标注: {missing}" 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""" from collections import Counter counts = Counter(API_RISK_MAP.values()) assert counts["L1"] == 21, counts assert counts["L2"] == 5, counts assert counts["L3"] == 3, counts assert counts["L4"] == 0, "Bot API面不得存在L4端点（安全底线）" def test_no_l4_bot_endpoints(self): """Bot API面不存在L4端点：无DELETE方法、无drop/truncate/delete危险路径""" danger_keywords = ("drop", "truncate", "delete") for r, methods in _bot_routes(): assert getattr(r.endpoint, "risk_level", None) != "L4", \ f"{r.path} 不应被标注为L4" assert "DELETE" not in methods, f"Bot路由不应有DELETE方法: {r.path}" low = r.path.lower() for kw in danger_keywords: assert kw not in low, f"Bot路由不应含危险路径片段: {r.path}" class TestRiskLevelsEndpoint: def test_risk_levels_requires_bot_key(self, client): """无X-BOT-KEY → 401""" resp = client.get("/api/cma/bot/risk-levels") assert resp.status_code == 401 def test_risk_levels_returns_200_with_definitions(self, client): """X-BOT-KEY → 200，含L1-L4定义与API→级别→处理方式清单""" resp = client.get("/api/cma/bot/risk-levels", headers=BOT_KEY) assert resp.status_code == 200 data = resp.json() # L1-L4 定义齐全 for level, label in RISK_LEVELS.items(): assert data["risk_levels"][level] == label, f"缺少 {level} 定义" # API→级别→处理方式 清单 apis = {f"{a['method']} {a['path']}": a for a in data["apis"]} assert apis["GET /api/cma/bot/ping"]["risk_level"] == "L1" assert apis["POST /api/cma/bot/kpi-value-with-check"]["risk_level"] == "L2" assert apis["POST /api/cma/bot/import"]["risk_level"] == "L3" assert apis["POST /api/cma/bot/import"]["handling"] # 处理方式非空 # 分级统计 assert data["summary"]["L1"] == 21 assert data["summary"]["L2"] == 5 assert data["summary"]["L3"] == 3 assert data["summary"]["L4"] == 0 class TestAuditLog: def test_audit_record_after_bot_call(self, client, monkeypatch): """调用Bot API后写出审计记录（monkeypatch捕获）""" records = [] monkeypatch.setattr(audit_log_module, "write_audit_line", lambda rec: records.append(rec)) resp = client.get("/api/cma/bot/ping") assert resp.status_code == 200 assert len(records) >= 1 rec = records[-1] assert rec["method"] == "GET" assert rec["endpoint"] == "/api/cma/bot/ping" assert rec["risk_level"] == "L1" assert rec["status"] == 200 # 字段齐全 for field in ("timestamp", "bot_name", "endpoint", "method", "risk_level", "entity_id", "status"): assert field in rec, f"审计记录缺少字段: {field}" def test_audit_file_written_json_lines(self, client, tmp_path, monkeypatch): """真实日志文件：调用Bot API后 bot_audit.log 追加JSON行""" log_file = tmp_path / "bot_audit.log" monkeypatch.setenv("CMA_BOT_AUDIT_LOG", str(log_file)) monkeypatch.setattr(audit_log_module, "_audit_logger", None) resp = client.get("/api/cma/bot/ping") assert resp.status_code == 200 assert log_file.exists() lines = log_file.read_text(encoding="utf-8").strip().splitlines() assert lines, "审计日志文件为空" rec = json.loads(lines[-1]) assert rec["endpoint"] == "/api/cma/bot/ping" assert rec["risk_level"] == "L1" assert rec["status"] == 200 def test_audit_entity_id_and_bot_name(self, client, db, monkeypatch): """审计记录含 bot_name（X-BOT-KEY映射）与 entity_id""" records = [] monkeypatch.setattr(audit_log_module, "write_audit_line", lambda rec: records.append(rec)) kpi = KPIDefinition(kpi_code="AUDIT_EID", kpi_name="审计实体", dimension="finance", status="active", target_value=1.0, entity_id=1) db.add(kpi) db.commit() resp = client.post("/api/cma/bot/kpi-value-with-check", json={"kpi_id": kpi.id, "actual_value": 66.0, "period": "2026-08", "entity_id": 1}, headers=BOT_KEY) assert resp.status_code == 200 assert records, "应有审计记录" rec = records[-1] assert rec["bot_name"] == "财务BOT" assert rec["entity_id"] == 1 assert rec["risk_level"] == "L2" assert rec["status"] == 200 # JSON body 读取未破坏业务 assert resp.json()["status"] == "ok" def test_l3_batch_write_records_rows(self, client, db, monkeypatch): """L3批量写（/import）：审计记录额外含 rows 行数""" kpi = KPIDefinition(kpi_code="AUDIT_ROWS", kpi_name="审计行数", dimension="finance", status="active", target_value=1.0) db.add(kpi) db.commit() wb = Workbook() ws = wb.active ws.append(["kpi_code", "period", "actual_value"]) ws.append(["AUDIT_ROWS", "2026-08", 88.0]) ws.append(["AUDIT_ROWS", "2026-07", 77.0]) buf = io.BytesIO() wb.save(buf) records = [] monkeypatch.setattr(audit_log_module, "write_audit_line", lambda rec: records.append(rec)) files = {"file": ("kpi.xlsx", buf.getvalue(), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")} resp = client.post("/api/cma/bot/import", headers=BOT_KEY, files=files) assert resp.status_code == 200 assert resp.json()["imported"] == 2 assert records, "应有审计记录" rec = records[-1] assert rec["risk_level"] == "L3" assert rec["rows"] == 2