kra-new/pkg/mq/subscription_test.go

package mq

import (
	"context"
	"errors"
	"reflect"
	"strings"
	"testing"
)

func TestNormalizeSubscriptionSetValidDeclaration(t *testing.T) {
	handler := func(context.Context, Message) {}
	set := SubscriptionSet{
		Owner:    "  orders  ",
		Provider: "  RABBITMQ ",
		Topics: []TopicSubscription{
			{Topic: "  orders.created  ", QoS: AtMostOnce, Handler: handler},
			{Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler},
			{Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler},
		},
	}

	got, err := NormalizeSubscriptionSet(set)
	if err != nil {
		t.Fatalf("NormalizeSubscriptionSet() error = %v", err)
	}

	want := SubscriptionSet{
		Owner:    "orders",
		Provider: ProviderRabbitMQ,
		Topics: []TopicSubscription{
			{Topic: "orders.created", QoS: AtMostOnce, Handler: handler},
			{Topic: "orders.updated", QoS: AtLeastOnce, Handler: handler},
			{Topic: "orders.deleted", QoS: AtLeastOnce, Handler: handler},
		},
	}
	if got.Owner != want.Owner || got.Provider != want.Provider {
		t.Fatalf("normalized identity = %#v, want owner=%q provider=%q", got, want.Owner, want.Provider)
	}
	if len(got.Topics) != len(want.Topics) {
		t.Fatalf("normalized topic count = %d, want %d", len(got.Topics), len(want.Topics))
	}
	for index := range want.Topics {
		if got.Topics[index].Topic != want.Topics[index].Topic {
			t.Errorf("topic[%d] = %q, want %q", index, got.Topics[index].Topic, want.Topics[index].Topic)
		}
		if got.Topics[index].QoS != want.Topics[index].QoS {
			t.Errorf("qos[%d] = %d, want %d", index, got.Topics[index].QoS, want.Topics[index].QoS)
		}
		if reflect.ValueOf(got.Topics[index].Handler).Pointer() != reflect.ValueOf(want.Topics[index].Handler).Pointer() {
			t.Errorf("handler[%d] was changed", index)
		}
	}
}

