feat: 增加 Agent Excel 批量测试

ai_knowledge_base_v2/apps/backend/tests/test_admin_permissions.py

@@ -13,6 +13,7 @@ def test_super_admin_has_all_permissions() -> None:
     admin = Admin(id=1, username="root", password="unused", name="root", status=1, is_super_admin=1)
     assert permissions_for(admin) == ALL_PERMISSION_CODES
     assert {"feedback.view", "feedback.detail", "feedback.export", "feedback.delete"} <= ALL_PERMISSION_CODES
+    assert "prompt.batch" in ALL_PERMISSION_CODES
 
 
 def test_role_permissions_are_restricted_to_catalog() -> None:

ai_knowledge_base_v2/apps/backend/tests/test_agent_batch_tests.py

@@ -0,0 +1,125 @@
+from io import BytesIO
+import json
+
+import pytest
+from fastapi import HTTPException, UploadFile
+from openpyxl import Workbook, load_workbook
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+from sqlalchemy.pool import StaticPool
+
+from app.models import Base
+from app.models.admin import Admin
+from app.models.agent_batch import AgentBatchTestItem
+from app.models.ai_config import ModelConfig
+from app.services.agent_batch_test_service import AgentBatchTestService, MAX_BATCH_QUESTIONS
+
+
+def _database() -> Session:
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(engine)
+    return Session(engine)
+
+
+def _admin() -> Admin:
+    return Admin(id=1, username="admin", password="hash", name="系统管理员", status=1)
+
+
+def _model() -> ModelConfig:
+    return ModelConfig(
+        id=1,
+        provider="mock",
+        display_name="批量测试模型",
+        api_type="openai_compatible",
+        model_name="batch-model",
+        api_url="",
+        api_key="",
+        auth_type="bearer",
+        enabled=1,
+    )
+
+
+def test_template_can_be_downloaded_and_imported() -> None:
+    stream = AgentBatchTestService.template_workbook()
+    workbook = load_workbook(stream, data_only=True)
+
+    assert workbook.sheetnames == ["批量测试问题", "填写说明"]
+    assert [cell.value for cell in workbook["批量测试问题"][1]] == ["序号", "问题"]
+    assert workbook["填写说明"]["B3"].value == f"最多 {MAX_BATCH_QUESTIONS} 个问题"
+
+    upload = UploadFile(filename="批量测试.xlsx", file=BytesIO(stream.getvalue()))
+    rows = AgentBatchTestService.parse_questions(upload)
+    assert len(rows) == 2
+    assert rows[0]["externalNo"] == "1"
+    assert "测试的问题" in rows[0]["question"]
+
+
+def test_create_job_snapshots_config_and_export_keeps_failed_rows() -> None:
+    with _database() as db:
+        admin = _admin()
+        model = _model()
+        db.add_all([admin, model])
+        db.commit()
+        job = AgentBatchTestService.create_job(
+            db,
+            admin=admin,
+            filename="回归问题.xlsx",
+            rows=[
+                {"rowNumber": 2, "externalNo": "Q-1", "question": "第一个问题"},
+                {"rowNumber": 3, "externalNo": "Q-2", "question": "第二个问题"},
+            ],
+            config={
+                "name": "提示词回归测试",
+                "promptContent": "只根据知识库回答",
+                "modelId": model.id,
+                "knowledgeIds": [],
+                "temperature": 0.2,
+                "maxToken": 8192,
+                "responseDepth": 35,
+            },
+        )
+        items = list(db.scalars(select(AgentBatchTestItem).order_by(AgentBatchTestItem.row_number)))
+        items[0].status = "success"
+        items[0].answer = "第一个答案"
+        items[0].model_name = "批量测试模型"
+        items[0].retrieve_count = 3
+        items[0].duration_ms = 1250
+        items[1].status = "failed"
+        items[1].error_message = "供应商超时"
+        job.status = "completed_with_errors"
+        job.processed_count = 2
+        job.success_count = 1
+        job.failed_count = 1
+        db.commit()
+
+        assert json.loads(job.generation_config)["maxToken"] == 8192
+        assert json.loads(job.knowledge_names) == ["全部正式开放知识库"]
+
+        exported = AgentBatchTestService.export_workbook(job, items)
+        workbook = load_workbook(exported, data_only=True)
+        sheet = workbook["批量测试结果"]
+        assert sheet["C2"].value == "第一个答案"
+        assert sheet["D2"].value == "成功"
+        assert sheet["D3"].value == "失败"
+        assert sheet["E3"].value == "供应商超时"
+        assert sheet.freeze_panes == "A2"
+
+
+def test_import_rejects_excel_formula_question() -> None:
+    workbook = Workbook()
+    sheet = workbook.active
+    sheet.append(["序号", "问题"])
+    sheet.append([1, '=HYPERLINK("https://example.com","问题")'])
+    stream = BytesIO()
+    workbook.save(stream)
+    stream.seek(0)
+
+    with pytest.raises(HTTPException) as exc:
+        AgentBatchTestService.parse_questions(UploadFile(filename="危险问题.xlsx", file=stream))
+
+    assert exc.value.status_code == 400
+    assert "不能使用 Excel 公式" in exc.value.detail