53 lines
1.7 KiB
Go
53 lines
1.7 KiB
Go
package mq
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"testing"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
func TestConsumerRequestStopDoesNotWaitForItself(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
consumer := &Consumer{cancel: cancel, done: make(chan struct{})}
|
|
consumer.RequestStop()
|
|
select {
|
|
case <-ctx.Done():
|
|
default:
|
|
t.Fatal("consumer cancellation was not requested")
|
|
}
|
|
consumer.RequestStop()
|
|
}
|
|
|
|
type currentRejectRecorder struct {
|
|
rejected, acked, nacked int
|
|
requeued bool
|
|
}
|
|
|
|
func (a *currentRejectRecorder) Ack(uint64, bool) error { a.acked++; return nil }
|
|
func (a *currentRejectRecorder) Nack(_ uint64, _ bool, requeue bool) error {
|
|
a.nacked++
|
|
a.requeued = requeue
|
|
return nil
|
|
}
|
|
func (a *currentRejectRecorder) Reject(uint64, bool) error { a.rejected++; return nil }
|
|
|
|
func TestMalformedControlWithoutDurableHandlerRequeues(t *testing.T) {
|
|
id := "c046b893-8628-4589-ae50-619d049248a6"
|
|
route, err := tenant.ControlRoute(id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
broker := &Broker{dispatcherID: id, controlQueue: route.Queue}
|
|
ack := ¤tRejectRecorder{}
|
|
deliveries := make(chan amqp.Delivery, 1)
|
|
deliveries <- amqp.Delivery{Body: []byte("{"), RoutingKey: route.BindingKey, Acknowledger: ack}
|
|
close(deliveries)
|
|
err = broker.consumeDeliveries(context.Background(), id, route.Queue, deliveries, func(context.Context, string, []byte) error { t.Fatal("invalid control reached handler"); return nil })
|
|
if err == nil || !strings.Contains(err.Error(), "no durable rejection handler") || ack.rejected != 0 || ack.nacked != 1 || !ack.requeued {
|
|
t.Fatalf("malformed control was silently discarded before admission: rejected=%d err=%v", ack.rejected, err)
|
|
}
|
|
}
|