ae: add remaining test cases

agent/ae/ae.go

@@ -212,8 +212,8 @@ func (s *StateSyncer) nextFSMState(fs fsmState) fsmState {
 	}
 }
 
-// event defines a timing or notification event from a multiple
-// timers and channels.
+// event defines a timing or notification event from multiple timers and
+// channels.
 type event string
 
 const (
@@ -224,7 +224,9 @@ const (
 )
 
 // retrySyncFullEventFn waits for an event which triggers a retry
-// of a full sync or a termination signal.
+// of a full sync or a termination signal. This function should not be
+// called directly but through s.retryFullSyncState to allow mocking for
+// testing.
 func (s *StateSyncer) retrySyncFullEventFn() event {
 	select {
 	// trigger a full sync immediately.
@@ -249,7 +251,9 @@ func (s *StateSyncer) retrySyncFullEventFn() event {
 }
 
 // syncChangesEventFn waits for a event which either triggers a full
-// or a partial sync or a termination signal.
+// or a partial sync or a termination signal. This function should not
+// be called directly but through s.syncChangesEvent to allow mocking
+// for testing.
 func (s *StateSyncer) syncChangesEventFn() event {
 	select {
 	// trigger a full sync immediately
@@ -276,9 +280,16 @@ func (s *StateSyncer) syncChangesEventFn() event {
 	}
 }
 
+// stubbed out for testing
+var libRandomStagger = lib.RandomStagger
+
+// staggerFn returns a random duration which depends on the cluster size
+// and a random factor which should provide some timely distribution of
+// cluster wide events. This function should not be called directly
+// but through s.stagger to allow mocking for testing.
 func (s *StateSyncer) staggerFn(d time.Duration) time.Duration {
 	f := scaleFactor(s.ClusterSize())
-	return lib.RandomStagger(time.Duration(f) * d)
+	return libRandomStagger(time.Duration(f) * d)
 }
 
 // Pause temporarily disables sync runs.

agent/ae/ae_test.go

@@ -9,6 +9,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/hashicorp/consul/lib"
 )
 
 func TestAE_scaleFactor(t *testing.T) {
@@ -77,6 +79,20 @@ func TestAE_Pause_ResumeTriggersSyncChanges(t *testing.T) {
 	}
 }
 
+func TestAE_staggerDependsOnClusterSize(t *testing.T) {
+	libRandomStagger = func(d time.Duration) time.Duration { return d }
+	defer func() { libRandomStagger = lib.RandomStagger }()
+
+	l := testSyncer()
+	if got, want := l.staggerFn(10*time.Millisecond), 10*time.Millisecond; got != want {
+		t.Fatalf("got %v want %v", got, want)
+	}
+	l.ClusterSize = func() int { return 256 }
+	if got, want := l.staggerFn(10*time.Millisecond), 20*time.Millisecond; got != want {
+		t.Fatalf("got %v want %v", got, want)
+	}
+}
+
 func TestAE_Run_SyncFullBeforeChanges(t *testing.T) {
 	shutdownCh := make(chan struct{})
 	state := &mock{
@@ -106,17 +122,29 @@ func TestAE_Run_SyncFullBeforeChanges(t *testing.T) {
 }
 
 func TestAE_Run_Quit(t *testing.T) {
-	// start timer which explodes if runFSM does not quit
-	tm := time.AfterFunc(time.Second, func() { panic("timeout") })
+	t.Run("Run panics without ClusterSize", func(t *testing.T) {
+		defer func() {
+			err := recover()
+			if err == nil {
+				t.Fatal("Run should panic")
+			}
+		}()
+		l := testSyncer()
+		l.ClusterSize = nil
+		l.Run()
+	})
+	t.Run("runFSM quits", func(t *testing.T) {
+		// start timer which explodes if runFSM does not quit
+		tm := time.AfterFunc(time.Second, func() { panic("timeout") })
 
-	l := testSyncer()
-	l.runFSM(fullSyncState, func(fsmState) fsmState { return doneState })
-	// should just quit
-	tm.Stop()
+		l := testSyncer()
+		l.runFSM(fullSyncState, func(fsmState) fsmState { return doneState })
+		// should just quit
+		tm.Stop()
+	})
 }
 
 func TestAE_FSM(t *testing.T) {
-
 	t.Run("fullSyncState", func(t *testing.T) {
 		t.Run("Paused -> retryFullSyncState", func(t *testing.T) {
 			l := testSyncer()
@@ -220,6 +248,69 @@ func TestAE_FSM(t *testing.T) {
 				t.Fatalf("got state %v want %v", got, want)
 			}
 		})
+		t.Run("invalid event -> panic ", func(t *testing.T) {
+			defer func() {
+				err := recover()
+				if err == nil {
+					t.Fatal("invalid event should panic")
+				}
+			}()
+			test(event("invalid"), fsmState(""))
+		})
+	})
+	t.Run("invalid state -> panic ", func(t *testing.T) {
+		defer func() {
+			err := recover()
+			if err == nil {
+				t.Fatal("invalid state should panic")
+			}
+		}()
+		l := testSyncer()
+		l.nextFSMState(fsmState("invalid"))
+	})
+}
+
+func TestAE_RetrySyncFullEvent(t *testing.T) {
+	t.Run("trigger shutdownEvent", func(t *testing.T) {
+		l := testSyncer()
+		l.ShutdownCh = make(chan struct{})
+		evch := make(chan event)
+		go func() { evch <- l.retrySyncFullEvent() }()
+		close(l.ShutdownCh)
+		if got, want := <-evch, shutdownEvent; got != want {
+			t.Fatalf("got event %q want %q", got, want)
+		}
+	})
+	t.Run("trigger shutdownEvent during FullNotif", func(t *testing.T) {
+		l := testSyncer()
+		l.ShutdownCh = make(chan struct{})
+		evch := make(chan event)
+		go func() { evch <- l.retrySyncFullEvent() }()
+		l.SyncFull.Trigger()
+		time.Sleep(100 * time.Millisecond)
+		close(l.ShutdownCh)
+		if got, want := <-evch, shutdownEvent; got != want {
+			t.Fatalf("got event %q want %q", got, want)
+		}
+	})
+	t.Run("trigger syncFullNotifEvent", func(t *testing.T) {
+		l := testSyncer()
+		l.serverUpInterval = 10 * time.Millisecond
+		evch := make(chan event)
+		go func() { evch <- l.retrySyncFullEvent() }()
+		l.SyncFull.Trigger()
+		if got, want := <-evch, syncFullNotifEvent; got != want {
+			t.Fatalf("got event %q want %q", got, want)
+		}
+	})
+	t.Run("trigger syncFullTimerEvent", func(t *testing.T) {
+		l := testSyncer()
+		l.retryFailInterval = 10 * time.Millisecond
+		evch := make(chan event)
+		go func() { evch <- l.retrySyncFullEvent() }()
+		if got, want := <-evch, syncFullTimerEvent; got != want {
+			t.Fatalf("got event %q want %q", got, want)
+		}
 	})
 }