Use mocks for Queue/ReQueue in unit test

Author: Christian Wygoda

Gopkg.lock

@@ -11,7 +11,8 @@ import (
 )
 
 // Ensure the projectionNotificationProcessor.Queue is a ProjectionTrigger
-var _ ProjectionTrigger = (&projectionNotificationProcessor{}).Queue
+var _ ProjectionTrigger = (&notificationQueue{}).Queue
+var _ ProjectionTrigger = (&notificationQueue{}).ReQueue
 
 type (
 	// projectionNotificationProcessor provides a way to Trigger a notification using a set of background processes.
@@ -24,14 +25,64 @@ type (
 		logger  goengine.Logger
 		metrics Metrics
 
+		notificationQueue notificationQueueInterface
+	}
+
+	notificationQueueInterface interface {
+		Start(done chan struct{}, queue chan *ProjectionNotification)
+		Queue(ctx context.Context, notification *ProjectionNotification) error
+		ReQueue(ctx context.Context, notification *ProjectionNotification) error
+	}
+
+	notificationQueue struct {
 		retryDelay time.Duration
+		metrics    Metrics
+		done       chan struct{}
+		queue      chan *ProjectionNotification
 	}
 
 	// ProcessHandler is a func used to trigger a notification but with the addition of providing a Trigger func so
 	// the original notification can trigger other notifications
 	ProcessHandler func(context.Context, *ProjectionNotification, ProjectionTrigger) error
 )
 
+func newNotificationQueue(retryDelay time.Duration, metrics Metrics) *notificationQueue {
+	if retryDelay == 0 {
+		retryDelay = time.Millisecond * 50
+	}
+
+	return &notificationQueue{
+		retryDelay: retryDelay,
+		metrics:    metrics,
+	}
+}
+
+func (nq *notificationQueue) Start(done chan struct{}, queue chan *ProjectionNotification) {
+	nq.done = done
+	nq.queue = queue
+}
+
+func (nq *notificationQueue) Queue(ctx context.Context, notification *ProjectionNotification) error {
+	select {
+	default:
+	case <-ctx.Done():
+		return context.Canceled
+	case <-nq.done:
+		return errors.New("goengine: unable to queue notification because the processor was stopped")
+	}
+
+	nq.metrics.QueueNotification(notification)
+
+	nq.queue <- notification
+	return nil
+}
+
+func (nq *notificationQueue) ReQueue(ctx context.Context, notification *ProjectionNotification) error {
+	notification.ValidAfter = time.Now().Add(nq.retryDelay)
+
+	return nq.Queue(ctx, notification)
+}
+
 // newBackgroundProcessor create a new projectionNotificationProcessor
 func newBackgroundProcessor(
 	queueProcessors,
@@ -53,16 +104,12 @@ func newBackgroundProcessor(
 		metrics = NopMetrics
 	}
 
-	if retryDelay == 0 {
-		retryDelay = time.Millisecond * 50
-	}
-
 	return &projectionNotificationProcessor{
-		queueProcessors: queueProcessors,
-		queueBuffer:     queueBuffer,
-		logger:          logger,
-		metrics:         metrics,
-		retryDelay:      retryDelay,
+		queueProcessors:   queueProcessors,
+		queueBuffer:       queueBuffer,
+		logger:            logger,
+		metrics:           metrics,
+		notificationQueue: newNotificationQueue(retryDelay, metrics),
 	}, nil
 }
 
