48 lines
1.2 KiB
Go
48 lines
1.2 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.casaderoll.de/michael/urbm/internal/model"
|
|
)
|
|
|
|
func TestUpdateActiveIsVisibleWithoutPersistingIntermediateState(t *testing.T) {
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
persistCalls := 0
|
|
q := New(func(_ context.Context, run model.Run) model.Run {
|
|
close(started)
|
|
<-release
|
|
run.Status = "success"
|
|
return run
|
|
}, func([]model.Run) { persistCalls++ })
|
|
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
|
|
before := persistCalls
|
|
if !q.UpdateActive("run", func(run *model.Run) { run.ProgressPercent = 42 }) {
|
|
t.Fatal("active run was not updated")
|
|
}
|
|
if runs := q.Snapshot(); len(runs) != 1 || runs[0].ProgressPercent != 42 {
|
|
t.Fatalf("runs = %#v", runs)
|
|
}
|
|
if persistCalls != before {
|
|
t.Fatal("intermediate progress was persisted")
|
|
}
|
|
close(release)
|
|
deadline := time.Now().Add(time.Second)
|
|
for time.Now().Before(deadline) {
|
|
if runs := q.Snapshot(); len(runs) == 1 && runs[0].Status == "success" {
|
|
return
|
|
}
|
|
time.Sleep(time.Millisecond)
|
|
}
|
|
t.Fatal("run did not finish")
|
|
}
|