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

fix unit test

Browse Source 
Signed-off-by: Mohammed Sohail <sohailsameja@gmail.com>
This commit is contained in:
Phước Trung 2023-03-22 10:11:14 +07:00 committed by Ken Hibino
parent dffb78cca4
commit c72bfef094
5 changed files with 72 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
inspector_test.go
View File

@ -12,7 +12,6 @@ import (
"testing" "testing"
"time" "time"
"github.com/redis/go-redis/v9"
"github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts" "github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid" "github.com/google/uuid"
@ -20,6 +19,7 @@ import (
"github.com/hibiken/asynq/internal/rdb" "github.com/hibiken/asynq/internal/rdb"
h "github.com/hibiken/asynq/internal/testutil" h "github.com/hibiken/asynq/internal/testutil"
"github.com/hibiken/asynq/internal/timeutil" "github.com/hibiken/asynq/internal/timeutil"
"github.com/redis/go-redis/v9"
) )
func TestInspectorQueues(t *testing.T) { func TestInspectorQueues(t *testing.T) {
@ -1138,7 +1138,7 @@ func TestInspectorListAggregatingTasks(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -1152,7 +1152,7 @@ func TestInspectorListAggregatingTasks(t *testing.T) {
base.AllGroups("default"): {"group1", "group2"}, base.AllGroups("default"): {"group1", "group2"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-30 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-30 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
@ -3445,7 +3445,7 @@ func TestInspectorGroups(t *testing.T) {
fixtures := struct { fixtures := struct {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -3458,7 +3458,7 @@ func TestInspectorGroups(t *testing.T) {
base.AllGroups("default"): {"group1", "group2"}, base.AllGroups("default"): {"group1", "group2"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-10 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-10 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())},

66
internal/rdb/inspect_test.go
View File

@ -73,11 +73,11 @@ func TestCurrentStats(t *testing.T) {
allGroups map[string][]string allGroups map[string][]string
pending map[string][]string pending map[string][]string
active map[string][]string active map[string][]string
scheduled map[string][]*redis.Z scheduled map[string][]redis.Z
retry map[string][]*redis.Z retry map[string][]redis.Z
archived map[string][]*redis.Z archived map[string][]redis.Z
completed map[string][]*redis.Z completed map[string][]redis.Z
groups map[string][]*redis.Z groups map[string][]redis.Z
processed map[string]int processed map[string]int
failed map[string]int failed map[string]int
processedTotal map[string]int processedTotal map[string]int
@ -111,7 +111,7 @@ func TestCurrentStats(t *testing.T) {
base.ActiveKey("critical"): {}, base.ActiveKey("critical"): {},
base.ActiveKey("low"): {}, base.ActiveKey("low"): {},
}, },
scheduled: map[string][]*redis.Z{ scheduled: map[string][]redis.Z{
base.ScheduledKey("default"): { base.ScheduledKey("default"): {
{Member: m3.ID, Score: float64(now.Add(time.Hour).Unix())}, {Member: m3.ID, Score: float64(now.Add(time.Hour).Unix())},
{Member: m4.ID, Score: float64(now.Unix())}, {Member: m4.ID, Score: float64(now.Unix())},
@ -119,22 +119,22 @@ func TestCurrentStats(t *testing.T) {
base.ScheduledKey("critical"): {}, base.ScheduledKey("critical"): {},
base.ScheduledKey("low"): {}, base.ScheduledKey("low"): {},
}, },
retry: map[string][]*redis.Z{ retry: map[string][]redis.Z{
base.RetryKey("default"): {}, base.RetryKey("default"): {},
base.RetryKey("critical"): {}, base.RetryKey("critical"): {},
base.RetryKey("low"): {}, base.RetryKey("low"): {},
}, },
archived: map[string][]*redis.Z{ archived: map[string][]redis.Z{
base.ArchivedKey("default"): {}, base.ArchivedKey("default"): {},
base.ArchivedKey("critical"): {}, base.ArchivedKey("critical"): {},
base.ArchivedKey("low"): {}, base.ArchivedKey("low"): {},
}, },
completed: map[string][]*redis.Z{ completed: map[string][]redis.Z{
base.CompletedKey("default"): {}, base.CompletedKey("default"): {},
base.CompletedKey("critical"): {}, base.CompletedKey("critical"): {},
base.CompletedKey("low"): {}, base.CompletedKey("low"): {},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "sms:user1"): { base.GroupKey("default", "sms:user1"): {
{Member: m7.ID, Score: float64(now.Add(-3 * time.Second).Unix())}, {Member: m7.ID, Score: float64(now.Add(-3 * time.Second).Unix())},
}, },
@ -205,7 +205,7 @@ func TestCurrentStats(t *testing.T) {
base.ActiveKey("critical"): {}, base.ActiveKey("critical"): {},
base.ActiveKey("low"): {}, base.ActiveKey("low"): {},
}, },
scheduled: map[string][]*redis.Z{ scheduled: map[string][]redis.Z{
base.ScheduledKey("default"): { base.ScheduledKey("default"): {
{Member: m3.ID, Score: float64(now.Add(time.Hour).Unix())}, {Member: m3.ID, Score: float64(now.Add(time.Hour).Unix())},
{Member: m4.ID, Score: float64(now.Unix())}, {Member: m4.ID, Score: float64(now.Unix())},
@ -213,17 +213,17 @@ func TestCurrentStats(t *testing.T) {
base.ScheduledKey("critical"): {}, base.ScheduledKey("critical"): {},
base.ScheduledKey("low"): {}, base.ScheduledKey("low"): {},
}, },
retry: map[string][]*redis.Z{ retry: map[string][]redis.Z{
base.RetryKey("default"): {}, base.RetryKey("default"): {},
base.RetryKey("critical"): {}, base.RetryKey("critical"): {},
base.RetryKey("low"): {}, base.RetryKey("low"): {},
}, },
archived: map[string][]*redis.Z{ archived: map[string][]redis.Z{
base.ArchivedKey("default"): {}, base.ArchivedKey("default"): {},
base.ArchivedKey("critical"): {}, base.ArchivedKey("critical"): {},
base.ArchivedKey("low"): {}, base.ArchivedKey("low"): {},
}, },
completed: map[string][]*redis.Z{ completed: map[string][]redis.Z{
base.CompletedKey("default"): {}, base.CompletedKey("default"): {},
base.CompletedKey("critical"): {}, base.CompletedKey("critical"): {},
base.CompletedKey("low"): {}, base.CompletedKey("low"): {},
@ -435,7 +435,7 @@ func TestGroupStats(t *testing.T) {
fixtures := struct { fixtures := struct {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -448,7 +448,7 @@ func TestGroupStats(t *testing.T) {
base.AllGroups("default"): {"group1", "group2"}, base.AllGroups("default"): {"group1", "group2"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-10 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-10 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
@ -1583,7 +1583,7 @@ func TestListAggregating(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -1596,7 +1596,7 @@ func TestListAggregating(t *testing.T) {
base.AllGroups("default"): {"group1", "group2"}, base.AllGroups("default"): {"group1", "group2"},
base.AllGroups("custom"): {"group3"}, base.AllGroups("custom"): {"group3"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-30 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-30 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
@ -1663,14 +1663,14 @@ func TestListAggregatingPagination(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{}, // will be populated below tasks: []*h.TaskSeedData{}, // will be populated below
allQueues: []string{"default"}, allQueues: []string{"default"},
allGroups: map[string][]string{ allGroups: map[string][]string{
base.AllGroups("default"): {"mygroup"}, base.AllGroups("default"): {"mygroup"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
groupkey: {}, // will be populated below groupkey: {}, // will be populated below
}, },
} }
@ -1681,7 +1681,7 @@ func TestListAggregatingPagination(t *testing.T) {
fxt.tasks = append(fxt.tasks, &h.TaskSeedData{ fxt.tasks = append(fxt.tasks, &h.TaskSeedData{
Msg: msg, State: base.TaskStateAggregating, Msg: msg, State: base.TaskStateAggregating,
}) })
fxt.groups[groupkey] = append(fxt.groups[groupkey], &redis.Z{ fxt.groups[groupkey] = append(fxt.groups[groupkey], redis.Z{
Member: msg.ID, Member: msg.ID,
Score: float64(now.Add(-time.Duration(100-i) * time.Second).Unix()), Score: float64(now.Add(-time.Duration(100-i) * time.Second).Unix()),
}) })
@ -1997,7 +1997,7 @@ func TestRunAggregatingTask(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -2009,7 +2009,7 @@ func TestRunAggregatingTask(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
@ -2688,7 +2688,7 @@ func TestRunAllAggregatingTasks(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -2700,7 +2700,7 @@ func TestRunAllAggregatingTasks(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group2"}, base.AllGroups("custom"): {"group2"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
@ -2998,7 +2998,7 @@ func TestArchiveAggregatingTask(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -3010,7 +3010,7 @@ func TestArchiveAggregatingTask(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
@ -3483,7 +3483,7 @@ func TestArchiveAllAggregatingTasks(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -3495,7 +3495,7 @@ func TestArchiveAllAggregatingTasks(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group2"}, base.AllGroups("custom"): {"group2"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
@ -4122,7 +4122,7 @@ func TestDeleteAggregatingTask(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -4134,7 +4134,7 @@ func TestDeleteAggregatingTask(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},
@ -4756,7 +4756,7 @@ func TestDeleteAllAggregatingTasks(t *testing.T) {
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
allQueues []string allQueues []string
allGroups map[string][]string allGroups map[string][]string
groups map[string][]*redis.Z groups map[string][]redis.Z
}{ }{
tasks: []*h.TaskSeedData{ tasks: []*h.TaskSeedData{
{Msg: m1, State: base.TaskStateAggregating}, {Msg: m1, State: base.TaskStateAggregating},
@ -4768,7 +4768,7 @@ func TestDeleteAllAggregatingTasks(t *testing.T) {
base.AllGroups("default"): {"group1"}, base.AllGroups("default"): {"group1"},
base.AllGroups("custom"): {"group1"}, base.AllGroups("custom"): {"group1"},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "group1"): { base.GroupKey("default", "group1"): {
{Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())}, {Member: m1.ID, Score: float64(now.Add(-20 * time.Second).Unix())},
{Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())}, {Member: m2.ID, Score: float64(now.Add(-25 * time.Second).Unix())},

50
internal/rdb/rdb_test.go
View File

@ -3119,7 +3119,7 @@ func TestAggregationCheck(t *testing.T) {
desc string desc string
// initial data // initial data
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
groups map[string][]*redis.Z groups map[string][]redis.Z
allGroups map[string][]string allGroups map[string][]string
// args // args
@ -3138,7 +3138,7 @@ func TestAggregationCheck(t *testing.T) {
{ {
desc: "with an empty group", desc: "with an empty group",
tasks: []*h.TaskSeedData{}, tasks: []*h.TaskSeedData{},
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): {}, base.GroupKey("default", "mygroup"): {},
}, },
allGroups: map[string][]string{ allGroups: map[string][]string{
@ -3165,7 +3165,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3198,7 +3198,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3232,7 +3232,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg2, State: base.TaskStateAggregating}, {Msg: msg2, State: base.TaskStateAggregating},
{Msg: msg3, State: base.TaskStateAggregating}, {Msg: msg3, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3263,7 +3263,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3296,7 +3296,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3335,7 +3335,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3368,7 +3368,7 @@ func TestAggregationCheck(t *testing.T) {
{Msg: msg4, State: base.TaskStateAggregating}, {Msg: msg4, State: base.TaskStateAggregating},
{Msg: msg5, State: base.TaskStateAggregating}, {Msg: msg5, State: base.TaskStateAggregating},
}, },
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "mygroup"): { base.GroupKey("default", "mygroup"): {
{Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())}, {Member: msg1.ID, Score: float64(now.Add(-15 * time.Minute).Unix())},
{Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: msg2.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
@ -3470,8 +3470,8 @@ func TestDeleteAggregationSet(t *testing.T) {
desc string desc string
// initial data // initial data
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
aggregationSets map[string][]*redis.Z aggregationSets map[string][]redis.Z
allAggregationSets map[string][]*redis.Z allAggregationSets map[string][]redis.Z
// args // args
ctx context.Context ctx context.Context
@ -3491,14 +3491,14 @@ func TestDeleteAggregationSet(t *testing.T) {
{Msg: m2, State: base.TaskStateAggregating}, {Msg: m2, State: base.TaskStateAggregating},
{Msg: m3, State: base.TaskStateAggregating}, {Msg: m3, State: base.TaskStateAggregating},
}, },
aggregationSets: map[string][]*redis.Z{ aggregationSets: map[string][]redis.Z{
base.AggregationSetKey("default", "mygroup", setID): { base.AggregationSetKey("default", "mygroup", setID): {
{Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
{Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())}, {Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())},
{Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
}, },
}, },
allAggregationSets: map[string][]*redis.Z{ allAggregationSets: map[string][]redis.Z{
base.AllAggregationSets("default"): { base.AllAggregationSets("default"): {
{Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())}, {Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())},
}, },
@ -3525,7 +3525,7 @@ func TestDeleteAggregationSet(t *testing.T) {
{Msg: m2, State: base.TaskStateAggregating}, {Msg: m2, State: base.TaskStateAggregating},
{Msg: m3, State: base.TaskStateAggregating}, {Msg: m3, State: base.TaskStateAggregating},
}, },
aggregationSets: map[string][]*redis.Z{ aggregationSets: map[string][]redis.Z{
base.AggregationSetKey("default", "mygroup", setID): { base.AggregationSetKey("default", "mygroup", setID): {
{Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
}, },
@ -3534,7 +3534,7 @@ func TestDeleteAggregationSet(t *testing.T) {
{Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
}, },
}, },
allAggregationSets: map[string][]*redis.Z{ allAggregationSets: map[string][]redis.Z{
base.AllAggregationSets("default"): { base.AllAggregationSets("default"): {
{Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())}, {Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())},
{Member: base.AggregationSetKey("default", "mygroup", otherSetID), Score: float64(now.Add(aggregationTimeout).Unix())}, {Member: base.AggregationSetKey("default", "mygroup", otherSetID), Score: float64(now.Add(aggregationTimeout).Unix())},
@ -3599,8 +3599,8 @@ func TestDeleteAggregationSetError(t *testing.T) {
desc string desc string
// initial data // initial data
tasks []*h.TaskSeedData tasks []*h.TaskSeedData
aggregationSets map[string][]*redis.Z aggregationSets map[string][]redis.Z
allAggregationSets map[string][]*redis.Z allAggregationSets map[string][]redis.Z
// args // args
ctx context.Context ctx context.Context
@ -3619,14 +3619,14 @@ func TestDeleteAggregationSetError(t *testing.T) {
{Msg: m2, State: base.TaskStateAggregating}, {Msg: m2, State: base.TaskStateAggregating},
{Msg: m3, State: base.TaskStateAggregating}, {Msg: m3, State: base.TaskStateAggregating},
}, },
aggregationSets: map[string][]*redis.Z{ aggregationSets: map[string][]redis.Z{
base.AggregationSetKey("default", "mygroup", setID): { base.AggregationSetKey("default", "mygroup", setID): {
{Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())}, {Member: m1.ID, Score: float64(now.Add(-5 * time.Minute).Unix())},
{Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())}, {Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())},
{Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: m3.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
}, },
}, },
allAggregationSets: map[string][]*redis.Z{ allAggregationSets: map[string][]redis.Z{
base.AllAggregationSets("default"): { base.AllAggregationSets("default"): {
{Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())}, {Member: base.AggregationSetKey("default", "mygroup", setID), Score: float64(now.Add(aggregationTimeout).Unix())},
}, },
@ -3685,23 +3685,23 @@ func TestReclaimStaleAggregationSets(t *testing.T) {
// Note: In this test, we're trying out a new way to test RDB by exactly describing how // Note: In this test, we're trying out a new way to test RDB by exactly describing how
// keys and values are represented in Redis. // keys and values are represented in Redis.
tests := []struct { tests := []struct {
groups map[string][]*redis.Z // map redis-key to redis-zset groups map[string][]redis.Z // map redis-key to redis-zset
aggregationSets map[string][]*redis.Z aggregationSets map[string][]redis.Z
allAggregationSets map[string][]*redis.Z allAggregationSets map[string][]redis.Z
qname string qname string
wantGroups map[string][]redis.Z wantGroups map[string][]redis.Z
wantAggregationSets map[string][]redis.Z wantAggregationSets map[string][]redis.Z
wantAllAggregationSets map[string][]redis.Z wantAllAggregationSets map[string][]redis.Z
}{ }{
{ {
groups: map[string][]*redis.Z{ groups: map[string][]redis.Z{
base.GroupKey("default", "foo"): {}, base.GroupKey("default", "foo"): {},
base.GroupKey("default", "bar"): {}, base.GroupKey("default", "bar"): {},
base.GroupKey("default", "qux"): { base.GroupKey("default", "qux"): {
{Member: m4.ID, Score: float64(now.Add(-10 * time.Second).Unix())}, {Member: m4.ID, Score: float64(now.Add(-10 * time.Second).Unix())},
}, },
}, },
aggregationSets: map[string][]*redis.Z{ aggregationSets: map[string][]redis.Z{
base.AggregationSetKey("default", "foo", "set1"): { base.AggregationSetKey("default", "foo", "set1"): {
{Member: m1.ID, Score: float64(now.Add(-3 * time.Minute).Unix())}, {Member: m1.ID, Score: float64(now.Add(-3 * time.Minute).Unix())},
{Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())}, {Member: m2.ID, Score: float64(now.Add(-4 * time.Minute).Unix())},
@ -3710,7 +3710,7 @@ func TestReclaimStaleAggregationSets(t *testing.T) {
{Member: m3.ID, Score: float64(now.Add(-1 * time.Minute).Unix())}, {Member: m3.ID, Score: float64(now.Add(-1 * time.Minute).Unix())},
}, },
}, },
allAggregationSets: map[string][]*redis.Z{ allAggregationSets: map[string][]redis.Z{
base.AllAggregationSets("default"): { base.AllAggregationSets("default"): {
{Member: base.AggregationSetKey("default", "foo", "set1"), Score: float64(now.Add(-10 * time.Second).Unix())}, // set1 is expired {Member: base.AggregationSetKey("default", "foo", "set1"), Score: float64(now.Add(-10 * time.Second).Unix())}, // set1 is expired
{Member: base.AggregationSetKey("default", "bar", "set2"), Score: float64(now.Add(40 * time.Second).Unix())}, // set2 is not expired {Member: base.AggregationSetKey("default", "bar", "set2"), Score: float64(now.Add(40 * time.Second).Unix())}, // set2 is not expired

6
internal/testutil/testutil.go
View File

@ -13,12 +13,12 @@ import (
"testing" "testing"
"time" "time"
"github.com/redis/go-redis/v9"
"github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts" "github.com/google/go-cmp/cmp/cmpopts"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/timeutil" "github.com/hibiken/asynq/internal/timeutil"
"github.com/redis/go-redis/v9"
) )
// EquateInt64Approx returns a Comparer option that treats int64 values // EquateInt64Approx returns a Comparer option that treats int64 values
@ -377,7 +377,7 @@ func seedRedisZSet(tb testing.TB, c redis.UniversalClient, key string,
for _, item := range items { for _, item := range items {
msg := item.Message msg := item.Message
encoded := MustMarshal(tb, msg) encoded := MustMarshal(tb, msg)
z := &redis.Z{Member: msg.ID, Score: float64(item.Score)} z := redis.Z{Member: msg.ID, Score: float64(item.Score)}
if err := c.ZAdd(context.Background(), key, z).Err(); err != nil { if err := c.ZAdd(context.Background(), key, z).Err(); err != nil {
tb.Fatal(err) tb.Fatal(err)
} }
@ -570,7 +570,7 @@ func SeedTasks(tb testing.TB, r redis.UniversalClient, taskData []*TaskSeedData)
} }
} }
func SeedRedisZSets(tb testing.TB, r redis.UniversalClient, zsets map[string][]*redis.Z) { func SeedRedisZSets(tb testing.TB, r redis.UniversalClient, zsets map[string][]redis.Z) {
for key, zs := range zsets { for key, zs := range zsets {
// FIXME: How come we can't simply do ZAdd(ctx, key, zs...) here? // FIXME: How come we can't simply do ZAdd(ctx, key, zs...) here?
for _, z := range zs { for _, z := range zs {

12
x/rate/semaphore_test.go
View File

@ -8,11 +8,11 @@ import (
"testing" "testing"
"time" "time"
"github.com/redis/go-redis/v9"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/hibiken/asynq" "github.com/hibiken/asynq"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
asynqcontext "github.com/hibiken/asynq/internal/context" asynqcontext "github.com/hibiken/asynq/internal/context"
"github.com/redis/go-redis/v9"
) )
var ( var (
@ -211,7 +211,7 @@ func TestNewSemaphore_Acquire_StaleToken(t *testing.T) {
// adding a set member to mimic the case where token is acquired but the goroutine crashed, // adding a set member to mimic the case where token is acquired but the goroutine crashed,
// in which case, the token will not be explicitly removed and should be present already // in which case, the token will not be explicitly removed and should be present already
rc.ZAdd(context.Background(), semaphoreKey("stale-token"), &redis.Z{ rc.ZAdd(context.Background(), semaphoreKey("stale-token"), redis.Z{
Score: float64(time.Now().Add(-10 * time.Second).Unix()), Score: float64(time.Now().Add(-10 * time.Second).Unix()),
Member: taskID, Member: taskID,
}) })
@ -277,9 +277,9 @@ func TestNewSemaphore_Release(t *testing.T) {
t.Errorf("%s;\nredis.UniversalClient.Del() got error %v", tt.desc, err) t.Errorf("%s;\nredis.UniversalClient.Del() got error %v", tt.desc, err)
} }
var members []*redis.Z var members []redis.Z
for i := 0; i < len(tt.taskIDs); i++ { for i := 0; i < len(tt.taskIDs); i++ {
members = append(members, &redis.Z{ members = append(members, redis.Z{
Score: float64(time.Now().Add(time.Duration(i) * time.Second).Unix()), Score: float64(time.Now().Add(time.Duration(i) * time.Second).Unix()),
Member: tt.taskIDs[i], Member: tt.taskIDs[i],
}) })
@ -356,9 +356,9 @@ func TestNewSemaphore_Release_Error(t *testing.T) {
t.Errorf("%s;\nredis.UniversalClient.Del() got error %v", tt.desc, err) t.Errorf("%s;\nredis.UniversalClient.Del() got error %v", tt.desc, err)
} }
var members []*redis.Z var members []redis.Z
for i := 0; i < len(tt.taskIDs); i++ { for i := 0; i < len(tt.taskIDs); i++ {
members = append(members, &redis.Z{ members = append(members, redis.Z{
Score: float64(time.Now().Add(time.Duration(i) * time.Second).Unix()), Score: float64(time.Now().Add(time.Duration(i) * time.Second).Unix()),
Member: tt.taskIDs[i], Member: tt.taskIDs[i],
}) })