123 lines
3.2 KiB
Go
123 lines
3.2 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.casaderoll.de/michael/urbm/internal/model"
|
|
)
|
|
|
|
func TestQueueDeduplicatesJobAndRunsTask(t *testing.T) {
|
|
done := make(chan struct{})
|
|
q := New(func(_ context.Context, run model.Run) model.Run { run.Status = "success"; close(done); return run }, nil)
|
|
run := model.Run{ID: "one", JobID: "job", TaskType: "backup", CreatedAt: time.Now()}
|
|
if err := q.Enqueue(run); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := q.Enqueue(model.Run{ID: "two", JobID: "job"}); err != ErrDuplicate {
|
|
t.Fatalf("duplicate error = %v", err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go q.Run(ctx)
|
|
select {
|
|
case <-done:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("queue did not execute task")
|
|
}
|
|
q.Stop()
|
|
}
|
|
|
|
func TestClearHistoryKeepsActiveAndPendingRuns(t *testing.T) {
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
q := New(func(_ context.Context, run model.Run) model.Run {
|
|
close(started)
|
|
<-release
|
|
run.Status = "success"
|
|
return run
|
|
}, nil)
|
|
q.RestoreHistory([]model.Run{{ID: "old", Status: "success"}})
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go q.Run(ctx)
|
|
if err := q.Enqueue(model.Run{ID: "active", JobID: "active-job"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
<-started
|
|
if err := q.Enqueue(model.Run{ID: "pending", JobID: "pending-job"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if cleared := q.ClearHistory(); cleared != 1 {
|
|
t.Fatalf("cleared = %d", cleared)
|
|
}
|
|
runs := q.Snapshot()
|
|
if len(runs) != 2 || runs[0].ID != "active" || runs[1].ID != "pending" {
|
|
t.Fatalf("runs = %#v", runs)
|
|
}
|
|
q.Stop()
|
|
close(release)
|
|
}
|
|
|
|
func TestPauseAndResumeActiveRun(t *testing.T) {
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
q := New(func(_ context.Context, run model.Run) model.Run {
|
|
close(started)
|
|
<-release
|
|
run.Status = "success"
|
|
return run
|
|
}, nil)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
go q.Run(ctx)
|
|
if err := q.Enqueue(model.Run{ID: "run", JobID: "job"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
<-started
|
|
if !q.SetPaused("run", true) {
|
|
t.Fatal("active run could not be paused")
|
|
}
|
|
if runs := q.Snapshot(); len(runs) != 1 || runs[0].Status != "paused" {
|
|
t.Fatalf("paused runs = %#v", runs)
|
|
}
|
|
if !q.SetPaused("run", false) {
|
|
t.Fatal("paused run could not be resumed")
|
|
}
|
|
if runs := q.Snapshot(); len(runs) != 1 || runs[0].Status != "running" {
|
|
t.Fatalf("resumed runs = %#v", runs)
|
|
}
|
|
close(release)
|
|
}
|
|
|
|
func TestWaitDoesNotReturnBeforeCancelledHandlerCleanup(t *testing.T) {
|
|
started := make(chan struct{})
|
|
cleanup := make(chan struct{})
|
|
q := New(func(ctx context.Context, run model.Run) model.Run {
|
|
close(started)
|
|
<-ctx.Done()
|
|
close(cleanup)
|
|
run.Status = "cancelled"
|
|
return run
|
|
}, nil)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
go q.Run(ctx)
|
|
if err := q.Enqueue(model.Run{ID: "run", JobID: "job"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
<-started
|
|
q.Stop()
|
|
waitCtx, waitCancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer waitCancel()
|
|
if err := q.Wait(waitCtx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case <-cleanup:
|
|
default:
|
|
t.Fatal("queue wait returned before handler cleanup")
|
|
}
|
|
cancel()
|
|
}
|