forked from hibiken/asynq
-
Notifications
You must be signed in to change notification settings - Fork 0
/
asynq_test.go
107 lines (94 loc) · 2.4 KB
/
asynq_test.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package asynq
import (
"encoding/json"
"math/rand"
"sort"
"testing"
"time"
"github.com/go-redis/redis/v7"
"github.com/google/go-cmp/cmp"
"github.com/google/uuid"
"github.com/hibiken/asynq/internal/rdb"
)
// This file defines test helper functions used by
// other test files.
func init() {
rand.Seed(time.Now().UnixNano())
}
// Redis keys
const (
queuePrefix = "asynq:queues:" // LIST - asynq:queues:<qname>
defaultQ = queuePrefix + "default" // LIST
scheduledQ = "asynq:scheduled" // ZSET
retryQ = "asynq:retry" // ZSET
deadQ = "asynq:dead" // ZSET
inProgressQ = "asynq:in_progress" // LIST
)
func setup(t *testing.T) *redis.Client {
t.Helper()
r := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
DB: 14,
})
// Start each test with a clean slate.
if err := r.FlushDB().Err(); err != nil {
panic(err)
}
return r
}
var sortTaskOpt = cmp.Transformer("SortMsg", func(in []*Task) []*Task {
out := append([]*Task(nil), in...) // Copy input to avoid mutating it
sort.Slice(out, func(i, j int) bool {
return out[i].Type < out[j].Type
})
return out
})
var sortMsgOpt = cmp.Transformer("SortMsg", func(in []*rdb.TaskMessage) []*rdb.TaskMessage {
out := append([]*rdb.TaskMessage(nil), in...) // Copy input to avoid mutating it
sort.Slice(out, func(i, j int) bool {
return out[i].ID.String() < out[j].ID.String()
})
return out
})
func randomTask(taskType, qname string, payload map[string]interface{}) *rdb.TaskMessage {
return &rdb.TaskMessage{
ID: uuid.New(),
Type: taskType,
Queue: qname,
Retry: defaultMaxRetry,
Payload: make(map[string]interface{}),
}
}
func mustMarshal(t *testing.T, task *rdb.TaskMessage) string {
t.Helper()
data, err := json.Marshal(task)
if err != nil {
t.Fatal(err)
}
return string(data)
}
func mustUnmarshal(t *testing.T, data string) *rdb.TaskMessage {
t.Helper()
var task rdb.TaskMessage
err := json.Unmarshal([]byte(data), &task)
if err != nil {
t.Fatal(err)
}
return &task
}
func mustMarshalSlice(t *testing.T, tasks []*rdb.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) []*rdb.TaskMessage {
t.Helper()
var tasks []*rdb.TaskMessage
for _, s := range data {
tasks = append(tasks, mustUnmarshal(t, s))
}
return tasks
}