99 lines
3.1 KiB
Go
99 lines
3.1 KiB
Go
package cli
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
|
|
kafkamgmtv1 "forgejo.riotpiao.com/rock/kmsvc-proto/gen/kafkamgmt/v1"
|
|
)
|
|
|
|
func TestDLQRedriveHappyPath(t *testing.T) {
|
|
var sendCalled, deleteCalled atomic.Bool
|
|
var sendQueue string
|
|
|
|
fake := &fakeQueueService{
|
|
receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
|
return &kafkamgmtv1.ReceiveMessageResponse{
|
|
Messages: []*kafkamgmtv1.Message{{MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("payload")}},
|
|
}, nil
|
|
},
|
|
sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
|
sendCalled.Store(true)
|
|
sendQueue = req.QueueName
|
|
if deleteCalled.Load() {
|
|
t.Error("delete called before send completed")
|
|
}
|
|
return &kafkamgmtv1.SendMessageResponse{MessageId: "m1-redriven"}, nil
|
|
},
|
|
deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
|
deleteCalled.Store(true)
|
|
if !sendCalled.Load() {
|
|
t.Error("delete called before send")
|
|
}
|
|
return &kafkamgmtv1.DeleteMessageResponse{}, nil
|
|
},
|
|
}
|
|
addr := startTestServer(t, fake)
|
|
|
|
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
|
cmd := newDLQRedriveCmd(flags)
|
|
cmd.SetArgs([]string{"--queue", "orders.dlq", "--to", "orders"})
|
|
|
|
var out bytes.Buffer
|
|
cmd.SetOut(&out)
|
|
if err := cmd.Execute(); err != nil {
|
|
t.Fatalf("Execute: %v", err)
|
|
}
|
|
if !sendCalled.Load() || !deleteCalled.Load() {
|
|
t.Fatal("expected both send and delete to be called")
|
|
}
|
|
if sendQueue != "orders" {
|
|
t.Errorf("send queue = %q, want orders", sendQueue)
|
|
}
|
|
}
|
|
|
|
func TestDLQRedriveSurfacesDeleteFailure(t *testing.T) {
|
|
fake := &fakeQueueService{
|
|
receiveMessage: func(ctx context.Context, req *kafkamgmtv1.ReceiveMessageRequest) (*kafkamgmtv1.ReceiveMessageResponse, error) {
|
|
return &kafkamgmtv1.ReceiveMessageResponse{
|
|
Messages: []*kafkamgmtv1.Message{{MessageId: "m1", ReceiptHandle: "rh1", Body: []byte("payload")}},
|
|
}, nil
|
|
},
|
|
sendMessage: func(ctx context.Context, req *kafkamgmtv1.SendMessageRequest) (*kafkamgmtv1.SendMessageResponse, error) {
|
|
return &kafkamgmtv1.SendMessageResponse{MessageId: "m1-redriven"}, nil
|
|
},
|
|
deleteMessage: func(ctx context.Context, req *kafkamgmtv1.DeleteMessageRequest) (*kafkamgmtv1.DeleteMessageResponse, error) {
|
|
return nil, errBoom
|
|
},
|
|
}
|
|
addr := startTestServer(t, fake)
|
|
|
|
flags := &globalFlags{server: addr, output: "table", insecure: true}
|
|
cmd := newDLQRedriveCmd(flags)
|
|
cmd.SetArgs([]string{"--queue", "orders.dlq", "--to", "orders"})
|
|
cmd.SilenceUsage = true
|
|
cmd.SilenceErrors = true
|
|
|
|
var out bytes.Buffer
|
|
cmd.SetOut(&out)
|
|
|
|
err := cmd.Execute()
|
|
if err == nil {
|
|
t.Fatal("expected redrive to report failure when delete fails")
|
|
}
|
|
|
|
output := out.String()
|
|
if !strings.Contains(output, "may be duplicated") {
|
|
t.Errorf("output = %q, want a duplicate-risk warning", output)
|
|
}
|
|
}
|
|
|
|
var errBoom = &boomError{}
|
|
|
|
type boomError struct{}
|
|
|
|
func (e *boomError) Error() string { return "boom" }
|