func TestNormalizeSubscriptionSetAcceptsMQTTQoS2(t *testing.T) {
	set, err := NormalizeSubscriptionSet(SubscriptionSet{
		Owner:    "devices",
		Provider: ProviderEMQX,
		Topics:   []TopicSubscription{{Topic: "devices/status", QoS: ExactlyOnce, Handler: func(context.Context, Message) {}}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if set.Topics[0].QoS != ExactlyOnce {
		t.Fatalf("qos = %d, want %d", set.Topics[0].QoS, ExactlyOnce)
	}
}

func TestNormalizeSubscriptionSetAcceptsKafka(t *testing.T) {
	set, err := NormalizeSubscriptionSet(SubscriptionSet{
		Owner:    "orders",
		Provider: " KAFKA ",
		Topics:   []TopicSubscription{{Topic: "orders.created", QoS: AtLeastOnce, Handler: func(context.Context, Message) {}}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if set.Provider != ProviderKafka {
		t.Fatalf("provider = %q, want %q", set.Provider, ProviderKafka)
	}
}

func TestNormalizeSubscriptionSetRejectsInvalidDeclarations(t *testing.T) {
	handler := func(context.Context, Message) {}
	base := func() SubscriptionSet {
		return SubscriptionSet{
			Owner:    "orders",
			Provider: ProviderEMQX,
			Topics: []TopicSubscription{
				{Topic: "orders.created", QoS: AtLeastOnce, Handler: handler},
			},
		}
	}

	tests := []struct {
		name    string
		set     SubscriptionSet
		wantErr string
	}{
		{
			name:    "empty owner",
			set:     func() SubscriptionSet { set := base(); set.Owner = "   "; return set }(),
			wantErr: "owner is empty",
		},
		{
			name:    "empty provider",
			set:     func() SubscriptionSet { set := base(); set.Provider = "   "; return set }(),
			wantErr: "unsupported mq provider",
		},
		{
			name:    "unsupported provider",
			set:     func() SubscriptionSet { set := base(); set.Provider = "pulsar"; return set }(),
			wantErr: "unsupported mq provider",
		},
		{
			name:    "empty topics",
			set:     func() SubscriptionSet { set := base(); set.Topics = nil; return set }(),
			wantErr: "topics are empty",
		},
		{
			name: "empty topic",
			set: func() SubscriptionSet {
				set := base()
				set.Topics[0].Topic = "  "
				return set
			}(),
			wantErr: "topic at index 0 is empty",
		},
		{
			name: "duplicate topic",
			set: func() SubscriptionSet {
				set := base()
				set.Topics = append(set.Topics, TopicSubscription{
					Topic: " orders.created ", QoS: AtMostOnce, Handler: handler,
				})
				return set
			}(),
			wantErr: "duplicate mq subscription topic",
		},
		{
			name: "nil handler",
			set: func() SubscriptionSet {
				set := base()
				set.Topics[0].Handler = nil
				return set
			}(),
			wantErr: "handler is nil",
		},
		{
			name: "invalid qos",
			set: func() SubscriptionSet {
				set := base()
				set.Topics[0].QoS = ExactlyOnce + 1
				return set
			}(),
			wantErr: "invalid mq qos",
		},
		{
			name: "kafka qos 2",
			set: func() SubscriptionSet {
				set := base()
				set.Provider = ProviderKafka
				set.Topics[0].QoS = ExactlyOnce
				return set
			}(),
			wantErr: "supports qos 0 or 1",
		},
		{
			name: "rabbitmq qos 2",
			set: func() SubscriptionSet {
				set := base()
				set.Provider = ProviderRabbitMQ
				set.Topics[0].QoS = ExactlyOnce
				return set
			}(),
			wantErr: "supports qos 0 or 1",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			_, err := NormalizeSubscriptionSet(test.set)
			if err == nil {
				t.Fatal("NormalizeSubscriptionSet() error = nil, want error")
			}
			if !strings.Contains(err.Error(), test.wantErr) {
				t.Fatalf("error = %q, want substring %q", err, test.wantErr)
			}
		})
	}
}

type subscriptionTestRegistrar struct {
	sets []SubscriptionSet
}

func (r *subscriptionTestRegistrar) Register(set SubscriptionSet) error {
	r.sets = append(r.sets, set)
	return nil
}

func (*subscriptionTestRegistrar) Unregister(string) error { return nil }

type subscriptionTestContributor struct {
	name  string
	set   SubscriptionSet
	order *[]string
	err   error
}

func (c *subscriptionTestContributor) RegisterSubscriptions(registrar SubscriptionRegistrar) error {
	*c.order = append(*c.order, c.name)
	if c.err != nil {
		return c.err
	}
	return registrar.Register(c.set)
}

func TestApplySubscriptionsAppliesMultipleContributorsInOrder(t *testing.T) {
	order := make([]string, 0, 2)
	registrar := &subscriptionTestRegistrar{}
	handler := func(context.Context, Message) {}
	first := &subscriptionTestContributor{
		name:  "orders",
		order: &order,
		set: SubscriptionSet{
			Owner:    "orders",
			Provider: ProviderRabbitMQ,
			Topics:   []TopicSubscription{{Topic: "orders.created", Handler: handler}},
		},
	}
	second := &subscriptionTestContributor{
		name:  "notifications",
		order: &order,
		set: SubscriptionSet{
			Owner:    "notifications",
			Provider: ProviderEMQX,
			Topics:   []TopicSubscription{{Topic: "notifications.sent", Handler: handler}},
		},
	}

	if err := ApplySubscriptions(registrar, first, nil, second); err != nil {
		t.Fatalf("ApplySubscriptions() error = %v", err)
	}
	if !reflect.DeepEqual(order, []string{"orders", "notifications"}) {
		t.Fatalf("contributor order = %#v, want %#v", order, []string{"orders", "notifications"})
	}
	if len(registrar.sets) != 2 {
		t.Fatalf("registered sets = %d, want 2", len(registrar.sets))
	}
	if registrar.sets[0].Owner != "orders" || registrar.sets[1].Owner != "notifications" {
		t.Fatalf("registered owners = %q, %q", registrar.sets[0].Owner, registrar.sets[1].Owner)
	}
}

func TestApplySubscriptionsPropagatesContributorError(t *testing.T) {
	order := make([]string, 0, 2)
	registrar := &subscriptionTestRegistrar{}
	wantErr := errors.New("registration failed")
	failing := &subscriptionTestContributor{name: "failing", order: &order, err: wantErr}
	following := &subscriptionTestContributor{name: "following", order: &order}

	err := ApplySubscriptions(registrar, failing, following)
	if !errors.Is(err, wantErr) {
		t.Fatalf("ApplySubscriptions() error = %v, want %v", err, wantErr)
	}
	if !reflect.DeepEqual(order, []string{"failing"}) {
		t.Fatalf("contributors called = %#v, want %#v", order, []string{"failing"})
	}
	if len(registrar.sets) != 0 {
		t.Fatalf("registered sets = %d, want 0", len(registrar.sets))
	}
}

func TestApplySubscriptionsRejectsNilRegistrar(t *testing.T) {
	if err := ApplySubscriptions(nil); err == nil || !strings.Contains(err.Error(), "registrar is nil") {
		t.Fatalf("ApplySubscriptions(nil) error = %v, want nil-registrar error", err)
	}
}