From 16c4dc129ec0745159c7aeb92270e9019df71d3e Mon Sep 17 00:00:00 2001
From: Zubair Haque <haque.zubair@gmail.com>
Date: Sat, 27 Jan 2024 00:51:54 -0500
Subject: [PATCH] adding base/kafka_test.go for code coverage (#2990)

Signed-off-by: zhaque44 <haque.zubair@gmail.com>
---
 eventbus/kafka/base/kafka_test.go | 105 ++++++++++++++++++++++++++++++
 1 file changed, 105 insertions(+)
 create mode 100644 eventbus/kafka/base/kafka_test.go

diff --git a/eventbus/kafka/base/kafka_test.go b/eventbus/kafka/base/kafka_test.go
new file mode 100644
index 0000000000..06e70d751c
--- /dev/null
+++ b/eventbus/kafka/base/kafka_test.go
@@ -0,0 +1,105 @@
+package base
+
+import (
+	"testing"
+
+	"github.com/IBM/sarama"
+	eventbusv1alpha1 "github.com/argoproj/argo-events/pkg/apis/eventbus/v1alpha1"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+)
+
+func TestBrokers(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL: "broker1:9092,broker2:9092",
+	}
+
+	logger := zap.NewNop().Sugar()
+	kafka := NewKafka(config, logger)
+
+	expectedBrokers := []string{"broker1:9092", "broker2:9092"}
+	actualBrokers := kafka.Brokers()
+
+	assert.Equal(t, expectedBrokers, actualBrokers)
+}
+
+func TestConfig(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL: "localhost:9092",
+	}
+
+	logger := zap.NewNop().Sugar()
+
+	kafka := NewKafka(config, logger)
+
+	saramaConfig, err := kafka.Config()
+
+	assert.NoError(t, err)
+	assert.NotNil(t, saramaConfig)
+	assert.Equal(t, sarama.OffsetNewest, saramaConfig.Consumer.Offsets.Initial)
+	assert.Equal(t, sarama.WaitForAll, saramaConfig.Producer.RequiredAcks)
+}
+
+func TestConfig_StartOldest(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL: "localhost:9092",
+		ConsumerGroup: &eventbusv1alpha1.KafkaConsumerGroup{
+			StartOldest: true,
+		},
+	}
+
+	logger := zap.NewNop().Sugar()
+
+	kafka := NewKafka(config, logger)
+
+	saramaConfig, err := kafka.Config()
+
+	assert.NoError(t, err)
+	assert.NotNil(t, saramaConfig)
+	assert.Equal(t, sarama.OffsetOldest, saramaConfig.Consumer.Offsets.Initial)
+}
+
+func TestConfig_NoSASL(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL:  "localhost:9092",
+		SASL: nil,
+	}
+
+	logger := zap.NewNop().Sugar()
+
+	kafka := NewKafka(config, logger)
+
+	saramaConfig, err := kafka.Config()
+
+	assert.NoError(t, err)
+	assert.NotNil(t, saramaConfig)
+	assert.False(t, saramaConfig.Net.SASL.Enable)
+}
+
+func TestNewKafka(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL: "localhost:9092",
+	}
+
+	logger := zap.NewNop().Sugar()
+
+	kafka := NewKafka(config, logger)
+
+	assert.NotNil(t, kafka)
+	assert.NotNil(t, kafka.Logger)
+	assert.NotNil(t, kafka.config)
+}
+
+func TestNewKafka_EmptyURL(t *testing.T) {
+	config := &eventbusv1alpha1.KafkaBus{
+		URL: "",
+	}
+
+	logger := zap.NewNop().Sugar()
+
+	kafka := NewKafka(config, logger)
+
+	assert.NotNil(t, kafka)
+	assert.NotNil(t, kafka.Logger)
+	assert.NotNil(t, kafka.config)
+}