71 lines
2.2 KiB
Go
71 lines
2.2 KiB
Go
package natscap
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sendico.io/tech/sxchnge/shared/messaging/capability"
|
|
"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
func TestFetchPollDeadlineIsBounded(t *testing.T) {
|
|
parent, cancel := context.WithTimeout(context.Background(), time.Hour)
|
|
defer cancel()
|
|
ctx, stop, derived := fetchPollContext(parent, time.Second)
|
|
if stop != nil {
|
|
defer stop()
|
|
}
|
|
deadline, ok := ctx.Deadline()
|
|
if !ok || !derived || time.Until(deadline) > time.Second {
|
|
t.Fatal("fetch escaped poll deadline")
|
|
}
|
|
}
|
|
func TestClosedSubscriptionAndAck(t *testing.T) {
|
|
s := &natsSubscription{}
|
|
if err := s.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := s.Fetch(context.Background(), 1, time.Second); !errors.Is(err, capability.ErrClosed) {
|
|
t.Fatal(err)
|
|
}
|
|
d := &natsDelivery{sub: s}
|
|
if err := d.Ack(context.Background()); !errors.Is(err, capability.ErrClosed) {
|
|
t.Fatal(err)
|
|
}
|
|
d.acked = true
|
|
if err := d.Ack(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
func TestConsumerDriftRejected(t *testing.T) {
|
|
want := nats.ConsumerConfig{Durable: "email", FilterSubject: "notifications.email", AckPolicy: nats.AckExplicitPolicy, DeliverPolicy: nats.DeliverAllPolicy, AckWait: time.Minute}
|
|
current := want
|
|
current.FilterSubject = "other"
|
|
if err := validateConsumerConfig(current, want, true); err == nil {
|
|
t.Fatal("subject drift accepted")
|
|
}
|
|
current = want
|
|
current.AckWait = time.Second
|
|
if err := validateConsumerConfig(current, want, true); err == nil {
|
|
t.Fatal("ack drift accepted")
|
|
}
|
|
if _, err := natsDeliverPolicy(reliable.DurableDeliverPolicy("invalid")); err == nil {
|
|
t.Fatal("invalid deliver policy accepted")
|
|
}
|
|
}
|
|
|
|
func TestProductionStreamRetentionDriftRejected(t *testing.T) {
|
|
expected := nats.StreamConfig{Name: "NOTIFICATIONS", Subjects: []string{"notifications.email"}, Storage: nats.FileStorage, Retention: nats.LimitsPolicy, Replicas: 3, MaxAge: 48 * time.Hour}
|
|
current := expected
|
|
current.MaxAge = time.Hour
|
|
if err := validateStreamConfig(current, expected, true); err == nil {
|
|
t.Fatal("production retention drift accepted")
|
|
}
|
|
if err := validateStreamConfig(current, expected, false); err != nil {
|
|
t.Fatal("development retention cannot be reconciled", err)
|
|
}
|
|
}
|