Fix tests flakyness (#78)

Author: lucasmdrs

extension/amqp/listener.go

@@ -5,10 +5,11 @@ import (
 	"io"
 	"time"
 
-	"github.com/hellofresh/goengine/v2"
-	"github.com/hellofresh/goengine/v2/driver/sql"
 	"github.com/mailru/easyjson"
 	"github.com/streadway/amqp"
+
+	"github.com/hellofresh/goengine/v2"
+	"github.com/hellofresh/goengine/v2/driver/sql"
 )
 
 // Ensure Listener implements sql.Listener
@@ -24,6 +25,7 @@ type (
 		minReconnectInterval time.Duration
 		maxReconnectInterval time.Duration
 		logger               goengine.Logger
+		waitFn               func(time.Duration)
 	}
 )
 
@@ -48,9 +50,15 @@ func NewListener(
 		minReconnectInterval: minReconnectInterval,
 		maxReconnectInterval: maxReconnectInterval,
 		logger:               logger,
+		waitFn:               time.Sleep,
 	}, nil
 }
 
+// WithWaitFn replaces the default function called to wait (time.Sleep)
+func (l *Listener) WithWaitFn(fn func(time.Duration)) {
+	l.waitFn = fn
+}
+
 // Listen receives messages from a queue, transforms them into a sql.ProjectionNotification and calls the trigger
 func (l *Listener) Listen(ctx context.Context, trigger sql.ProjectionTrigger) error {
 	var nextReconnect time.Time
@@ -69,7 +77,7 @@ func (l *Listener) Listen(ctx context.Context, trigger sql.ProjectionTrigger) er
 				entry.String("reconnect_in", reconnectInterval.String())
 			})
 
-			time.Sleep(reconnectInterval)
+			l.waitFn(reconnectInterval)
 			reconnectInterval *= 2
 			if reconnectInterval > l.maxReconnectInterval {
 				reconnectInterval = l.maxReconnectInterval
@@ -85,7 +93,7 @@ func (l *Listener) Listen(ctx context.Context, trigger sql.ProjectionTrigger) er
 		case <-ctx.Done():
 			return context.Canceled
 		default:
-			time.Sleep(time.Until(nextReconnect))
+			l.waitFn(time.Until(nextReconnect))
 		}
 	}
 }

extension/amqp/listener_test.go

@@ -10,15 +10,16 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hellofresh/goengine/v2"
-	"github.com/hellofresh/goengine/v2/driver/sql"
-	"github.com/hellofresh/goengine/v2/extension/amqp"
-	goengineLogger "github.com/hellofresh/goengine/v2/extension/logrus"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
 	libamqp "github.com/streadway/amqp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/hellofresh/goengine/v2"
+	"github.com/hellofresh/goengine/v2/driver/sql"
+	"github.com/hellofresh/goengine/v2/extension/amqp"
+	goengineLogger "github.com/hellofresh/goengine/v2/extension/logrus"
 )
 
 func TestListener_Listen(t *testing.T) {
@@ -79,47 +80,42 @@ func TestListener_Listen(t *testing.T) {
 		ctx, ctxCancel := context.WithTimeout(context.Background(), time.Second)
 		defer ctxCancel()
 
-		var consumeCalls []time.Time
+		var waitCalls []time.Duration
 		consume := func() (io.Closer, <-chan libamqp.Delivery, error) {
-			consumeCalls = append(consumeCalls, time.Now())
-			if len(consumeCalls) == 5 {
+			if len(waitCalls) == 5 {
 				ctxCancel()
 			}
 
-			return nil, nil, fmt.Errorf("failure %d", len(consumeCalls))
+			return nil, nil, fmt.Errorf("failure %d", len(waitCalls))
 		}
 
 		logger, loggerHook := getLogger()
 
 		listener, err := amqp.NewListener(consume, time.Millisecond, 6*time.Millisecond, logger)
 		ensure.NoError(err)
 
+		listener.WithWaitFn(func(d time.Duration) {
+			waitCalls = append(waitCalls, d)
+		})
+
 		err = listener.Listen(ctx, func(ctx context.Context, notification *sql.ProjectionNotification) error {
 			ensure.Fail("Trigger should ever be called")
 			return nil
 		})
 
 		ensure.Equal(context.Canceled, err)
 
-		reconnectIntervals := []time.Duration{time.Millisecond, time.Millisecond * 2, time.Millisecond * 4, time.Millisecond * 6, time.Millisecond * 6}
-		ensure.Len(consumeCalls, len(reconnectIntervals))
-		for i := 1; i < len(reconnectIntervals); i++ {
-			expectedInterval := reconnectIntervals[i-1]
-			interval := consumeCalls[i].Sub(consumeCalls[i-1])
-
-			if expectedInterval > interval || interval > (expectedInterval+time.Millisecond*2) {
-				assert.Fail(t, fmt.Sprintf("Invalid interval after consume %d (got %s expected between %s and %s)", i, interval, expectedInterval, expectedInterval+time.Millisecond))
-			}
-		}
+		reconnectIntervals := []time.Duration{time.Millisecond, time.Millisecond * 2, time.Millisecond * 4, time.Millisecond * 6, time.Millisecond * 6, time.Millisecond * 6}
+		ensure.Equal(waitCalls, reconnectIntervals)
 
 		// Ensure we get log output
 		logEntries := loggerHook.AllEntries()
-		ensure.Len(logEntries, len(reconnectIntervals))
+		ensure.Len(logEntries, len(waitCalls))
 		for i, log := range logEntries {
 			assert.Equal(t, log.Level, logrus.ErrorLevel)
 			assert.Equal(t, log.Message, "failed to start consuming amqp messages")
-			assert.Equal(t, fmt.Errorf("failure %d", i+1), log.Data["error"])
-			assert.Equal(t, reconnectIntervals[i].String(), log.Data["reconnect_in"])
+			assert.Equal(t, fmt.Errorf("failure %d", i), log.Data["error"])
+			assert.Equal(t, waitCalls[i].String(), log.Data["reconnect_in"])
 		}
 	})