测试指南
1 概述
Asynq 的 Handler 本质上是纯函数 func(ctx context.Context, t *Task) error，因此测试非常直观——不需要启动 Redis 或 Server，直接构造 Task 和 Context 调用 Handler 即可。
2 Handler 单元测试
2.1 基础测试模式
func TestHandleEmailTask(t *testing.T) {
// 1. 构造 Payload
payload, _ := json.Marshal(EmailPayload{
UserID: 42,
TemplateID: "welcome",
})
// 2. 创建 Task
task := asynq.NewTask("email:send", payload)
// 3. 准备 context
ctx := context.Background()
// 4. 调用 Handler
err := HandleEmailTask(ctx, task)
// 5. 断言
if err != nil {
t.Errorf("unexpected error: %v", err)
}
}2.2 带依赖注入的测试
func TestEmailHandler_ProcessTask(t *testing.T) {
// 使用 mock
mailer := &MockMailer{}
handler := &EmailHandler{mailer: mailer}
payload, _ := json.Marshal(EmailPayload{UserID: 42, TemplateID: "welcome"})
task := asynq.NewTask("email:send", payload)
err := handler.ProcessTask(context.Background(), task)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
if !mailer.SendCalled {
t.Error("expected mailer.Send to be called")
}
}
type MockMailer struct {
SendCalled bool
}
func (m *MockMailer) Send(ctx context.Context, to, subject, body string) error {
m.SendCalled = true
return nil
}2.3 测试错误处理
func TestHandleEmailTask_InvalidPayload(t *testing.T) {
// 构造无效 Payload
task := asynq.NewTask("email:send", []byte("invalid json"))
err := HandleEmailTask(context.Background(), task)
if err == nil {
t.Fatal("expected error, got nil")
}
if !errors.Is(err, asynq.SkipRetry) {
t.Errorf("expected SkipRetry, got: %v", err)
}
}
func TestHandleEmailTask_TemporaryError(t *testing.T) {
payload, _ := json.Marshal(EmailPayload{UserID: 42, TemplateID: "welcome"})
task := asynq.NewTask("email:send", payload)
// 模拟临时错误
handler := &EmailHandler{
mailer: &FailingMailer{err: errors.New("connection refused")},
}
err := handler.ProcessTask(context.Background(), task)
if err == nil {
t.Fatal("expected error, got nil")
}
if errors.Is(err, asynq.SkipRetry) {
t.Error("unexpected SkipRetry for temporary error")
}
}2.4 前后对比测试
func TestHandleEmailTask(t *testing.T) {
tests := []struct {
name string
payload interface{}
wantErr bool
wantSkipRetry bool
}{
{
name: "valid payload",
payload: EmailPayload{UserID: 42, TemplateID: "welcome"},
wantErr: false,
},
{
name: "invalid json",
payload: "not-json",
wantErr: true,
wantSkipRetry: true,
},
{
name: "missing user",
payload: EmailPayload{UserID: 0, TemplateID: "welcome"},
wantErr: true,
wantSkipRetry: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
payload, _ := json.Marshal(tt.payload)
task := asynq.NewTask("email:send", payload)
err := HandleEmailTask(context.Background(), task)
if (err != nil) != tt.wantErr {
t.Errorf("error = %v, wantErr = %v", err, tt.wantErr)
}
if tt.wantSkipRetry && !errors.Is(err, asynq.SkipRetry) {
t.Errorf("expected SkipRetry, got: %v", err)
}
})
}
}3 测试超时行为
func TestHandler_Timeout(t *testing.T) {
payload, _ := json.Marshal(MyPayload{})
task := asynq.NewTask("slow:task", payload)
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
err := SlowHandler(ctx, task)
if !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("expected DeadlineExceeded, got: %v", err)
}
}4 测试中间件
func TestLoggingMiddleware(t *testing.T) {
var buf bytes.Buffer
logger := log.New(&buf, "", 0)
// 创建一个简单的中间件
mid := func(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
logger.Printf("before: %s", t.Type())
err := next.ProcessTask(ctx, t)
logger.Printf("after: %s err=%v", t.Type(), err)
return err
})
}
// 包装 Handler
handler := mid(asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
return nil
}))
task := asynq.NewTask("test:task", []byte(`{}`))
err := handler.ProcessTask(context.Background(), task)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
if !strings.Contains(buf.String(), "before: test:task") {
t.Error("expected 'before' log")
}
}5 集成测试
5.1 使用真实 Redis
func TestIntegration_EnqueueAndProcess(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test")
}
// 连接测试 Redis（使用独立 DB）
redisOpt := asynq.RedisClientOpt{
Addr: "localhost:6379",
DB: 15, // 使用独立 DB，隔离测试数据
}
// 清理
inspector := asynq.NewInspector(redisOpt)
defer inspector.Close()
// 1. 入队
client := asynq.NewClient(redisOpt)
defer client.Close()
task := asynq.NewTask("test:echo", []byte(`{"msg":"hello"}`))
info, err := client.Enqueue(task,
asynq.MaxRetry(0),
asynq.Retention(1*time.Minute),
)
if err != nil {
t.Fatal(err)
}
// 2. 等待处理
time.Sleep(2 * time.Second)
// 3. 查询状态
taskInfo, err := inspector.GetTaskInfo("default", info.ID)
if err != nil {
t.Fatal(err)
}
t.Logf("Task state: %s", taskInfo.State)
}5.2 完整端到端测试
func TestE2E_TaskLifecycle(t *testing.T) {
if testing.Short() {
t.Skip("skipping e2e test")
}
redisOpt := asynq.RedisClientOpt{Addr: "localhost:6379", DB: 15}
// 启动 Server
var processed atomic.Bool
mux := asynq.NewServeMux()
mux.HandleFunc("test:ping", func(ctx context.Context, t *asynq.Task) error {
processed.Store(true)
return nil
})
srv := asynq.NewServer(redisOpt, asynq.Config{Concurrency: 1})
if err := srv.Start(mux); err != nil {
t.Fatal(err)
}
defer srv.Shutdown()
// 入队
client := asynq.NewClient(redisOpt)
defer client.Close()
task := asynq.NewTask("test:ping", []byte(`{}`))
_, err := client.Enqueue(task, asynq.MaxRetry(0))
if err != nil {
t.Fatal(err)
}
// 等待处理
time.Sleep(3 * time.Second)
if !processed.Load() {
t.Error("task was not processed")
}
}6 Mock 与 Stub 模式
6.1 Mock Redis Client
对于不需要真实 Redis 的测试，可以抽象接口：
// 定义接口
type TaskClient interface {
Enqueue(task *asynq.Task, opts ...asynq.Option) (*asynq.TaskInfo, error)
Close() error
}
// 生产实现
type RedisTaskClient struct {
client *asynq.Client
}
func (c *RedisTaskClient) Enqueue(task *asynq.Task, opts ...asynq.Option) (*asynq.TaskInfo, error) {
return c.client.Enqueue(task, opts...)
}
func (c *RedisTaskClient) Close() error {
return c.client.Close()
}
// 测试 Mock
type MockTaskClient struct {
EnqueueCalled bool
LastTask *asynq.Task
}
func (m *MockTaskClient) Enqueue(task *asynq.Task, opts ...asynq.Option) (*asynq.TaskInfo, error) {
m.EnqueueCalled = true
m.LastTask = task
return &asynq.TaskInfo{ID: "mock-id", Queue: "default"}, nil
}
func (m *MockTaskClient) Close() error { return nil }6.2 测试业务逻辑（不涉及 Asynq）
// 业务逻辑独立于 Asynq Handler
func ProcessEmail(ctx context.Context, userID int, templateID string) error {
// 实际业务逻辑
return nil
}
// Handler 仅负责解码和调用
func HandleEmailTask(ctx context.Context, t *asynq.Task) error {
var p EmailPayload
if err := json.Unmarshal(t.Payload(), &p); err != nil {
return fmt.Errorf("unmarshal: %w", asynq.SkipRetry)
}
return ProcessEmail(ctx, p.UserID, p.TemplateID)
}
// 测试业务逻辑时不需要构造 Task
func TestProcessEmail(t *testing.T) {
err := ProcessEmail(context.Background(), 42, "welcome")
if err != nil {
t.Errorf("unexpected error: %v", err)
}
}💡 最佳实践：将业务逻辑从 Handler 中分离。Handler 只负责 Payload 解析和错误转换，核心逻辑放在独立的函数中便于测试。
7 测试重试逻辑
func TestRetryDecision(t *testing.T) {
tests := []struct {
name string
err error
wantRetry bool
wantSkipRetry bool
}{
{
name: "network timeout → retry",
err: errors.New("dial tcp: i/o timeout"),
wantRetry: true,
},
{
name: "invalid payload → skip retry",
err: fmt.Errorf("unmarshal: %w", asynq.SkipRetry),
wantSkipRetry: true,
},
{
name: "user not found → skip retry",
err: fmt.Errorf("user not found: %w", asynq.SkipRetry),
wantSkipRetry: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
isSkipRetry := errors.Is(tt.err, asynq.SkipRetry)
if tt.wantSkipRetry != isSkipRetry {
t.Errorf("SkipRetry = %v, want %v", isSkipRetry, tt.wantSkipRetry)
}
})
}
}8 测试辅助函数
// 快速创建测试 Task
func newTestTask(typename string, payload interface{}) *asynq.Task {
data, _ := json.Marshal(payload)
return asynq.NewTask(typename, data)
}
// 添加 Retry 判断辅助
func mustSkipRetry(t *testing.T, err error) {
t.Helper()
if !errors.Is(err, asynq.SkipRetry) {
t.Errorf("expected SkipRetry, got: %v", err)
}
}
func shouldRetry(t *testing.T, err error) {
t.Helper()
if errors.Is(err, asynq.SkipRetry) {
t.Errorf("expected retry, but got SkipRetry")
}
}