@@ -76,7 +123,7 @@ func (b *projectionNotificationProcessor) Execute(ctx context.Context, handler P
 	defer stopExecutor()
 
 	// Execute a run of the internal.
-	if err := b.Queue(ctx, nil); err != nil {
+	if err := b.notificationQueue.Queue(ctx, nil); err != nil {
 		return err
 	}
 
@@ -94,6 +141,8 @@ func (b *projectionNotificationProcessor) Start(ctx context.Context, handler Pro
 	b.done = make(chan struct{})
 	b.queue = make(chan *ProjectionNotification, b.queueBuffer)
 
+	b.notificationQueue.Start(b.done, b.queue)
+
 	var wg sync.WaitGroup
 	wg.Add(b.queueProcessors)
 	for i := 0; i < b.queueProcessors; i++ {
@@ -115,25 +164,12 @@ func (b *projectionNotificationProcessor) Start(ctx context.Context, handler Pro
 
 // Queue puts the notification on the queue to be processed
 func (b *projectionNotificationProcessor) Queue(ctx context.Context, notification *ProjectionNotification) error {
-	select {
-	default:
-	case <-ctx.Done():
-		return context.Canceled
-	case <-b.done:
-		return errors.New("goengine: unable to queue notification because the processor was stopped")
-	}
-
-	b.metrics.QueueNotification(notification)
-
-	b.queue <- notification
-	return nil
+	return b.notificationQueue.Queue(ctx, notification)
 }
 
 // ReQueue puts the notification again on the queue to be processed with a ValidAfter set
 func (b *projectionNotificationProcessor) ReQueue(ctx context.Context, notification *ProjectionNotification) error {
-	notification.ValidAfter = time.Now().Add(b.retryDelay)
-
-	return b.Queue(ctx, notification)
+	return b.notificationQueue.ReQueue(ctx, notification)
 }
 
 func (b *projectionNotificationProcessor) startProcessor(ctx context.Context, handler ProcessHandler) {
@@ -147,9 +183,9 @@ ProcessorLoop:
 		case notification := <-b.queue:
 			var queueFunc ProjectionTrigger
 			if notification == nil {
-				queueFunc = b.Queue
+				queueFunc = b.notificationQueue.Queue
 			} else {
-				queueFunc = b.ReQueue
+				queueFunc = b.notificationQueue.ReQueue
 
 				if notification.ValidAfter.After(time.Now()) {
 					b.queue <- notification

driver/sql/projection_notification_processor.go

@@ -2,35 +2,52 @@ package sql
 
 import (
 	"context"
-	"reflect"
-	"runtime"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
+type notificationQueueMock struct {
+	mock.Mock
+	done  chan struct{}
+	queue chan *ProjectionNotification
+}
+
+func (m *notificationQueueMock) Start(done chan struct{}, queue chan *ProjectionNotification) {
+	m.Called(done, queue)
+
+	m.done = done
+	m.queue = queue
+}
+
+func (m *notificationQueueMock) Queue(ctx context.Context, notification *ProjectionNotification) error {
+	args := m.Called(ctx, notification)
+
+	m.queue <- notification
+
+	return args.Error(0)
+}
+
+func (m *notificationQueueMock) ReQueue(ctx context.Context, notification *ProjectionNotification) error {
+	args := m.Called(ctx, notification)
+
+	return args.Error(0)
+}
+
 func TestStartProcessor(t *testing.T) {
 	testCases := []struct {
 		title        string
 		notification func() *ProjectionNotification
-		mockHandler  func(*testing.T, *projectionNotificationProcessor, context.CancelFunc) ProcessHandler
+		queueMethod  string
 	}{
 		{
 			"Handle nil notification",
 			func() *ProjectionNotification {
 				return nil
 			},
-			func(t *testing.T, processor *projectionNotificationProcessor, cancel context.CancelFunc) ProcessHandler {
-				return func(ctx context.Context, notification *ProjectionNotification, queue ProjectionTrigger) error {
-					defer cancel()
-
-					expectedFuncName := runtime.FuncForPC(reflect.ValueOf(processor.Queue).Pointer()).Name()
-					actualFuncName := runtime.FuncForPC(reflect.ValueOf(queue).Pointer()).Name()
-					require.Equal(t, expectedFuncName, actualFuncName)
-					return nil
-				}
-			},
+			"Queue",
 		},
 		{
 			"Handle new notification",
@@ -40,53 +57,55 @@ func TestStartProcessor(t *testing.T) {
 					AggregateID: "abc",
 				}
 			},
-			func(t *testing.T, processor *projectionNotificationProcessor, cancel context.CancelFunc) ProcessHandler {
-				return func(ctx context.Context, notification *ProjectionNotification, queue ProjectionTrigger) error {
-					defer cancel()
-
-					expectedFuncName := runtime.FuncForPC(reflect.ValueOf(processor.ReQueue).Pointer()).Name()
-					actualFuncName := runtime.FuncForPC(reflect.ValueOf(queue).Pointer()).Name()
-					require.Equal(t, expectedFuncName, actualFuncName)
-					return nil
-				}
-			},
+			"ReQueue",
 		},
 		{
 			"Handle retried notification",
 			func() *ProjectionNotification {
-				now := time.Now().Add(time.Millisecond * 200)
 				return &ProjectionNotification{
 					No:          1,
 					AggregateID: "abc",
-					ValidAfter:  now,
-				}
-			},
-			func(t *testing.T, processor *projectionNotificationProcessor, cancel context.CancelFunc) ProcessHandler {
-				return func(ctx context.Context, notification *ProjectionNotification, queue ProjectionTrigger) error {
-					defer cancel()
-
-					expectedFuncName := runtime.FuncForPC(reflect.ValueOf(processor.ReQueue).Pointer()).Name()
-					actualFuncName := runtime.FuncForPC(reflect.ValueOf(queue).Pointer()).Name()
-					require.Equal(t, expectedFuncName, actualFuncName)
-
-					require.True(t, notification.ValidAfter.Before(time.Now()))
-					return nil
+					ValidAfter:  time.Now().Add(time.Millisecond * 200),
 				}
 			},
+			"ReQueue",
 		},
 	}
 
 	for _, testCase := range testCases {
 		t.Run(testCase.title, func(t *testing.T) {
-			processor, err := newBackgroundProcessor(1, 1, nil, nil, 0)
-			require.Nil(t, err)
+			bufferSize := 1
+			queueProcessorsCount := 1
+			retryDelay := time.Millisecond * 0
 
+			queue := make(chan *ProjectionNotification, bufferSize)
+			done := make(chan struct{})
 			ctx, cancel := context.WithCancel(context.Background())
-			processor.queue = make(chan *ProjectionNotification, processor.queueBuffer)
+			notification := testCase.notification()
 
-			err = processor.Queue(ctx, testCase.notification())
-			require.Nil(t, err)
-			processor.startProcessor(ctx, testCase.mockHandler(t, processor, cancel))
+			processor, err := newBackgroundProcessor(queueProcessorsCount, bufferSize, nil, nil, retryDelay)
+			require.NoError(t, err)
+			processor.done = done
+			processor.queue = queue
+
+			nqMock := &notificationQueueMock{}
+			nqMock.On("Start", mock.Anything, queue).Return()
+			nqMock.On(testCase.queueMethod, ctx, notification).Return(nil)
+			nqMock.Start(processor.done, processor.queue)
+			processor.notificationQueue = nqMock
+
+			queue <- notification
+
+			handler := func(ctx context.Context, notification *ProjectionNotification, queue ProjectionTrigger) error {
+				defer cancel()
+
+				err := queue(ctx, notification)
+				require.NoError(t, err)
+
+				nqMock.AssertExpectations(t)
+				return nil
+			}
+			processor.startProcessor(ctx, handler)
 		})
 	}
 }