Tests for publisher.Publish

Author: bilaljaved

extension/amqp/amqp.go

@@ -4,8 +4,8 @@ import (
 	"github.com/streadway/amqp"
 )
 
-// setup returns a connection and channel to be used for consumption with the Queue setup
-func setup(url, queue string) (*amqp.Connection, *amqp.Channel, error) {
+// setup returns a connection and channel to be used for the Queue setup
+func setup(url, queue string) (NotificationConnection, NotificationChannel, error) {
 	conn, err := amqp.Dial(url)
 	if err != nil {
 		return nil, nil, err
@@ -22,3 +22,13 @@ func setup(url, queue string) (*amqp.Connection, *amqp.Channel, error) {
 
 	return conn, ch, nil
 }
+
+// NotificationChannel represents a channel for notifications
+type NotificationChannel interface {
+	Publish(exchange, queue string, mandatory, immediate bool, msg amqp.Publishing) error
+}
+
+// NotificationConnection represents a connection to the notification queue
+type NotificationConnection interface {
+	Close() error
+}

extension/amqp/amqp_test.go

@@ -0,0 +1,23 @@
+package amqp_test
+
+import "github.com/streadway/amqp"
+
+type mockConnection struct {
+}
+
+type mockChannel struct {
+}
+
+func (cn mockConnection) Close() error {
+	return nil
+}
+
+func (ch mockChannel) Publish(
+	exchange string,
+	queue string,
+	mandatory bool,
+	immediate bool,
+	msg amqp.Publishing,
+) error {
+	return nil
+}

extension/amqp/publisher.go

@@ -17,16 +17,29 @@ type NotificationPublisher struct {
 	queue   string
 	logger  goengine.Logger
 
-	connection *amqp.Connection
-	channel    *amqp.Channel
+	connection NotificationConnection
+	channel    NotificationChannel
 }
 
 // NewNotificationPublisher returns an instance of NotificationPublisher
-func NewNotificationPublisher(amqpDSN, queue string, logger goengine.Logger) (*NotificationPublisher, error) {
+func NewNotificationPublisher(amqpDSN, queue string,
+	logger goengine.Logger,
+	connection NotificationConnection,
+	channel NotificationChannel,
+) (*NotificationPublisher, error) {
+
+	if _, err := amqp.ParseURI(amqpDSN); err != nil {
+		return nil, goengine.InvalidArgumentError("amqpDSN")
+	}
+	if len(queue) == 0 {
+		return nil, goengine.InvalidArgumentError("queue")
+	}
 	return &NotificationPublisher{
-		amqpDSN: amqpDSN,
-		queue:   queue,
-		logger:  logger,
+		amqpDSN:    amqpDSN,
+		queue:      queue,
+		logger:     logger,
+		connection: connection,
+		channel:    channel,
 	}, nil
 }
 

extension/amqp/publisher_test.go

@@ -0,0 +1,68 @@
+package amqp_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/hellofresh/goengine"
+	"github.com/hellofresh/goengine/driver/sql"
+	goengineAmqp "github.com/hellofresh/goengine/extension/amqp"
+	goengineLogger "github.com/hellofresh/goengine/extension/logrus"
+	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNotificationPublisher_Publish(t *testing.T) {
+
+	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Second)
+	defer ctxCancel()
+
+	channel := &mockChannel{}
+	connection := &mockConnection{}
+
+	t.Run("Invalid arguments", func(t *testing.T) {
+		logger, _ := getLogger()
+
+		_, err := goengineAmqp.NewNotificationPublisher("http://localhost:5672/", "my-queue", logger, connection, channel)
+		assert.Equal(t, goengine.InvalidArgumentError("amqpDSN"), err)
+
+		_, err = goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "", logger, connection, channel)
+		assert.Equal(t, goengine.InvalidArgumentError("queue"), err)
+
+	})
+
+	t.Run("Publish Nil Notification Message", func(t *testing.T) {
+		ensure := require.New(t)
+		logger, loggerHook := getLogger()
+
+		publisher, err := goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "my-queue", logger, connection, channel)
+		ensure.NoError(err)
+		err = publisher.Publish(ctx, nil)
+		ensure.Nil(err)
+		ensure.Len(loggerHook.Entries, 1)
+		ensure.Equal("unable to handle nil notification, skipping", loggerHook.LastEntry().Message)
+	})
+
+	t.Run("Publish Message", func(t *testing.T) {
+		ensure := require.New(t)
+		logger, loggerHook := getLogger()
+
+		publisher, err := goengineAmqp.NewNotificationPublisher("amqp://localhost:5672/", "my-queue", logger, connection, channel)
+		ensure.NoError(err)
+
+		err = publisher.Publish(ctx, &sql.ProjectionNotification{No: 1, AggregateID: "8150276e-34fe-49d9-aeae-a35af0040a4f"})
+
+		ensure.NoError(err)
+		ensure.Len(loggerHook.Entries, 0)
+	})
+}
+
+func getLogger() (goengine.Logger, *test.Hook) {
+	logger, loggerHook := test.NewNullLogger()
+	logger.SetLevel(logrus.DebugLevel)
+
+	return goengineLogger.Wrap(logger), loggerHook
+}