chore: add unit test

Author: siddhanta-rath

app/server_test.go

@@ -69,10 +69,11 @@ func isClosed(ch <-chan collection.CollectRequest) bool {
 // mockKafkaClient is a mock for the Client interface
 type mockKafkaClient struct {
 	// Tracking flags
-	ProduceCalled bool
-	CloseCalled   bool
-	FlushCalled   bool
-	EventsCalled  bool
+	ProduceCalled     bool
+	CloseCalled       bool
+	FlushCalled       bool
+	EventsCalled      bool
+	GetMetadataCalled bool
 
 	ReturnFlushLeft int
 	EventChan       chan kafka.Event
@@ -99,3 +100,8 @@ func (m *mockKafkaClient) Events() chan kafka.Event {
 	}
 	return m.EventChan
 }
+
+func (m *mockKafkaClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error) {
+	m.GetMetadataCalled = true
+	return &kafka.Metadata{}, nil
+}

config/load_test.go

@@ -52,6 +52,35 @@ func TestGRPCServerConfig(t *testing.T) {
 	assert.Equal(t, "8081", ServerGRPC.Port)
 }
 
+func TestServerMQTTConfig(t *testing.T) {
+	os.Setenv("SERVER_MQTT_CONSUL_ADDRESS", "consul:8081")
+	os.Setenv("SERVER_MQTT_CONSUL_KV_KEY", "kv/path")
+	os.Setenv("SERVER_MQTT_CONSUL_HEALTH_ONLY", "true")
+	os.Setenv("SERVER_MQTT_CONSUL_WAIT_TIME", "300")
+	os.Setenv("SERVER_MQTT_AUTH_USERNAME", "test")
+	os.Setenv("SERVER_MQTT_AUTH_PASSWORD", "pass")
+	os.Setenv("SERVER_MQTT_CONSUMER_RETRY_INTERVAL_IN_SEC", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_WRITE_TIMEOUT_IN_SEC", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_LOG_LEVEL", "warn")
+	os.Setenv("SERVER_MQTT_CONSUMER_POOL_SIZE", "1")
+	os.Setenv("SERVER_MQTT_CONSUMER_TOPIC_FORMAT", "default-topic")
+	os.Setenv("SERVER_MQTT_CONNECTION_GROUP", "consumer")
+	serverMQTTConfigLoader()
+	assert.Equal(t, "consul:8081", ServerMQTT.ConsulConfig.Address)
+	assert.Equal(t, "kv/path", ServerMQTT.ConsulConfig.KVKey)
+	assert.Equal(t, true, ServerMQTT.ConsulConfig.HealthOnly)
+	assert.Equal(t, 300*time.Second, ServerMQTT.ConsulConfig.WaitTime)
+	assert.Equal(t, "test", ServerMQTT.AuthConfig.Username)
+	assert.Equal(t, "pass", ServerMQTT.AuthConfig.Password)
+	assert.Equal(t, 1*time.Second, ServerMQTT.ConsumerConfig.RetryIntervalInSec)
+	assert.Equal(t, 1*time.Second, ServerMQTT.ConsumerConfig.WriteTimeoutInSec)
+	assert.Equal(t, "warn", ServerMQTT.ConsumerConfig.LogLevel)
+	assert.Equal(t, 1, ServerMQTT.ConsumerConfig.PoolSize)
+	assert.Equal(t, "default-topic", ServerMQTT.ConsumerConfig.TopicFormat)
+	assert.Equal(t, "consumer", ServerMQTT.ConnGroup)
+
+}
+
 func TestDynamicConfigLoad(t *testing.T) {
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_RANDOM", "anything")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_BOOTSTRAP_SERVERS", "localhost:9092")
@@ -67,6 +96,8 @@ func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_ACKS", "1")
 	os.Setenv("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "10000")
 	os.Setenv("SOMETHING_PUBLISHER_KAFKA_CLIENT_SOMETHING", "anything")
+	os.Setenv("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME", "test-log")
+	os.Setenv("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS", "5000")
 	publisherKafkaConfigLoader()
 	kafkaConfig := PublisherKafka.ToKafkaConfigMap()
 	bootstrapServer, _ := kafkaConfig.Get("bootstrap.servers", "")
@@ -76,6 +107,8 @@ func TestKafkaConfig_ToKafkaConfigMap(t *testing.T) {
 	assert.Equal(t, "", topic)
 	assert.NotEqual(t, something, "anything")
 	assert.Equal(t, 4, len(*kafkaConfig))
+	assert.Equal(t, "test-log", PublisherKafka.HealthCheckConfig.TopicName)
+	assert.Equal(t, 5000, PublisherKafka.HealthCheckConfig.TimeOut)
 }
 
 func TestWorkerConfig(t *testing.T) {

config/publisher.go

@@ -14,7 +14,13 @@ var PublisherKafka publisherKafka
 var dynamicKafkaClientConfigPrefix = "PUBLISHER_KAFKA_CLIENT_"
 
 type publisherKafka struct {
-	FlushInterval int
+	FlushInterval     int
+	HealthCheckConfig healthcheck
+}
+
+type healthcheck struct {
+	TopicName string
+	TimeOut   int
 }
 
 func (k publisherKafka) ToKafkaConfigMap() *confluent.ConfigMap {
@@ -43,9 +49,15 @@ func dynamicKafkaClientConfigLoad() []byte {
 func publisherKafkaConfigLoader() {
 	viper.SetDefault("PUBLISHER_KAFKA_CLIENT_QUEUE_BUFFERING_MAX_MESSAGES", "100000")
 	viper.SetDefault("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS", "1000")
+	viper.SetDefault("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME", "clickstream-test-log")
+	viper.SetDefault("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS", "5000")
 	viper.MergeConfig(bytes.NewBuffer(dynamicKafkaClientConfigLoad()))
 
 	PublisherKafka = publisherKafka{
 		FlushInterval: util.MustGetInt("PUBLISHER_KAFKA_FLUSH_INTERVAL_MS"),
+		HealthCheckConfig: healthcheck{
+			TopicName: util.MustGetString("PUBLISHER_KAFKA_HEALTHCHECK_TOPIC_NAME"),
+			TimeOut:   util.MustGetInt("PUBLISHER_KAFKA_HEALTHCHECK_TIMEOUT_MS"),
+		},
 	}
 }

health/registry_test.go

@@ -0,0 +1,86 @@
+package health
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+)
+
+func TestRegister(t *testing.T) {
+	resetCheckers()
+
+	mockChecker := func() error { return nil }
+
+	Register("db", mockChecker)
+
+	mu.RLock()
+	defer mu.RUnlock()
+
+	if len(checkers) != 1 {
+		t.Fatalf("expected 1 checker, got %d", len(checkers))
+	}
+
+	if _, ok := checkers["db"]; !ok {
+		t.Errorf("expected checker 'db' to be registered")
+	}
+}
+
+func TestCheckAll(t *testing.T) {
+	tests := []struct {
+		name     string
+		register map[string]Checker
+		expected map[string]string
+	}{
+		{
+			name: "single healthy service on register map",
+			register: map[string]Checker{
+				"serviceA": func() error { return nil },
+			},
+			expected: map[string]string{
+				"serviceA": "healthy",
+			},
+		},
+		{
+			name: "single unhealthy service on register map",
+			register: map[string]Checker{
+				"serviceB": func() error { return errors.New("timeout") },
+			},
+			expected: map[string]string{
+				"serviceB": "unhealthy: timeout",
+			},
+		},
+		{
+			name: "multiple services on register map",
+			register: map[string]Checker{
+				"ok":   func() error { return nil },
+				"fail": func() error { return errors.New("db down") },
+			},
+			expected: map[string]string{
+				"ok":   "healthy",
+				"fail": "unhealthy: db down",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			//rest the register map for every test scenario
+			resetCheckers()
+
+			for name, fn := range tt.register {
+				Register(name, fn)
+			}
+
+			actual := CheckAll()
+			if !reflect.DeepEqual(actual, tt.expected) {
+				t.Errorf("CheckAll() = %v, want %v", actual, tt.expected)
+			}
+		})
+	}
+}
+
+func resetCheckers() {
+	mu.Lock()
+	defer mu.Unlock()
+	checkers = make(map[string]Checker)
+}

publisher/kafka.go

@@ -220,7 +220,7 @@ func (b BulkError) Error() string {
 }
 
 func (pr *Kafka) HealthCheck() error {
-	topic := "clickstream-page-log"
-	_, err := pr.kp.GetMetadata(&topic, false, 5000)
+	topic := config.PublisherKafka.HealthCheckConfig.TopicName
+	_, err := pr.kp.GetMetadata(&topic, false, config.PublisherKafka.HealthCheckConfig.TimeOut)
 	return err
 }

publisher/mockclient.go

@@ -26,3 +26,8 @@ func (p *mockClient) Flush(config int) int {
 func (p *mockClient) Events() chan kafka.Event {
 	return make(chan kafka.Event)
 }
+
+func (p *mockClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error) {
+	args := p.Called(topic, allTopics, timeoutMs)
+	return &kafka.Metadata{}, args.Error(0)
+}

services/mqtt/client/cred_test.go

@@ -0,0 +1,73 @@
+package client
+
+import (
+	"context"
+	"testing"
+
+	"github.com/gojek/courier-go"
+	"github.com/goto/raccoon/config"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewCredentialFetcher(t *testing.T) {
+	tests := []struct {
+		name        string
+		username    string
+		password    string
+		expectError bool
+	}{
+		{
+			name:        "success - valid username and password",
+			username:    "testUser",
+			password:    "testPass",
+			expectError: false,
+		},
+		{
+			name:        "error - missing username",
+			username:    "",
+			password:    "somePass",
+			expectError: true,
+		},
+		{
+			name:        "error - missing password",
+			username:    "someUser",
+			password:    "",
+			expectError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// mock config
+			config.ServerMQTT.AuthConfig.Username = tt.username
+			config.ServerMQTT.AuthConfig.Password = tt.password
+
+			fetcher, err := newCredentialFetcher()
+
+			if tt.expectError {
+				assert.Nil(t, fetcher)
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, fetcher)
+				assert.Equal(t, tt.username, fetcher.username)
+				assert.Equal(t, tt.password, fetcher.password)
+			}
+		})
+	}
+}
+
+func TestCredentials(t *testing.T) {
+	fetcher := &credentialFetcher{
+		username: "myUser",
+		password: "myPass",
+	}
+
+	cred, err := fetcher.Credentials(context.Background())
+	assert.NoError(t, err)
+	assert.NotNil(t, cred)
+	assert.Equal(t, &courier.Credential{
+		Username: "myUser",
+		Password: "myPass",
+	}, cred)
+}

services/mqtt/client/logger_test.go

@@ -0,0 +1,79 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/goto/raccoon/logger"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+// testHook captures log entries emitted by logrus.
+type testHook struct {
+	entries []logrus.Entry
+}
+
+func (h *testHook) Levels() []logrus.Level {
+	return logrus.AllLevels
+}
+
+func (h *testHook) Fire(e *logrus.Entry) error {
+	h.entries = append(h.entries, *e)
+	return nil
+}
+
+func TestLogger_InfoAndError(t *testing.T) {
+	// Arrange
+	baseLogger := logrus.New()
+	hook := &testHook{}
+	baseLogger.AddHook(hook)
+
+	l := Logger{log: baseLogger}
+	ctx := context.Background()
+
+	tests := []struct {
+		name        string
+		callFunc    func()
+		expected    logrus.Level
+		expectedMsg string
+	}{
+		{
+			name: "Info logs message with attrs",
+			callFunc: func() {
+				l.Info(ctx, "info message", map[string]any{"key": "value"})
+			},
+			expected:    logrus.InfoLevel,
+			expectedMsg: "info message",
+		},
+		{
+			name: "Error logs error with attrs",
+			callFunc: func() {
+				l.Error(ctx, errors.New("some error"), map[string]any{"code": 500})
+			},
+			expected:    logrus.ErrorLevel,
+			expectedMsg: "some error",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			hook.entries = nil // reset before each run
+			tt.callFunc()
+
+			assert.NotEmpty(t, hook.entries)
+			lastEntry := hook.entries[len(hook.entries)-1]
+
+			assert.Equal(t, tt.expected, lastEntry.Level)
+			assert.Contains(t, lastEntry.Message, tt.expectedMsg)
+		})
+	}
+}
+
+func TestNewLogger(t *testing.T) {
+	l := NewLogger()
+	assert.NotNil(t, l)
+	assert.NotNil(t, l.log)
+	assert.Equal(t, logger.GetLogger(), l.log)
+}

services/rest/handler.go

@@ -98,7 +98,7 @@ func (h *Handler) RESTAPIHandler(rw http.ResponseWriter, r *http.Request) {
 
 	b, err := ioutil.ReadAll(r.Body)
 	if err != nil {
-		logger.Errorf(fmt.Sprintf("[rest.GetRESTAPIHandler] %s error reading request body, error: %v", identifier, err))
+		logger.Errorf("[rest.GetRESTAPIHandler] %s error reading request body, error: %v", identifier, err)
 		metrics.Increment("batches_read_total", fmt.Sprintf("status=failed,reason=readerr,conn_group=%s", identifier.Group))
 		rw.WriteHeader(http.StatusInternalServerError)
 		_, err := res.SetCode(pb.Code_CODE_INTERNAL_ERROR).SetStatus(pb.Status_STATUS_ERROR).SetReason("deserialization failure").