Files
Agent-System/internal/server/task_run_observer_test.go
weichen 3145550a59 feat(security): complete P0 release blocking remediation
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.
2026-09-13 02:50:22 +00:00

96 lines
2.7 KiB
Go

package server
import (
"bufio"
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/config"
"chat-agent-system/internal/stream"
"chat-agent-system/internal/taskrun"
)
func TestTaskRun_DisconnectDoesNotCancelWorker(t *testing.T) {
cfg := &config.Config{
Env: "test",
AgentBackend: "stub",
DataDir: t.TempDir(),
}
svc := bizdev.MustNewService(t, cfg)
srv := NewServer(cfg, svc)
ts := httptest.NewServer(srv.Routes())
defer ts.Close()
runID := "run_disconnect_test"
record := &taskrun.RunRecord{RunID: runID, TaskKey: "intake", Status: "running"}
taskReq := bizdev.TaskRunRequest{TaskKey: "intake", ProjectName: "disconnect-test"}
_, err := srv.taskRuns.CreateAndStart(context.Background(), runID, record, taskReq, bizdev.StageRunRequest{}, "")
if err != nil {
t.Fatalf("CreateAndStart: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/api/v1/tasks/runs/"+runID+"/events", nil)
req.Header.Set("Accept", "text/event-stream")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("subscribe failed: %v", err)
}
time.Sleep(200 * time.Millisecond)
cancel()
_ = resp.Body.Close()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
rec, getErr := srv.taskRuns.GetRun(runID)
if getErr == nil && rec.Status != "running" {
if rec.Status == "failed" {
t.Fatalf("worker marked failed after disconnect, status=%s err=%s", rec.Status, rec.ErrorMessage)
}
return
}
time.Sleep(100 * time.Millisecond)
}
t.Fatal("worker did not finish after disconnect")
}
func TestCreateTaskRun_RequiresSessionProject(t *testing.T) {
cfg := &config.Config{Env: "test", DataDir: t.TempDir(), AgentBackend: "stub"}
srv := NewServer(cfg, bizdev.MustNewService(t, cfg))
ts := httptest.NewServer(srv.Routes())
defer ts.Close()
body, _ := json.Marshal(map[string]any{"task_key": "intake"})
resp, err := http.Post(ts.URL+"/api/v1/tasks/runs", "application/json", bytes.NewReader(body))
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusBadRequest {
t.Fatalf("status=%d want 400", resp.StatusCode)
}
}
func readFirstEventType(t *testing.T, resp *http.Response) stream.EventType {
t.Helper()
scanner := bufio.NewScanner(resp.Body)
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: ") && currentType != "" {
return stream.EventType(currentType)
}
}
return ""
}