orchestrator: Make global tests more robust; use generic Expect function

The global orchestrator tests were pausing 200 ms to wait for some
events to pass. It's more reliable to just consume these specific
events. This changes them to do that, and also replaces
testutils.Expect* with a single Expect function. This is nicer than
having a separate function for every event type. The Checks in the event
structure can be used to match specific event properties.

In the future, I may also convert testutils.Watch* to a similar Watch
function.

Signed-off-by: Aaron Lehmann <aaron.lehmann@docker.com>

manager/orchestrator/global/global_test.go

@@ -3,6 +3,7 @@ package global
 import (
 	"testing"
 
+	"github.com/docker/go-events"
 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/manager/orchestrator/testutils"
 	"github.com/docker/swarmkit/manager/state"
@@ -60,7 +61,7 @@ var (
 	}
 )
 
-func SetupCluster(t *testing.T, store *store.MemoryStore) {
+func SetupCluster(t *testing.T, store *store.MemoryStore, watch chan events.Event) *api.Task {
 	ctx := context.Background()
 	// Start the global orchestrator.
 	global := NewGlobalOrchestrator(store)
@@ -69,7 +70,15 @@ func SetupCluster(t *testing.T, store *store.MemoryStore) {
 	}()
 
 	addService(t, store, service1)
+	testutils.Expect(t, watch, state.EventCreateService{})
+	testutils.Expect(t, watch, state.EventCommit{})
+
 	addNode(t, store, node1)
+	testutils.Expect(t, watch, state.EventCreateNode{})
+	testutils.Expect(t, watch, state.EventCommit{})
+
+	// return task creation from orchestrator
+	return testutils.WatchTaskCreate(t, watch)
 }
 
 func TestSetup(t *testing.T) {
@@ -80,9 +89,8 @@ func TestSetup(t *testing.T) {
 	watch, cancel := state.Watch(store.WatchQueue() /*state.EventCreateTask{}, state.EventUpdateTask{}*/)
 	defer cancel()
 
-	SetupCluster(t, store)
+	observedTask1 := SetupCluster(t, store, watch)
 
-	observedTask1 := testutils.WatchTaskCreate(t, watch)
 	assert.Equal(t, observedTask1.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask1.ServiceAnnotations.Name, "name1")
 	assert.Equal(t, observedTask1.NodeID, "id1")
@@ -93,12 +101,10 @@ func TestAddNode(t *testing.T) {
 	assert.NotNil(t, store)
 	defer store.Close()
 
-	SetupCluster(t, store)
-
 	watch, cancel := state.Watch(store.WatchQueue())
 	defer cancel()
 
-	testutils.SkipEvents(t, watch)
+	SetupCluster(t, store, watch)
 
 	addNode(t, store, node2)
 	observedTask2 := testutils.WatchTaskCreate(t, watch)
@@ -112,12 +118,10 @@ func TestDeleteNode(t *testing.T) {
 	assert.NotNil(t, store)
 	defer store.Close()
 
-	SetupCluster(t, store)
-
 	watch, cancel := state.Watch(store.WatchQueue())
 	defer cancel()
 
-	testutils.SkipEvents(t, watch)
+	SetupCluster(t, store, watch)
 
 	deleteNode(t, store, node1)
 	// task should be set to dead
@@ -131,12 +135,10 @@ func TestNodeAvailability(t *testing.T) {
 	assert.NotNil(t, store)
 	defer store.Close()
 
-	SetupCluster(t, store)
-
 	watch, cancel := state.Watch(store.WatchQueue())
 	defer cancel()
 
-	testutils.SkipEvents(t, watch)
+	SetupCluster(t, store, watch)
 
 	node1.Status.State = api.NodeStatus_READY
 	node1.Spec.Availability = api.NodeAvailabilityActive
@@ -163,12 +165,10 @@ func TestAddService(t *testing.T) {
 	assert.NotNil(t, store)
 	defer store.Close()
 
-	SetupCluster(t, store)
-
 	watch, cancel := state.Watch(store.WatchQueue())
 	defer cancel()
 
-	testutils.SkipEvents(t, watch)
+	SetupCluster(t, store, watch)
 
 	addService(t, store, service2)
 	observedTask := testutils.WatchTaskCreate(t, watch)
@@ -182,12 +182,10 @@ func TestDeleteService(t *testing.T) {
 	assert.NotNil(t, store)
 	defer store.Close()
 
-	SetupCluster(t, store)
-
 	watch, cancel := state.Watch(store.WatchQueue())
 	defer cancel()
 
-	testutils.SkipEvents(t, watch)
+	SetupCluster(t, store, watch)
 
 	deleteService(t, store, service1)
 	// task should be deleted
@@ -204,10 +202,8 @@ func TestRemoveTask(t *testing.T) {
 	watch, cancel := state.Watch(store.WatchQueue() /*state.EventCreateTask{}, state.EventUpdateTask{}*/)
 	defer cancel()
 
-	SetupCluster(t, store)
+	observedTask1 := SetupCluster(t, store, watch)
 
-	// get the task
-	observedTask1 := testutils.WatchTaskCreate(t, watch)
 	assert.Equal(t, observedTask1.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask1.ServiceAnnotations.Name, "name1")
 	assert.Equal(t, observedTask1.NodeID, "id1")

manager/orchestrator/replicated/restart_test.go

@@ -77,16 +77,16 @@ func TestOrchestratorRestartOnAny(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask3 := testutils.WatchTaskCreate(t, watch)
 	assert.Equal(t, observedTask3.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask3.ServiceAnnotations.Name, "name1")
 
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	observedTask4 := testutils.WatchTaskUpdate(t, watch)
 	assert.Equal(t, observedTask4.DesiredState, api.TaskStateRunning)
@@ -100,16 +100,16 @@ func TestOrchestratorRestartOnAny(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask5 := testutils.WatchTaskCreate(t, watch)
 	assert.Equal(t, observedTask5.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask5.ServiceAnnotations.Name, "name1")
 
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	observedTask6 := testutils.WatchTaskUpdate(t, watch)
 	assert.Equal(t, observedTask6.DesiredState, api.TaskStateRunning)
@@ -180,17 +180,17 @@ func TestOrchestratorRestartOnFailure(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask3 := testutils.WatchTaskCreate(t, watch)
 	assert.Equal(t, observedTask3.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask3.DesiredState, api.TaskStateReady)
 	assert.Equal(t, observedTask3.ServiceAnnotations.Name, "name1")
 
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	observedTask4 := testutils.WatchTaskUpdate(t, watch)
 	assert.Equal(t, observedTask4.DesiredState, api.TaskStateRunning)
@@ -204,11 +204,11 @@ func TestOrchestratorRestartOnFailure(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	select {
 	case <-watch:
@@ -280,11 +280,11 @@ func TestOrchestratorRestartOnNone(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	select {
 	case <-watch:
@@ -300,10 +300,10 @@ func TestOrchestratorRestartOnNone(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	select {
 	case <-watch:
@@ -377,13 +377,13 @@ func TestOrchestratorRestartDelay(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask3 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask3.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask3.DesiredState, api.TaskStateReady)
 	assert.Equal(t, observedTask3.ServiceAnnotations.Name, "name1")
@@ -470,13 +470,13 @@ func TestOrchestratorRestartMaxAttempts(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask3 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask3.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask3.DesiredState, api.TaskStateReady)
 	assert.Equal(t, observedTask3.ServiceAnnotations.Name, "name1")
@@ -502,18 +502,18 @@ func TestOrchestratorRestartMaxAttempts(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask5 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask5.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask5.DesiredState, api.TaskStateReady)
 
 	observedTask6 := testutils.WatchTaskUpdate(t, watch) // task gets started after a delay
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask6.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask6.DesiredState, api.TaskStateRunning)
 	assert.Equal(t, observedTask6.ServiceAnnotations.Name, "name1")
@@ -526,10 +526,10 @@ func TestOrchestratorRestartMaxAttempts(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	select {
 	case <-watch:
@@ -604,13 +604,13 @@ func TestOrchestratorRestartWindow(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask3 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask3.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask3.DesiredState, api.TaskStateReady)
 	assert.Equal(t, observedTask3.ServiceAnnotations.Name, "name1")
@@ -636,19 +636,19 @@ func TestOrchestratorRestartWindow(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask5 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask5.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask5.DesiredState, api.TaskStateReady)
 	assert.Equal(t, observedTask5.ServiceAnnotations.Name, "name1")
 
 	observedTask6 := testutils.WatchTaskUpdate(t, watch) // task gets started after a delay
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask6.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask6.DesiredState, api.TaskStateRunning)
 	assert.Equal(t, observedTask6.ServiceAnnotations.Name, "name1")
@@ -661,10 +661,10 @@ func TestOrchestratorRestartWindow(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
 
 	select {
 	case <-watch:
@@ -684,12 +684,12 @@ func TestOrchestratorRestartWindow(t *testing.T) {
 		return nil
 	})
 	assert.NoError(t, err)
-	testutils.ExpectTaskUpdate(t, watch)
-	testutils.ExpectCommit(t, watch)
-	testutils.ExpectTaskUpdate(t, watch)
+	testutils.Expect(t, watch, state.EventUpdateTask{})
+	testutils.Expect(t, watch, state.EventCommit{})
+	testutils.Expect(t, watch, state.EventUpdateTask{})
 
 	observedTask7 := testutils.WatchTaskCreate(t, watch)
-	testutils.ExpectCommit(t, watch)
+	testutils.Expect(t, watch, state.EventCommit{})
 	assert.Equal(t, observedTask7.Status.State, api.TaskStateNew)
 	assert.Equal(t, observedTask7.DesiredState, api.TaskStateReady)