go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-27 00:02:19 +08:00
Code Issues Projects Releases Wiki Activity

Refactor rdb tests

Browse Source
This commit is contained in:
Ken Hibino 2019-11-28 07:17:07 -08:00
parent e5e3fee742
commit 481d3c6b77
1 changed files with 65 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
rdb_test.go
View File

@ -17,12 +17,6 @@ func init() {
rand.Seed(time.Now().UnixNano()) rand.Seed(time.Now().UnixNano())
} }
var sortStrOpt = cmp.Transformer("SortStr", func(in []string) []string {
out := append([]string(nil), in...) // Copy input to avoid mutating it
sort.Strings(out)
return out
})
var sortMsgOpt = cmp.Transformer("SortMsg", func(in []*taskMessage) []*taskMessage { var sortMsgOpt = cmp.Transformer("SortMsg", func(in []*taskMessage) []*taskMessage {
out := append([]*taskMessage(nil), in...) // Copy input to avoid mutating it out := append([]*taskMessage(nil), in...) // Copy input to avoid mutating it
sort.Slice(out, func(i, j int) bool { sort.Slice(out, func(i, j int) bool {
@ -75,6 +69,24 @@ func mustUnmarshal(t *testing.T, data string) *taskMessage {
return &task return &task
} }
func mustMarshalSlice(t *testing.T, tasks []*taskMessage) []string {
t.Helper()
var data []string
for _, task := range tasks {
data = append(data, mustMarshal(t, task))
}
return data
}
func mustUnmarshalSlice(t *testing.T, data []string) []*taskMessage {
t.Helper()
var tasks []*taskMessage
for _, s := range data {
tasks = append(tasks, mustUnmarshal(t, s))
}
return tasks
}
func TestEnqueue(t *testing.T) { func TestEnqueue(t *testing.T) {
r := setup(t) r := setup(t)
tests := []struct { tests := []struct {
@ -105,12 +117,7 @@ func TestEnqueue(t *testing.T) {
if !r.client.SIsMember(allQueues, defaultQueue).Val() { if !r.client.SIsMember(allQueues, defaultQueue).Val() {
t.Errorf("SISMEMBER %q %q = false, want true", allQueues, defaultQueue) t.Errorf("SISMEMBER %q %q = false, want true", allQueues, defaultQueue)
} }
var persisted taskMessage if diff := cmp.Diff(*tc.msg, *mustUnmarshal(t, res[0])); diff != "" {
if err := json.Unmarshal([]byte(res[0]), &persisted); err != nil {
t.Error(err)
continue
}
if diff := cmp.Diff(*tc.msg, persisted); diff != "" {
t.Errorf("persisted data differed from the original input (-want, +got)\n%s", diff) t.Errorf("persisted data differed from the original input (-want, +got)\n%s", diff)
} }
} }
@ -212,42 +219,30 @@ func TestMoveAll(t *testing.T) {
t1 := randomTask("send_email", "default", nil) t1 := randomTask("send_email", "default", nil)
t2 := randomTask("export_csv", "csv", nil) t2 := randomTask("export_csv", "csv", nil)
t3 := randomTask("sync_stuff", "sync", nil) t3 := randomTask("sync_stuff", "sync", nil)
json1, err := json.Marshal(t1)
if err != nil {
t.Fatal(err)
}
json2, err := json.Marshal(t2)
if err != nil {
t.Fatal(err)
}
json3, err := json.Marshal(t3)
if err != nil {
t.Fatal(err)
}
tests := []struct { tests := []struct {
beforeSrc []string beforeSrc []*taskMessage
beforeDst []string beforeDst []*taskMessage
afterSrc []string afterSrc []*taskMessage
afterDst []string afterDst []*taskMessage
}{ }{
{ {
beforeSrc: []string{string(json1), string(json2), string(json3)}, beforeSrc: []*taskMessage{t1, t2, t3},
beforeDst: []string{}, beforeDst: []*taskMessage{},
afterSrc: []string{}, afterSrc: []*taskMessage{},
afterDst: []string{string(json1), string(json2), string(json3)}, afterDst: []*taskMessage{t1, t2, t3},
}, },
{ {
beforeSrc: []string{}, beforeSrc: []*taskMessage{},
beforeDst: []string{string(json1), string(json2), string(json3)}, beforeDst: []*taskMessage{t1, t2, t3},
afterSrc: []string{}, afterSrc: []*taskMessage{},
afterDst: []string{string(json1), string(json2), string(json3)}, afterDst: []*taskMessage{t1, t2, t3},
}, },
{ {
beforeSrc: []string{string(json2), string(json3)}, beforeSrc: []*taskMessage{t2, t3},
beforeDst: []string{string(json1)}, beforeDst: []*taskMessage{t1},
afterSrc: []string{}, afterSrc: []*taskMessage{},
afterDst: []string{string(json1), string(json2), string(json3)}, afterDst: []*taskMessage{t1, t2, t3},
}, },
} }
@ -259,11 +254,11 @@ func TestMoveAll(t *testing.T) {
} }
// seed src list. // seed src list.
for _, msg := range tc.beforeSrc { for _, msg := range tc.beforeSrc {
r.client.LPush(inProgress, msg) r.client.LPush(inProgress, mustMarshal(t, msg))
} }
// seed dst list. // seed dst list.
for _, msg := range tc.beforeDst { for _, msg := range tc.beforeDst {
r.client.LPush(defaultQueue, msg) r.client.LPush(defaultQueue, mustMarshal(t, msg))
} }
if err := r.moveAll(inProgress, defaultQueue); err != nil { if err := r.moveAll(inProgress, defaultQueue); err != nil {
@ -271,12 +266,14 @@ func TestMoveAll(t *testing.T) {
continue continue
} }
gotSrc := r.client.LRange(inProgress, 0, -1).Val() src := r.client.LRange(inProgress, 0, -1).Val()
if diff := cmp.Diff(tc.afterSrc, gotSrc, sortStrOpt); diff != "" { gotSrc := mustUnmarshalSlice(t, src)
if diff := cmp.Diff(tc.afterSrc, gotSrc, sortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q (-want, +got)\n%s", inProgress, diff) t.Errorf("mismatch found in %q (-want, +got)\n%s", inProgress, diff)
} }
gotDst := r.client.LRange(defaultQueue, 0, -1).Val() dst := r.client.LRange(defaultQueue, 0, -1).Val()
if diff := cmp.Diff(tc.afterDst, gotDst, sortStrOpt); diff != "" { gotDst := mustUnmarshalSlice(t, dst)
if diff := cmp.Diff(tc.afterDst, gotDst, sortMsgOpt); diff != "" {
t.Errorf("mismatch found in %q (-want, +got)\n%s", defaultQueue, diff) t.Errorf("mismatch found in %q (-want, +got)\n%s", defaultQueue, diff)
} }
} }
@ -286,42 +283,34 @@ func TestForward(t *testing.T) {
r := setup(t) r := setup(t)
t1 := randomTask("send_email", defaultQueue, nil) t1 := randomTask("send_email", defaultQueue, nil)
t2 := randomTask("generate_csv", defaultQueue, nil) t2 := randomTask("generate_csv", defaultQueue, nil)
json1, err := json.Marshal(t1)
if err != nil {
t.Fatal(err)
}
json2, err := json.Marshal(t2)
if err != nil {
t.Fatal(err)
}
secondAgo := time.Now().Add(-time.Second) secondAgo := time.Now().Add(-time.Second)
hourFromNow := time.Now().Add(time.Hour) hourFromNow := time.Now().Add(time.Hour)
tests := []struct { tests := []struct {
tasks []*redis.Z // scheduled tasks with timestamp as a score tasks []*redis.Z // scheduled tasks with timestamp as a score
wantQueued []string // queue after calling forward wantQueued []*taskMessage // queue after calling forward
wantScheduled []string // scheduled queue after calling forward wantScheduled []*taskMessage // scheduled queue after calling forward
}{ }{
{ {
tasks: []*redis.Z{ tasks: []*redis.Z{
&redis.Z{Member: string(json1), Score: float64(secondAgo.Unix())}, &redis.Z{Member: mustMarshal(t, t1), Score: float64(secondAgo.Unix())},
&redis.Z{Member: string(json2), Score: float64(secondAgo.Unix())}}, &redis.Z{Member: mustMarshal(t, t2), Score: float64(secondAgo.Unix())}},
wantQueued: []string{string(json1), string(json2)}, wantQueued: []*taskMessage{t1, t2},
wantScheduled: []string{}, wantScheduled: []*taskMessage{},
}, },
{ {
tasks: []*redis.Z{ tasks: []*redis.Z{
&redis.Z{Member: string(json1), Score: float64(hourFromNow.Unix())}, &redis.Z{Member: mustMarshal(t, t1), Score: float64(hourFromNow.Unix())},
&redis.Z{Member: string(json2), Score: float64(secondAgo.Unix())}}, &redis.Z{Member: mustMarshal(t, t2), Score: float64(secondAgo.Unix())}},
wantQueued: []string{string(json2)}, wantQueued: []*taskMessage{t2},
wantScheduled: []string{string(json1)}, wantScheduled: []*taskMessage{t1},
}, },
{ {
tasks: []*redis.Z{ tasks: []*redis.Z{
&redis.Z{Member: string(json1), Score: float64(hourFromNow.Unix())}, &redis.Z{Member: mustMarshal(t, t1), Score: float64(hourFromNow.Unix())},
&redis.Z{Member: string(json2), Score: float64(hourFromNow.Unix())}}, &redis.Z{Member: mustMarshal(t, t2), Score: float64(hourFromNow.Unix())}},
wantQueued: []string{}, wantQueued: []*taskMessage{},
wantScheduled: []string{string(json1), string(json2)}, wantScheduled: []*taskMessage{t1, t2},
}, },
} }
@ -335,18 +324,20 @@ func TestForward(t *testing.T) {
continue continue
} }
err = r.forward(scheduled) err := r.forward(scheduled)
if err != nil { if err != nil {
t.Errorf("(*rdb).forward(%q) = %v, want nil", scheduled, err) t.Errorf("(*rdb).forward(%q) = %v, want nil", scheduled, err)
continue continue
} }
gotQueued := r.client.LRange(defaultQueue, 0, -1).Val() queued := r.client.LRange(defaultQueue, 0, -1).Val()
if diff := cmp.Diff(tc.wantQueued, gotQueued, sortStrOpt); diff != "" { gotQueued := mustUnmarshalSlice(t, queued)
if diff := cmp.Diff(tc.wantQueued, gotQueued, sortMsgOpt); diff != "" {
t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", defaultQueue, len(gotQueued), len(tc.wantQueued), diff) t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", defaultQueue, len(gotQueued), len(tc.wantQueued), diff)
continue continue
} }
gotScheduled := r.client.ZRangeByScore(scheduled, &redis.ZRangeBy{Min: "-inf", Max: "+inf"}).Val() scheduled := r.client.ZRangeByScore(scheduled, &redis.ZRangeBy{Min: "-inf", Max: "+inf"}).Val()
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, sortStrOpt); diff != "" { gotScheduled := mustUnmarshalSlice(t, scheduled)
if diff := cmp.Diff(tc.wantScheduled, gotScheduled, sortMsgOpt); diff != "" {
t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", scheduled, len(gotScheduled), len(tc.wantScheduled), diff) t.Errorf("%q has %d tasks, want %d tasks; (-want, +got)\n%s", scheduled, len(gotScheduled), len(tc.wantScheduled), diff)
continue continue
} }