chore: add more unit test case

siddhanta-rath · siddhanta-rath · commit a4911aa58d59 · 2025-11-10T02:17:09.000+05:30
diff --git a/services/mqtt/client/pubsub.go b/services/mqtt/client/pubsub.go
@@ -13,8 +13,18 @@ import (
 
 // MqttPubSubClient wraps a courier MQTT client with start/stop lifecycle management.
 type MqttPubSubClient struct {
-	client         *courier.Client
-	consulResolver *consul.Resolver
+	client   client
+	resolver resolver
+}
+
+type client interface {
+	Start() error
+	Stop()
+	IsConnected() bool
+}
+
+type resolver interface {
+	Start()
 }
 
 // NewMqttPubSubClient initializes a new MQTT client with Consul-based service discovery and credentials.
@@ -28,11 +38,13 @@ func NewMqttPubSubClient(ctx context.Context, handler courier.MessageHandler, cl
 		WaitTime:      consulCfg.WaitTime,
 	})
 	if err != nil {
+		logger.Infof("error while creating the consul new resolver %v", err)
 		return nil, fmt.Errorf("failed to create consul resolver: %w", err)
 	}
 
 	credFetcher, err := newCredentialFetcher()
 	if err != nil {
+		logger.Infof("error while creating credential %v", err)
 		return nil, fmt.Errorf("failed to create credential fetcher: %w", err)
 	}
 
@@ -51,13 +63,13 @@ func NewMqttPubSubClient(ctx context.Context, handler courier.MessageHandler, cl
 		courier.WithCustomDecoder(protoDecoder),
 	}
 
-	client, err := courier.NewClient(clientOpts...)
+	c, err := courier.NewClient(clientOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize MQTT client: %w", err)
 	}
 
 	logger.Infof("MQTT client initialized successfully for clientID=%s", clientID)
-	return &MqttPubSubClient{client: client, consulResolver: rs}, nil
+	return &MqttPubSubClient{client: c, resolver: rs}, nil
 }
 
 // registerHandler registers the subscription handler when the client connects.
