Test: domain produce operator

Author: asheswook

internal/queue/kafka/operator_test.go

@@ -8,20 +8,38 @@ import (
 
 var brokers = []string{"kafka.vp-datacenter-1.violetpay.net:9092", "kafka.vp-datacenter-1.violetpay.net:9093", "kafka.vp-datacenter-1.violetpay.net:9094"}
 
-func mapOperatorProvider(queueName string, serializer queue.MessageSerializer[*sarama.ConsumerMessage, map[string]string], callback queue.Callback[map[string]string]) queue.ConsumeOperator[*sarama.ConsumerMessage, map[string]string] {
+func mapConsumeOperatorProvider(queueName string, serializer queue.MessageSerializer[*sarama.ConsumerMessage, map[string]string], callback queue.Callback[map[string]string]) queue.ConsumeOperator[*sarama.ConsumerMessage, map[string]string] {
 	return NewConsumeOperator(serializer, callback, brokers, queueName, "test-group-id", sarama.NewConfig())
 }
 
-func rawMsgProvider() *sarama.ConsumerMessage {
+func consumeMessageProvider() *sarama.ConsumerMessage {
 	return &sarama.ConsumerMessage{}
 }
 
 func TestConsumeOperator(t *testing.T) {
 	t.Run("map[string]string", func(t *testing.T) {
 		queue.TestSuiteConsumeOperator[*sarama.ConsumerMessage, map[string]string](
 			t,
-			mapOperatorProvider,
-			rawMsgProvider,
+			mapConsumeOperatorProvider,
+			consumeMessageProvider,
 		)
 	})
 }
+
+func bytesProduceOperatorProvider(queueName string) queue.ProduceOperator[[]byte] {
+	return NewBytesProduceOperator(brokers, queueName, func() *sarama.Config {
+		return sarama.NewConfig()
+	})
+}
+
+func produceBytesMessageProvider() []byte {
+	return []byte("test")
+}
+
+func TestBytesProduceOperator(t *testing.T) {
+	queue.TestSuiteProduceOperator(
+		t,
+		bytesProduceOperatorProvider,
+		produceBytesMessageProvider,
+	)
+}

queue/operator.go

@@ -4,7 +4,7 @@ type ConsumeOperator[InMsg any, Msg any] interface {
 	QueueName() string
 	Serializer() MessageSerializer[InMsg, Msg]
 	Callback() Callback[Msg]
-	Consume(msg InMsg)
+	Consume(message InMsg)
 	StartConsume() error
 	StopConsume() error
 }

queue/operator_tests.go

@@ -28,7 +28,7 @@ func newMockSerializer[Input any, Output any]() *mockSerializer[Input, Output] {
 func TestSuiteConsumeOperator[InMsg any, Msg any](
 	t *testing.T,
 	operatorProvider func(queueName string, serializer MessageSerializer[InMsg, Msg], callback Callback[Msg]) ConsumeOperator[InMsg, Msg],
-	rawMessageProvider func() InMsg,
+	testMessageProvider func() InMsg,
 ) {
 	t.Run("ConsumeOperator Test Suite", func(t *testing.T) {
 		var queueName string
@@ -72,7 +72,7 @@ func TestSuiteConsumeOperator[InMsg any, Msg any](
 			})
 
 			assert.NotNil(t, operator.Serializer(), serializer)
-			_, err := serializer.Serialize(rawMessageProvider())
+			_, err := serializer.Serialize(testMessageProvider())
 			if err != nil {
 				return
 			}
@@ -107,7 +107,7 @@ func TestSuiteConsumeOperator[InMsg any, Msg any](
 				operator = operatorProvider(queueName, serializer, callback)
 			})
 
-			msg := rawMessageProvider()
+			msg := testMessageProvider()
 			assert.NotNil(t, msg)
 
 			expectedSerializedValue, err := serializer.Serialize(msg)
@@ -132,10 +132,8 @@ func TestSuiteConsumeOperator[InMsg any, Msg any](
 				operator = operatorProvider(queueName, serializer, callback)
 			})
 
-			assert.NotPanics(t, func() {
-				operator.StartConsume()
-			})
-
+			err := operator.StartConsume()
+			assert.Nil(t, err)
 		})
 
 		t.Run("StopConsume", func(t *testing.T) {
@@ -150,12 +148,45 @@ func TestSuiteConsumeOperator[InMsg any, Msg any](
 				operator = operatorProvider(queueName, serializer, callback)
 			})
 
-			assert.NotPanics(t, func() {
-				operator.StartConsume()
+			err := operator.StartConsume()
+			assert.Nil(t, err)
+
+			err = operator.StopConsume()
+			assert.Nil(t, err)
+		})
+	})
+}
+
+func TestSuiteProduceOperator[Msg any](
+	t *testing.T,
+	operatorProvider func(queueName string) ProduceOperator[Msg],
+	testMessageProvider func() Msg,
+) {
+	t.Run("ProduceOperator Test Suite", func(t *testing.T) {
+		queueName := "testQueueName"
+		var operator ProduceOperator[Msg]
+
+		t.Run("QueueName", func(t *testing.T) {
+			t.Cleanup(func() {
+				queueName = "testQueueName"
+				operator = nil
 			})
 
-			assert.NotPanics(t, func() {
-				operator.StopConsume()
+			operator = operatorProvider(queueName)
+			assert.Equal(t, operator.QueueName(), queueName)
+		})
+
+		t.Run("Produce", func(t *testing.T) {
+			t.Run("Produce with valid message", func(t *testing.T) {
+				t.Cleanup(func() {
+					queueName = "testQueueName"
+					operator = nil
+				})
+
+				operator = operatorProvider(queueName)
+				msg := testMessageProvider()
+				err := operator.Produce(msg)
+				assert.Nil(t, err)
 			})
 		})
 	})