weichen
f8730791b9
聊天入口只把项目 ID 放在授权字段，协调器仍读空的 UserCtx.Project.ID，受控任务因此以 project_id=0 跳过账本投影，回复却称正文已保存。
255 lines
8.2 KiB
Go
255 lines
8.2 KiB
Go
package server
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"encoding/json"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"os"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"chat-agent-system/internal/bizdev"
|
||
"chat-agent-system/internal/config"
|
||
"chat-agent-system/internal/stream"
|
||
)
|
||
|
||
// readSSEEvents 从 SSE 响应解析事件序列（测试断言辅助）
|
||
func readSSEEvents(t *testing.T, resp *http.Response) []stream.StreamEvent {
|
||
t.Helper()
|
||
scanner := bufio.NewScanner(resp.Body)
|
||
var events []stream.StreamEvent
|
||
var currentType string
|
||
for scanner.Scan() {
|
||
line := scanner.Text()
|
||
if strings.HasPrefix(line, "event: ") {
|
||
currentType = strings.TrimPrefix(line, "event: ")
|
||
} else if strings.HasPrefix(line, "data: ") {
|
||
var ev stream.StreamEvent
|
||
if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err == nil {
|
||
if ev.Type == "" {
|
||
ev.Type = stream.EventType(currentType)
|
||
}
|
||
events = append(events, ev)
|
||
}
|
||
}
|
||
}
|
||
return events
|
||
}
|
||
|
||
func findEvent(events []stream.StreamEvent, eventType stream.EventType) *stream.StreamEvent {
|
||
for i := range events {
|
||
if events[i].Type == eventType {
|
||
return &events[i]
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func hasAction(actions []string, want string) bool {
|
||
for _, a := range actions {
|
||
if a == want {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// TestTaskStream_TransientInjection_DeadPiAgent 会话级 transient 故障注入：
|
||
// pi-agent 指向不可达端口 → dial 失败 → SSE 必须输出 event:error 且 error_info.class=transient
|
||
// （补齐 2026-09-09 测试报告遗留建议 #2）
|
||
func TestTaskStream_TransientInjection_DeadPiAgent(t *testing.T) {
|
||
cfg := &config.Config{
|
||
Env: "test",
|
||
AgentBackend: "stub",
|
||
DataDir: t.TempDir(),
|
||
KbDbPort: 5432,
|
||
AdminUsername: "admin",
|
||
AdminPassword: "admin-test-pass",
|
||
PiAgentRunnerURL: "http://127.0.0.1:1", // 死端口：connection refused → 网络类瞬态错误
|
||
PiAgentModel: "MiniMax-M3",
|
||
PiAgentProvider: "minimax",
|
||
AgentTimeout: 5 * time.Second,
|
||
}
|
||
svc := bizdev.MustNewService(t, cfg)
|
||
srv := NewServer(cfg, svc)
|
||
|
||
ts := httptest.NewServer(srv.Routes())
|
||
defer ts.Close()
|
||
|
||
body, _ := json.Marshal(map[string]any{
|
||
"task_key": "intake",
|
||
"project_id": 1,
|
||
"project_name": "超声内镜系统",
|
||
"customer": "北京水木东方",
|
||
"backend": "pi-agent",
|
||
})
|
||
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
|
||
resp, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("post tasks run stream failed: %v", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if rid := resp.Header.Get("X-Request-Id"); rid == "" {
|
||
t.Fatal("expected non-empty X-Request-Id header")
|
||
}
|
||
|
||
events := readSSEEvents(t, resp)
|
||
errEv := findEvent(events, stream.EventError)
|
||
if errEv == nil {
|
||
t.Fatalf("expected event:error in stream, got %d events", len(events))
|
||
}
|
||
if errEv.ErrorInfo == nil {
|
||
t.Fatal("error_info is nil")
|
||
}
|
||
if errEv.ErrorInfo.Class != stream.ErrorClassTransient {
|
||
t.Fatalf("class=%q want transient (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream)
|
||
}
|
||
if !hasAction(errEv.ErrorInfo.Actions, "retry") {
|
||
t.Fatalf("actions=%v want retry", errEv.ErrorInfo.Actions)
|
||
}
|
||
if errEv.ErrorInfo.RequestID == "" {
|
||
t.Fatal("error_info.request_id is empty")
|
||
}
|
||
if errEv.ErrorInfo.PhaseID == "" {
|
||
t.Fatal("error_info.phase_id is empty (expected last streamed phase)")
|
||
}
|
||
}
|
||
|
||
// TestTaskStream_TypedReject_ClassifiedFatal runner 契约失败（typed result rejected）：
|
||
// 即使携带 WorkDir，也必须 class=fatal、仅 restart，不得伪装成可 continue 的中断。
|
||
func TestTaskStream_TypedReject_ClassifiedFatal(t *testing.T) {
|
||
brokenRunner := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||
http.Error(w, "simulated upstream business failure", http.StatusInternalServerError)
|
||
}))
|
||
defer brokenRunner.Close()
|
||
|
||
cfg := &config.Config{
|
||
Env: "test",
|
||
AgentBackend: "stub",
|
||
DataDir: t.TempDir(),
|
||
KbDbPort: 5432,
|
||
AdminUsername: "admin",
|
||
AdminPassword: "admin-test-pass",
|
||
PiAgentRunnerURL: brokenRunner.URL,
|
||
PiAgentModel: "MiniMax-M3",
|
||
PiAgentProvider: "minimax",
|
||
AgentTimeout: 10 * time.Second,
|
||
}
|
||
svc := bizdev.MustNewService(t, cfg)
|
||
srv := NewServer(cfg, svc)
|
||
|
||
ts := httptest.NewServer(srv.Routes())
|
||
defer ts.Close()
|
||
|
||
body, _ := json.Marshal(map[string]any{
|
||
"task_key": "intake",
|
||
"project_id": 1,
|
||
"project_name": "超声内镜系统",
|
||
"customer": "北京水木东方",
|
||
"backend": "pi-agent",
|
||
})
|
||
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
|
||
resp, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("post tasks run stream failed: %v", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
events := readSSEEvents(t, resp)
|
||
errEv := findEvent(events, stream.EventError)
|
||
if errEv == nil {
|
||
t.Fatalf("expected event:error in stream, got %d events", len(events))
|
||
}
|
||
if errEv.ErrorInfo == nil {
|
||
t.Fatal("error_info is nil")
|
||
}
|
||
if errEv.ErrorInfo.Class != stream.ErrorClassFatal {
|
||
t.Fatalf("class=%q want fatal (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream)
|
||
}
|
||
if errEv.ErrorInfo.Code != "RUNNER_RESULT_REJECTED" {
|
||
t.Fatalf("code=%q want RUNNER_RESULT_REJECTED", errEv.ErrorInfo.Code)
|
||
}
|
||
if hasAction(errEv.ErrorInfo.Actions, "continue") {
|
||
t.Fatalf("actions=%v must not include continue for typed reject", errEv.ErrorInfo.Actions)
|
||
}
|
||
if !hasAction(errEv.ErrorInfo.Actions, "restart") {
|
||
t.Fatalf("actions=%v want restart", errEv.ErrorInfo.Actions)
|
||
}
|
||
}
|
||
|
||
// TestTaskStream_MaterialsInjection 验证前端传入的 materials（Base64 编码的附件内容）
|
||
// 能被后端 HTTP handler 正确反序列化，透传给 ExecuteTask，并写入工作区的 materials_text/ 目录
|
||
func TestTaskStream_MaterialsInjection(t *testing.T) {
|
||
cfg := &config.Config{
|
||
Env: "test",
|
||
AgentBackend: "stub",
|
||
DataDir: t.TempDir(),
|
||
KbDbPort: 5432,
|
||
AdminUsername: "admin",
|
||
AdminPassword: "admin-test-pass",
|
||
}
|
||
svc := bizdev.MustNewService(t, cfg)
|
||
srv := NewServer(cfg, svc)
|
||
|
||
ts := httptest.NewServer(srv.Routes())
|
||
defer ts.Close()
|
||
|
||
// 模拟前端发来的包含 materials 字段的请求
|
||
// Content 经过 Base64 编码："用户上传文档核心需求：需要支持 1080P 超声实时渲染。"
|
||
const rawText = "用户上传文档核心需求：需要支持 1080P 超声实时渲染。"
|
||
b64Content := "55So5oi35LiK5Lyg5paH5qGj5qC45b+D6ZyA5rGC77ya6ZyA6KaB5pSv5oyBIDEwODBQIOi2heWjsOWunuaXtua4suafk+OAgg=="
|
||
|
||
body, _ := json.Marshal(map[string]any{
|
||
"task_key": "intake",
|
||
"project_id": 1,
|
||
"project_name": "超声材料注入测试",
|
||
"customer_name": "北京水木东方",
|
||
"device_type": "超声诊断设备",
|
||
"backend": "stub",
|
||
"materials": []map[string]string{
|
||
{
|
||
"filename": "initial_requirements.txt",
|
||
"content": b64Content,
|
||
},
|
||
},
|
||
})
|
||
|
||
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run/stream", bytes.NewReader(body))
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Authorization", "Bearer "+testBearerToken(t, ts.URL, cfg))
|
||
resp, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
t.Fatalf("post tasks run stream failed: %v", err)
|
||
}
|
||
defer resp.Body.Close()
|
||
|
||
if resp.StatusCode != http.StatusOK {
|
||
t.Fatalf("expected status 200, got %d", resp.StatusCode)
|
||
}
|
||
|
||
events := readSSEEvents(t, resp)
|
||
doneEv := findEvent(events, stream.EventDone)
|
||
if doneEv == nil || doneEv.Done == nil || doneEv.Done.WorkspaceID == "" {
|
||
t.Fatalf("expected event:done with workspace_id, got %d events", len(events))
|
||
}
|
||
workDir := doneEv.Done.WorkspaceID
|
||
|
||
// 验证 materials_text 目录下是否存在 initial_requirements.txt 且内容还原正确
|
||
savedFile := workDir + "/context/materials_text/initial_requirements.txt"
|
||
content, err := os.ReadFile(savedFile)
|
||
if err != nil {
|
||
t.Fatalf("expected material file at %s: %v", savedFile, err)
|
||
}
|
||
if string(content) != rawText {
|
||
t.Fatalf("material content mismatch: got %q, want %q", string(content), rawText)
|
||
}
|
||
}
|