Fixed flaky tests in scheduler

intelsdi-x · Mar 24, 2017 · 44f4418 · 44f4418
1 parent 4b98fbb
commit 44f4418
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 19 deletions.
diff --git a/scheduler/distributed_task_test.go b/scheduler/distributed_task_test.go
@@ -325,12 +325,13 @@ func TestDistributedSubscriptions(t *testing.T) {
 					So(localMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 					So(remoteMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 				})
-				Convey("Task should be ended after an interval", func() {
+				Convey("Task should be ended after one interval", func() {
 					// wait for the end of the task (or timeout)
 					select {
 					case <-lse.Ended:
-					case <-time.After(time.Duration(interval.Nanoseconds()*int64(count)+interval.Nanoseconds()) + 1*time.Second):
+					case <-time.After(time.Duration(int64(count)*interval.Nanoseconds()) + 1*time.Second):
 					}
+
 					So(t.State(), ShouldEqual, core.TaskEnded)
 
 					Convey("So all dependencies should have been usubscribed", func() {
@@ -340,6 +341,9 @@ func TestDistributedSubscriptions(t *testing.T) {
 				})
 			})
 			Convey("Task is expected to run until reaching determined stop time", func() {
+				lse := fixtures.NewListenToSchedulerEvent()
+				s.eventManager.RegisterHandler("Scheduler.TaskEnded", lse)
+
 				startWait := time.Millisecond * 50
 				windowSize := time.Millisecond * 500
 				interval := time.Millisecond * 100
@@ -366,11 +370,11 @@ func TestDistributedSubscriptions(t *testing.T) {
 					So(remoteMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 				})
 				Convey("Task should have been ended after reaching the end of window", func() {
-					// wait for the end of determined window
-					time.Sleep(startWait + windowSize)
-					// wait an interval to be sure that the task state has been updated
-					// we are ok to extend sleeping by 100ms to allow to complete post-schedule activities
-					time.Sleep(interval + time.Millisecond*100)
+					// wait for the end of the task (or timeout)
+					select {
+					case <-lse.Ended:
+					case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
+					}
 
 					// check if the task has ended
 					So(t.State(), ShouldEqual, core.TaskEnded)

diff --git a/scheduler/scheduler_medium_test.go b/scheduler/scheduler_medium_test.go
@@ -266,17 +266,20 @@ func TestCreateTask(t *testing.T) {
 				// wait for task ended event (or timeout)
 				select {
 				case <-lse.Ended:
-				case <-time.After(stop.Add(1 * time.Second).Sub(start)):
+				case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
 				}
-				// check if the task is ended
+
 				So(tsk.State(), ShouldEqual, core.TaskEnded)
 			})
 		})
 	}) //end of tests for a windowed scheduler
 
 	Convey("Calling CreateTask for a simple/windowed schedule with determined the count of runs", t, func() {
 		Convey("Single run task firing immediately", func() {
-			sch := schedule.NewWindowedSchedule(interval, nil, nil, 1)
+			lse := fixtures.NewListenToSchedulerEvent()
+			s.eventManager.RegisterHandler("Scheduler.TaskEnded", lse)
+			count := uint(1)
+			sch := schedule.NewWindowedSchedule(interval, nil, nil, count)
 			tsk, errs := s.CreateTask(sch, w, false)
 			So(errs.Errors(), ShouldBeEmpty)
 			So(tsk, ShouldNotBeNil)
@@ -285,11 +288,11 @@ func TestCreateTask(t *testing.T) {
 			task.Spin()
 
 			Convey("the task should be ended after reaching the end of window", func() {
-				// wait an interval to be sure that the task state has been updated
-				// we are ok to extend sleeping by 100ms to allow to complete post-schedule activities
-				time.Sleep(interval + time.Millisecond*100)
-				// check if the task is ended
-				So(tsk.State(), ShouldEqual, core.TaskEnded)
+				// wait for task ended event (or timeout)
+				select {
+				case <-lse.Ended:
+				case <-time.After(time.Duration(int64(count)*interval.Nanoseconds()) + 1*time.Second):
+				}
 			})
 		})
 		Convey("Single run task firing on defined start time", func() {
@@ -308,7 +311,7 @@ func TestCreateTask(t *testing.T) {
 				// wait for task ended event (or timeout)
 				select {
 				case <-lse.Ended:
-				case <-time.After(time.Duration(interval.Nanoseconds()*int64(count)+interval.Nanoseconds()) + 1*time.Second):
+				case <-time.After(time.Duration(int64(count)*interval.Nanoseconds()) + 1*time.Second):
 				}
 				// check if the task is ended
 				So(tsk.State(), ShouldEqual, core.TaskEnded)
@@ -422,8 +425,9 @@ func TestStopTask(t *testing.T) {
 		// wait for task ended event (or timeout)
 		select {
 		case <-lse.Ended:
-		case <-time.After(stop.Add(1 * time.Second).Sub(start)):
+		case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
 		}
+
 		// check if the task is ended
 		So(tsk.State(), ShouldEqual, core.TaskEnded)
 
@@ -510,7 +514,7 @@ func TestStartTask(t *testing.T) {
 		// wait for task ended event (or timeout)
 		select {
 		case <-lse.Ended:
-		case <-time.After(stop.Add(1 * time.Second).Sub(start)):
+		case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
 		}
 
 		// check if the task is ended
@@ -609,8 +613,9 @@ func TestEnableTask(t *testing.T) {
 		// wait for task ended event (or timeout)
 		select {
 		case <-lse.Ended:
-		case <-time.After(stop.Add(1 * time.Second).Sub(start)):
+		case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
 		}
+
 		// check if the task is ended
 		So(tsk.State(), ShouldEqual, core.TaskEnded)