@@ -74,7 +86,7 @@ func registerHandler(ctx context.Context, handler courier.MessageHandler) func(c
 
 // Start begins the MQTT client operation.
 func (m *MqttPubSubClient) Start() error {
-	go m.consulResolver.Start()
+	go m.resolver.Start()
 	if err := m.client.Start(); err != nil {
 		logger.Infof("MQTT client start failed due to %v", err)
 		return fmt.Errorf("failed to start MQTT client: %w", err)
diff --git a/services/mqtt/client/pubsub_test.go b/services/mqtt/client/pubsub_test.go
@@ -0,0 +1,151 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"github.com/gojek/courier-go"
+	"github.com/goto/raccoon/config"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+// ---- Mock Definitions ----
+
+type mockCourierClient struct {
+	mock.Mock
+}
+
+func (m *mockCourierClient) Start() error {
+	args := m.Called()
+	return args.Error(0)
+}
+func (m *mockCourierClient) Stop() {
+	m.Called()
+}
+func (m *mockCourierClient) IsConnected() bool {
+	args := m.Called()
+	return args.Bool(0)
+}
+
+type mockResolver struct {
+	mock.Mock
+	wg *sync.WaitGroup
+}
+
+func (m *mockResolver) Start() {
+	defer m.wg.Done()
+	m.Called()
+}
+
+// ---- Tests ----
+
+func TestMqttPubSubClient_Start(t *testing.T) {
+
+	t.Run("Start success", func(t *testing.T) {
+
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		mockResolver := &mockResolver{wg: wg}
+		mockResolver.On("Start").Return().Once()
+
+		mockClient := new(mockCourierClient)
+		mockClient.On("Start").Return(nil).Once()
+
+		m := &MqttPubSubClient{client: mockClient, resolver: mockResolver}
+		err := m.Start()
+		assert.NoError(t, err)
+
+		wg.Wait() // wait for goroutine completion
+		mockResolver.AssertExpectations(t)
+		mockClient.AssertExpectations(t)
+
+	})
+
+	t.Run("Start failure", func(t *testing.T) {
+
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		mockResolver := &mockResolver{wg: wg}
+		mockResolver.On("Start").Return().Once()
+
+		mockClient := new(mockCourierClient)
+		mockClient.On("Start").Return(errors.New("connection failed")).Once()
+
+		m := &MqttPubSubClient{client: mockClient, resolver: mockResolver}
+		err := m.Start()
+		assert.ErrorContains(t, err, "failed to start MQTT client")
+
+		wg.Wait() // wait for goroutine completion
+		mockResolver.AssertExpectations(t)
+		mockClient.AssertExpectations(t)
+	})
+
+}
+
+func TestMqttPubSubClient_Stop(t *testing.T) {
+
+	t.Run("Stop always succeeds", func(t *testing.T) {
+		mockClient := new(mockCourierClient)
+		mockResolver := new(mockResolver)
+
+		mockClient.On("Stop").Return().Once()
+
+		m := &MqttPubSubClient{
+			client:   mockClient,
+			resolver: mockResolver,
+		}
+
+		err := m.Stop()
+		assert.NoError(t, err)
+		mockClient.AssertExpectations(t)
+	})
+}
+
+func TestMqttPubSubClient_IsConnected(t *testing.T) {
+
+	t.Run("IsConnected reflects client state", func(t *testing.T) {
+		mockClient := new(mockCourierClient)
+		mockResolver := new(mockResolver)
+
+		mockClient.On("IsConnected").Return(true).Once()
+
+		m := &MqttPubSubClient{
+			client:   mockClient,
+			resolver: mockResolver,
+		}
+
+		assert.True(t, m.IsConnected())
+		mockClient.AssertExpectations(t)
+	})
+}
+
+func TestNewMqttPubSubClient(t *testing.T) {
+
+	t.Run("new pubsub client should be created", func(t *testing.T) {
+		mockPubSub := new(MockPubSub)
+		config.ServerMQTT.ConsulConfig.KVKey = "/test/path"
+		config.ServerMQTT.ConsulConfig.Address = "localhost:8085"
+		config.ServerMQTT.ConsulConfig.HealthOnly = true
+		config.ServerMQTT.ConsulConfig.WaitTime = 1 * time.Second
+		config.ServerMQTT.AuthConfig.Username = "test"
+		config.ServerMQTT.AuthConfig.Password = "pass"
+		c, err := NewMqttPubSubClient(context.Background(), mockPubSub.Subscribe, "test-client-1")
+		assert.Nil(t, err)
+		assert.NotNil(t, c.client)
+		assert.NotNil(t, c.resolver)
+	})
+
+}
+
+type MockPubSub struct {
+	mock.Mock
+}
+
+func (m *MockPubSub) Subscribe(ctx context.Context, c courier.PubSub, message *courier.Message) {
+	m.Called(ctx, c, message)
+	return
+}
diff --git a/services/mqtt/consumer_test.go b/services/mqtt/consumer_test.go
@@ -0,0 +1,128 @@
+package mqtt
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+)
+
+func TestConsumer_Init(t *testing.T) {
+	tests := []struct {
+		name      string
+		startErr  error
+		expectErr bool
+	}{
+		{
+			name:      "successfully starts client",
+			startErr:  nil,
+			expectErr: false,
+		},
+		{
+			name:      "fails to start client",
+			startErr:  errors.New("connection failed"),
+			expectErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockClient := new(mockPubSubClient)
+			mockClient.On("Start").Return(tt.startErr).Once()
+
+			c := NewConsumer(mockClient)
+			err := c.Init()
+
+			if tt.expectErr {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "failed to start consumer")
+			} else {
+				assert.NoError(t, err)
+			}
+
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConsumer_Shutdown(t *testing.T) {
+	tests := []struct {
+		name      string
+		stopErr   error
+		expectErr bool
+	}{
+		{
+			name:      "successfully stops client",
+			stopErr:   nil,
+			expectErr: false,
+		},
+		{
+			name:      "fails to stop client",
+			stopErr:   errors.New("disconnect failed"),
+			expectErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockClient := new(mockPubSubClient)
+			mockClient.On("Stop").Return(tt.stopErr).Once()
+
+			c := NewConsumer(mockClient)
+			err := c.Shutdown()
+
+			if tt.expectErr {
+				assert.Error(t, err)
+				assert.Contains(t, err.Error(), "failed to stop consumer")
+			} else {
+				assert.NoError(t, err)
+			}
+
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+func TestConsumer_IsHealthy(t *testing.T) {
+	tests := []struct {
+		name        string
+		isConnected bool
+	}{
+		{"client connected", true},
+		{"client disconnected", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockClient := new(mockPubSubClient)
+			mockClient.On("IsConnected").Return(tt.isConnected).Once()
+
+			c := NewConsumer(mockClient)
+			assert.Equal(t, tt.isConnected, c.IsHealthy())
+
+			mockClient.AssertExpectations(t)
+		})
+	}
+}
+
+// --- Mock Implementation ---
+
+type mockPubSubClient struct {
+	mock.Mock
+}
+
+func (m *mockPubSubClient) Start() error {
+	args := m.Called()
+	return args.Error(0)
+}
+
+func (m *mockPubSubClient) Stop() error {
+	args := m.Called()
+	return args.Error(0)
+}
+
+func (m *mockPubSubClient) IsConnected() bool {
+	args := m.Called()
+	return args.Bool(0)
+}
diff --git a/services/mqtt/handler.go b/services/mqtt/handler.go
@@ -39,14 +39,18 @@ func (h *Handler) MQTTHandler(ctx context.Context, c courier.PubSub, message *co
 		log.Errorf("mqtt message decoding failed due to : %v", err)
 		return
 	}
-	log.Infof("MQTT message content post deserialization %v", req)
+	//to be removed post end-to-end test
+	for _, event := range req.Events {
+		log.Infof("MQTT message content post deserialization %v", event)
+	}
+
 	//instrument the request number
 	metrics.Increment(
 		"batches_read_total",
 		fmt.Sprintf("status=success,conn_group=%s", identifier.Group),
 	)
 	//instrument the request size
-	reqBytes, err := serialization.SerializeProto(req)
+	reqBytes, err := serialization.SerializeProto(&req)
 	if err != nil {
 		log.Errorf("mqtt message serialization failed : %v", err)
 	} else {
diff --git a/services/mqtt/handler_test.go b/services/mqtt/handler_test.go
diff --git a/services/mqtt/service_test.go b/services/mqtt/service_test.go
