logan
2336c9433c
- Agent 支持 SSE（步骤边推、正文分块），JSON 非流式 fallback - tool calling 覆盖 openai-chat / anthropic / responses（单测解析回灌） - P1：mcp.json mode=read_write 才可写；需 confirm_danger；Agent 剥离模型自带令牌 - GET /api/ai/mcp-status、POST /api/ai/agent/confirm；审计 source=mcp|agent
73 lines
1.9 KiB
Go
73 lines
1.9 KiB
Go
package ai
|
|
|
|
import (
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
|
|
"opsxdb/server/internal/aitools"
|
|
)
|
|
|
|
func TestParseToolsTurnAnthropic(t *testing.T) {
|
|
raw, _ := json.Marshal(map[string]any{
|
|
"stop_reason": "tool_use",
|
|
"content": []map[string]any{
|
|
{"type": "text", "text": "先列表"},
|
|
{"type": "tool_use", "id": "tu_1", "name": "opsxdb_list_tables", "input": map[string]any{"connection_id": "c1"}},
|
|
},
|
|
})
|
|
turn, err := parseToolsTurnAnthropic(raw, 200)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if turn.Content != "先列表" {
|
|
t.Fatalf("content=%q", turn.Content)
|
|
}
|
|
if len(turn.ToolCalls) != 1 || turn.ToolCalls[0].Function.Name != "opsxdb_list_tables" {
|
|
t.Fatalf("calls=%+v", turn.ToolCalls)
|
|
}
|
|
if !strings.Contains(turn.ToolCalls[0].Function.Arguments, "c1") {
|
|
t.Fatalf("args=%s", turn.ToolCalls[0].Function.Arguments)
|
|
}
|
|
}
|
|
|
|
func TestParseToolsTurnResponses(t *testing.T) {
|
|
raw, _ := json.Marshal(map[string]any{
|
|
"output": []map[string]any{
|
|
{"type": "function_call", "call_id": "call_9", "name": "opsxdb_execute_query", "arguments": `{"connection_id":"c1","sql":"SELECT 1"}`},
|
|
{"type": "message", "content": []map[string]any{{"type": "output_text", "text": "done"}}},
|
|
},
|
|
})
|
|
turn, err := parseToolsTurnResponses(raw, 200)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(turn.ToolCalls) != 1 || turn.ToolCalls[0].ID != "call_9" {
|
|
t.Fatalf("calls=%+v", turn.ToolCalls)
|
|
}
|
|
if turn.Content != "done" {
|
|
t.Fatalf("content=%q", turn.Content)
|
|
}
|
|
}
|
|
|
|
func TestOpenAIToolsToAnthropic(t *testing.T) {
|
|
tools := aitools.OpenAITools()
|
|
anth := openaiToolsToAnthropic(tools)
|
|
if len(anth) != len(tools) {
|
|
t.Fatalf("%d vs %d", len(anth), len(tools))
|
|
}
|
|
if anth[0]["name"] != tools[0].Function.Name {
|
|
t.Fatal(anth[0]["name"])
|
|
}
|
|
if anth[0]["input_schema"] == nil {
|
|
t.Fatal("missing input_schema")
|
|
}
|
|
}
|
|
|
|
func TestChunkDelta(t *testing.T) {
|
|
parts := chunkDelta("abcdefghij", 3)
|
|
if len(parts) != 4 {
|
|
t.Fatalf("%v", parts)
|
|
}
|
|
}
|