97 lines
3.3 KiB
Go
97 lines
3.3 KiB
Go
package app
|
|
|
|
import (
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"syscall"
|
|
"testing"
|
|
"time"
|
|
|
|
"agent-system-backend/internal/store"
|
|
core "agent-system-core"
|
|
)
|
|
|
|
func TestTaskStartupReconcilesQueueBeforeServing(t *testing.T) {
|
|
if os.Getenv("TASK_STARTUP_CHILD") == "1" {
|
|
if err := Run(Options{Addr: os.Getenv("TASK_STARTUP_ADDR"), DataDir: os.Getenv("TASK_STARTUP_DATA"), PluginsDir: filepath.Join(os.Getenv("TASK_STARTUP_DATA"), "plugins"), ForceMock: true, InstanceID: "startup-test", DisableExecTools: true, DisableConvCardMirror: true}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return
|
|
}
|
|
data := t.TempDir()
|
|
st, err := store.OpenSQLite(filepath.Join(data, "agent-system.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { st.Close() })
|
|
p, err := st.CreateProject(store.OwnerDefault, "startup")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
conv, err := st.CreateConversation(store.OwnerDefault, p.ID, "startup", "")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
card := core.ActivityCard{ID: "card-startup", ProjectID: p.ID, ConversationID: conv.ID, Kind: core.CardKindProgress, Status: core.CardActive, Title: "整理"}
|
|
card.Task = &core.TaskSnapshot{SchemaVersion: 1, TaskID: "task-startup", AttemptID: "attempt-startup", CardID: card.ID, ProjectID: p.ID, ConversationID: conv.ID, OwnerInstanceID: "startup-test", Revision: 1, StreamScope: "conversation:" + conv.ID, ExecutionState: "queued", OutcomeCompleteness: "none", DeliveryState: "pending", Title: "整理"}
|
|
if _, err = st.CommitTaskProjection(store.TaskProjectionWrite{Owner: conv.Owner, Card: card}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
addr := ln.Addr().String()
|
|
ln.Close()
|
|
cmd := exec.Command(os.Args[0], "-test.run=^TestTaskStartupReconcilesQueueBeforeServing$")
|
|
cmd.Env = append(os.Environ(), "TASK_STARTUP_CHILD=1", "TASK_STARTUP_ADDR="+addr, "TASK_STARTUP_DATA="+data)
|
|
logfile, err := os.Create(filepath.Join(t.TempDir(), "startup.log"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer logfile.Close()
|
|
cmd.Stdout, cmd.Stderr = logfile, logfile
|
|
if err = cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
_ = cmd.Process.Signal(syscall.SIGTERM)
|
|
done := make(chan error, 1)
|
|
go func() { done <- cmd.Wait() }()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
_ = cmd.Process.Kill()
|
|
<-done
|
|
}
|
|
})
|
|
client := &http.Client{Timeout: time.Second}
|
|
deadline := time.Now().Add(10 * time.Second)
|
|
for time.Now().Before(deadline) {
|
|
resp, err := client.Get("http://" + addr + "/api/conversations/" + conv.ID + "/cards")
|
|
if err == nil {
|
|
resp.Body.Close()
|
|
if resp.StatusCode == http.StatusOK {
|
|
got, err := st.GetCard(conv.Owner, card.ID)
|
|
if err != nil || got.Task == nil || got.Task.ExecutionState != "interrupted" || got.Task.RunID != "" {
|
|
t.Fatalf("service became ready with orphan queue: %+v, %v", got.Task, err)
|
|
}
|
|
if got.Task.Conclusion != "排队任务未恢复，未开始执行" {
|
|
t.Fatal(got.Task.Conclusion)
|
|
}
|
|
projectEvents, err := st.EventsAfter("project:"+p.ID, 0, 100)
|
|
if err != nil || len(projectEvents) != 0 {
|
|
t.Fatalf("recovery ignored disabled card mirroring: %v %v", projectEvents, err)
|
|
}
|
|
return
|
|
}
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
b, _ := os.ReadFile(logfile.Name())
|
|
t.Fatalf("startup did not become ready: %s", b)
|
|
}
|