236 lines
7.0 KiB
Go
236 lines
7.0 KiB
Go
package task
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec"
|
|
)
|
|
|
|
// waitTerminalNoFail polls until terminal without failing the test immediately.
|
|
func waitTerminalNoFail(tk *Task, timeout time.Duration) Snapshot {
|
|
s, _ := waitTerminal(tk, timeout)
|
|
return s
|
|
}
|
|
|
|
// TestManager_Abort_CancelsTask_NoLeak verifies Abort marks the task aborted,
|
|
// releases the run goroutine, and that a second Abort is ErrAlreadyDone.
|
|
func TestManager_Abort_CancelsTask_NoLeak(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockRun("never", nil, 10*time.Second))
|
|
|
|
tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"sleep"}, Background: true})
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
waitActive(t, m, 1, 2*time.Second)
|
|
|
|
if err := m.Abort(context.Background(), tk.ID); err != nil {
|
|
t.Fatalf("Abort: %v", err)
|
|
}
|
|
if snap := waitTerminalNoFail(tk, 2*time.Second); snap.Status != StatusAborted {
|
|
t.Errorf("Status=%s, want aborted", snap.Status)
|
|
}
|
|
|
|
if err := m.Abort(context.Background(), tk.ID); !errors.Is(err, ErrAlreadyDone) {
|
|
t.Errorf("second Abort: err=%v, want ErrAlreadyDone", err)
|
|
}
|
|
|
|
// The run goroutine must have exited (no leak).
|
|
waitActive(t, m, 0, 2*time.Second)
|
|
}
|
|
|
|
// TestManager_ListGet verifies List and Get behavior, including the not-found path.
|
|
func TestManager_ListGet(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockRun("x", nil, 0))
|
|
|
|
a, err := m.Start(context.Background(), TaskSpec{Command: []string{"a"}, Background: true})
|
|
if err != nil {
|
|
t.Fatalf("Start a: %v", err)
|
|
}
|
|
b, err := m.Start(context.Background(), TaskSpec{Command: []string{"b"}, Background: true})
|
|
if err != nil {
|
|
t.Fatalf("Start b: %v", err)
|
|
}
|
|
|
|
list, err := m.List(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("List: %v", err)
|
|
}
|
|
if len(list) != 2 {
|
|
t.Fatalf("List len=%d, want 2", len(list))
|
|
}
|
|
found := map[string]bool{}
|
|
for _, tk := range list {
|
|
found[tk.ID] = true
|
|
}
|
|
if !found[a.ID] || !found[b.ID] {
|
|
t.Errorf("List missing tasks: %v", found)
|
|
}
|
|
|
|
got, err := m.Get(context.Background(), a.ID)
|
|
if err != nil {
|
|
t.Fatalf("Get: %v", err)
|
|
}
|
|
if got.ID != a.ID {
|
|
t.Errorf("Get ID=%q, want %q", got.ID, a.ID)
|
|
}
|
|
|
|
if _, err := m.Get(context.Background(), "nope"); !errors.Is(err, ErrNotFound) {
|
|
t.Errorf("Get unknown: err=%v, want ErrNotFound", err)
|
|
}
|
|
if err := m.Abort(context.Background(), "nope"); !errors.Is(err, ErrNotFound) {
|
|
t.Errorf("Abort unknown: err=%v, want ErrNotFound", err)
|
|
}
|
|
}
|
|
|
|
// TestManager_TaskOutputTruncated verifies truncation and exit code propagation.
|
|
func TestManager_TaskOutputTruncated(t *testing.T) {
|
|
t.Parallel()
|
|
p := mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) {
|
|
return &exec.ExecResult{Stdout: "truncated", Stderr: "err", ExitCode: 1, Truncated: true}, nil
|
|
}}
|
|
m := newTestManager(t, p)
|
|
|
|
tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"big"}, Background: true})
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
snap, ok := waitTerminal(tk, 2*time.Second)
|
|
if !ok {
|
|
t.Fatalf("task did not finish, status=%s", tk.Status())
|
|
}
|
|
if !snap.Truncated {
|
|
t.Errorf("Truncated=false, want true")
|
|
}
|
|
if snap.ExitCode != 1 {
|
|
t.Errorf("ExitCode=%d, want 1", snap.ExitCode)
|
|
}
|
|
if snap.Output.Stderr != "err" {
|
|
t.Errorf("Stderr=%q, want err", snap.Output.Stderr)
|
|
}
|
|
}
|
|
|
|
// TestManager_TaskFailed verifies a provider error surfaces as a failed task and
|
|
// is echoed in the done event.
|
|
func TestManager_TaskFailed(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockRun("", exec.ErrCommandNotAllowed, 0))
|
|
doneCh, unsub := subscribeDone(t, m)
|
|
defer unsub()
|
|
|
|
tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"bad"}, Background: true})
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
snap, ok := waitTerminal(tk, 2*time.Second)
|
|
if !ok {
|
|
t.Fatalf("task did not finish, status=%s", tk.Status())
|
|
}
|
|
if snap.Status != StatusFailed {
|
|
t.Errorf("Status=%s, want failed", snap.Status)
|
|
}
|
|
if snap.Err == "" {
|
|
t.Errorf("Err is empty, want an error string")
|
|
}
|
|
|
|
select {
|
|
case e := <-doneCh:
|
|
if e.Status != StatusFailed {
|
|
t.Errorf("done Status=%s, want failed", e.Status)
|
|
}
|
|
if e.Error == "" {
|
|
t.Errorf("done Error is empty")
|
|
}
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("no task.done event for failed task")
|
|
}
|
|
}
|
|
|
|
// TestManager_Start_EmptyCommand verifies Start rejects an empty command.
|
|
func TestManager_Start_EmptyCommand(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockProvider{})
|
|
if _, err := m.Start(context.Background(), TaskSpec{}); !errors.Is(err, ErrEmptyCommand) {
|
|
t.Errorf("err=%v, want ErrEmptyCommand", err)
|
|
}
|
|
}
|
|
|
|
// TestManager_Start_AfterClose verifies Start refuses after Close.
|
|
func TestManager_Start_AfterClose(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockProvider{})
|
|
if err := m.Close(); err != nil {
|
|
t.Fatalf("Close: %v", err)
|
|
}
|
|
if _, err := m.Start(context.Background(), TaskSpec{Command: []string{"a"}, Background: true}); !errors.Is(err, ErrManagerClosed) {
|
|
t.Errorf("err=%v, want ErrManagerClosed", err)
|
|
}
|
|
}
|
|
|
|
// TestManager_WithAutoBackgroundThreshold verifies the promotion threshold is
|
|
// configurable via WithAutoBackground.
|
|
func TestManager_WithAutoBackgroundThreshold(t *testing.T) {
|
|
t.Parallel()
|
|
m, err := New(&config.Config{}, mockRun("x", nil, 500*time.Millisecond), discardLogger(), WithAutoBackground(50*time.Millisecond))
|
|
if err != nil {
|
|
t.Fatalf("New: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = m.Close() })
|
|
|
|
start := time.Now()
|
|
tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"s"}, Background: false})
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
if elapsed := time.Since(start); elapsed >= 500*time.Millisecond {
|
|
t.Errorf("Start waited %v; want promotion well before 500ms", elapsed)
|
|
}
|
|
if tk.Kind != KindBackground {
|
|
t.Errorf("Kind=%s, want background after promotion", tk.Kind)
|
|
}
|
|
if s, ok := waitTerminal(tk, 2*time.Second); !ok || s.Status != StatusDone {
|
|
t.Errorf("failed to finish after promotion, snap=%+v", s)
|
|
}
|
|
}
|
|
|
|
// TestManager_ConcurrentTasks runs many background tasks concurrently and
|
|
// verifies none leak or corrupt shared state.
|
|
func TestManager_ConcurrentTasks(t *testing.T) {
|
|
t.Parallel()
|
|
m := newTestManager(t, mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
return &exec.ExecResult{Stdout: req.Command[0], ExitCode: 0}, nil
|
|
}})
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < 20; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"cmd"}, Background: true})
|
|
if err != nil {
|
|
t.Errorf("Start: %v", err)
|
|
return
|
|
}
|
|
if s, ok := waitTerminal(tk, 2*time.Second); !ok || s.Status != StatusDone {
|
|
t.Errorf("concurrent task not done: %+v", s)
|
|
}
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
if n := m.Active(); n != 0 {
|
|
t.Errorf("Active()=%d, want 0 (no leaked goroutines)", n)
|
|
}
|
|
}
|