fix(core): flaky test SchedulerConditionTest and SchedulerScheduleTest

kestra-io · Jun 26, 2023 · 6f40205 · 6f40205
1 parent 6ffc411
commit 6f40205
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 14 deletions.
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerConditionTest.java
@@ -6,6 +6,7 @@
 import io.kestra.core.models.triggers.Trigger;
 import io.kestra.core.models.triggers.types.Schedule;
 import io.kestra.core.runners.FlowListeners;
+import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
 import jakarta.inject.Inject;
 import org.junitpioneer.jupiter.RetryingTest;
@@ -71,18 +72,14 @@ void schedule() throws Exception {
             .when(flowListenersServiceSpy)
             .flows();
 
-        // start the worker as it execute polling triggers
-        Worker worker = new Worker(applicationContext, 8, null);
-        worker.run();
-
         // scheduler
         try (AbstractScheduler scheduler = new DefaultScheduler(
             applicationContext,
             flowListenersServiceSpy,
-            triggerState
-        )) {
+            triggerState);
+             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
             // wait for execution
-            executionQueue.receive(SchedulerConditionTest.class, execution -> {
+            Runnable assertionStop = executionQueue.receive(SchedulerConditionTest.class, execution -> {
                 if (execution.getState().getCurrent() == State.Type.CREATED) {
                     executionQueue.emit(execution.withState(State.Type.SUCCESS));
 
@@ -94,8 +91,11 @@ void schedule() throws Exception {
                 assertThat(execution.getFlowId(), is(flow.getId()));
             });
 
+            worker.run();
             scheduler.run();
             queueCount.await(15, TimeUnit.SECONDS);
+            // needed for RetryingTest to work since there is no context cleaning between method => we have to clear assertion receiver manually
+            assertionStop.run();
 
             assertThat(queueCount.getCount(), is(0L));
         }

diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerScheduleTest.java
@@ -4,9 +4,11 @@
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.triggers.types.Schedule;
 import io.kestra.core.runners.FlowListeners;
+import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.RetryingTest;
 
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
@@ -56,7 +58,7 @@ protected AbstractScheduler scheduler(FlowListeners flowListenersServiceSpy) {
         );
     }
 
-    @Test
+    @RetryingTest(5)
     void schedule() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
@@ -70,14 +72,11 @@ void schedule() throws Exception {
             .when(flowListenersServiceSpy)
             .flows();
 
-        // start the worker as it execute polling triggers
-        Worker worker = new Worker(applicationContext, 8, null);
-        worker.run();
-
         // scheduler
-        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy)) {
+        try (AbstractScheduler scheduler = scheduler(flowListenersServiceSpy);
+             Worker worker = new TestMethodScopedWorker(applicationContext, 8, null)) {
             // wait for execution
-            executionQueue.receive(execution -> {
+            Runnable assertionStop = executionQueue.receive(execution -> {
                 assertThat(execution.getInputs().get("testInputs"), is("test-inputs"));
                 assertThat(execution.getInputs().get("def"), is("awesome"));
 
@@ -91,8 +90,11 @@ void schedule() throws Exception {
                 assertThat(execution.getFlowId(), is(flow.getId()));
             });
 
+            worker.run();
             scheduler.run();
             queueCount.await(1, TimeUnit.MINUTES);
+            // needed for RetryingTest to work since there is no context cleaning between method => we have to clear assertion receiver manually
+            assertionStop.run();
 
             assertThat(queueCount.getCount(), is(0L));
             assertThat(date.size(), is(3));