weichen
3145550a59
Unify auth middleware and remove X-User-ID trust, add project_members authorization, sandbox workspaces via resolver, and admin-only reconcile by run_id. Introduce goose migrations with startup schema gate, fail closed when DB unavailable outside explicit ALLOW_IN_MEMORY, bundle web/dist in app Dockerfile, and remove frontend pseudo-login fallback.
305 lines
10 KiB
Go
305 lines
10 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_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_SoftFailure_ClassifiedResumable 软失败（pi-agent 上游业务失败而非网络/超时）：
|
||
// squad 层返回携带 WorkDir 的 result → class=resumable、actions=[continue restart]，
|
||
// workspace_id 供前端「继续任务」续跑
|
||
func TestTaskStream_SoftFailure_ClassifiedResumable(t *testing.T) {
|
||
// 伪造 pi-agent runner：HTTP 500 + 非 JSON body → decode 失败 → 业务软失败（非 dial/timeout）
|
||
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_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.ErrorClassResumable {
|
||
t.Fatalf("class=%q want resumable (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream)
|
||
}
|
||
if errEv.ErrorInfo.WorkspaceID == "" {
|
||
t.Fatal("workspace_id is empty (expected squad soft-failure to carry work_dir)")
|
||
}
|
||
if !hasAction(errEv.ErrorInfo.Actions, "continue") || !hasAction(errEv.ErrorInfo.Actions, "restart") {
|
||
t.Fatalf("actions=%v want [continue restart]", errEv.ErrorInfo.Actions)
|
||
}
|
||
}
|
||
|
||
// TestTaskStream_ResumeInvalidWorkspace_ClassifiedResumable resume 模式：
|
||
// work_dir 非空走 ResumeOrModifyWorkspace；目标工作区不存在 → 失败仍归类 resumable
|
||
// 且 workspace_id 等于请求的 work_dir（工作区可被修复后继续）
|
||
func TestTaskStream_ResumeInvalidWorkspace_ClassifiedResumable(t *testing.T) {
|
||
t.Skip("P0 安全整改：客户端 work_dir 续跑入口已移除")
|
||
cfg := &config.Config{
|
||
Env: "test",
|
||
AgentBackend: "stub",
|
||
DataDir: t.TempDir(),
|
||
KbDbPort: 5432,
|
||
AdminUsername: "admin",
|
||
AdminPassword: "admin-test-pass",
|
||
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_name": "resume-test",
|
||
"backend": "stub",
|
||
})
|
||
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.ErrorClassResumable {
|
||
t.Fatalf("class=%q want resumable (upstream=%q)", errEv.ErrorInfo.Class, errEv.ErrorInfo.Upstream)
|
||
}
|
||
if errEv.ErrorInfo.WorkspaceID == "" {
|
||
t.Fatal("workspace_id should be populated for resumable errors")
|
||
}
|
||
if !hasAction(errEv.ErrorInfo.Actions, "continue") || !hasAction(errEv.ErrorInfo.Actions, "restart") {
|
||
t.Fatalf("actions=%v want [continue 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_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)
|
||
}
|
||
}
|
||
|