diff --git a/vaadin-testbench-unit-junit5/src/test/java/com/vaadin/testbench/unit/SignalsTest.java b/vaadin-testbench-unit-junit5/src/test/java/com/vaadin/testbench/unit/SignalsTest.java index 4683fddcb..246b8d312 100644 --- a/vaadin-testbench-unit-junit5/src/test/java/com/vaadin/testbench/unit/SignalsTest.java +++ b/vaadin-testbench-unit-junit5/src/test/java/com/vaadin/testbench/unit/SignalsTest.java @@ -9,6 +9,7 @@ package com.vaadin.testbench.unit; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import com.example.base.signals.SignalsView; @@ -16,6 +17,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import com.vaadin.flow.signals.SignalEnvironment; + @ViewPackages(packages = "com.example.base.signals") @Timeout(10) public class SignalsTest extends UIUnitTest { @@ -67,6 +70,30 @@ void attachedComponent_triggerSignalFromNonUIThreadThroughComponentEffect_effect Assertions.assertEquals("Counter: 10", counterTester.getText()); } + @Test + void effectDispatcher_routesToTestQueue_notServiceThreadPool() + throws InterruptedException { + // On the test thread, both VaadinServiceEnvironment (from + // MockVaadin) and TestSignalEnvironment are active. The default + // effect dispatcher must resolve to TestSignalEnvironment's queue + // (via registerFirst) so that runPendingSignalsTasks() can drive + // effect execution deterministically. Without registerFirst, + // VaadinServiceEnvironment's thread pool would be used instead, + // making effects run asynchronously outside the test's control. + navigate(SignalsView.class); + + var latch = new CountDownLatch(1); + SignalEnvironment.getDefaultEffectDispatcher() + .execute(latch::countDown); + + Assertions.assertFalse(latch.await(50, TimeUnit.MILLISECONDS), + "Task should be queued, not executed immediately on a " + + "thread pool"); + runPendingSignalsTasks(); + Assertions.assertTrue(latch.await(0, TimeUnit.MILLISECONDS), + "Task should have executed after draining the test queue"); + } + @Test void attachedComponent_slowEffect_effectEvaluatedAsynchronously() { var view = navigate(SignalsView.class); diff --git a/vaadin-testbench-unit-shared/src/main/java/com/vaadin/testbench/unit/TestSignalEnvironment.java b/vaadin-testbench-unit-shared/src/main/java/com/vaadin/testbench/unit/TestSignalEnvironment.java index 960a63536..aca04ba59 100644 --- a/vaadin-testbench-unit-shared/src/main/java/com/vaadin/testbench/unit/TestSignalEnvironment.java +++ b/vaadin-testbench-unit-shared/src/main/java/com/vaadin/testbench/unit/TestSignalEnvironment.java @@ -25,8 +25,10 @@ *
* How it works: *
* If a {@link VaadinSession} lock is held by the current thread, it is - * temporarily released during the wait for the first task, allowing - * background threads to acquire the lock and enqueue tasks. The lock is - * automatically reacquired before this method returns. + * temporarily released while polling for tasks, allowing background threads + * to acquire the lock and enqueue tasks. The lock is reacquired before + * running each task and released again before the next poll. * *
* If the current thread is interrupted while waiting for tasks, the method * restores the interrupt status and fails with an {@link AssertionError}. * * @param maxWaitTime - * the maximum time to wait for the first task to arrive in the + * the maximum time to wait for the next task to arrive in the * given time unit. If <= 0, returns immediately if no tasks * are available. * @param unit @@ -138,7 +140,8 @@ private Executor createTaskEnqueueExecutor() { * @return {@code true} if any pending Signals tasks were processed. */ boolean runPendingTasks(long maxWaitTime, TimeUnit unit) { - long waitMillis = unit.toMillis(maxWaitTime); + long deadlineMillis = System.currentTimeMillis() + + unit.toMillis(maxWaitTime); VaadinSession session = VaadinSession.getCurrent(); boolean hadLock = false; if (session != null && session.hasLock()) { @@ -146,35 +149,50 @@ boolean runPendingTasks(long maxWaitTime, TimeUnit unit) { session.unlock(); } try { - // Wait for the first task with the specified timeout - Runnable task; - try { - task = waitMillis > 0 - ? tasks.poll(waitMillis, TimeUnit.MILLISECONDS) - : tasks.poll(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new AssertionError( - "Thread interrupted while waiting for pending Signals tasks"); - } - if (task == null) { - LoggerFactory.getLogger(TestSignalEnvironment.class).debug( - "No pending Signals tasks found after waiting for {} {}", - maxWaitTime, unit); - return false; - } - while (task != null) { - task.run(); - // Process remaining tasks immediately, do not wait for - // additional tasks to be enqueued - task = tasks.poll(); + boolean processedAny = false; + while (true) { + long remainingMillis = deadlineMillis + - System.currentTimeMillis(); + Runnable task; + try { + task = remainingMillis > 0 + ? tasks.poll(remainingMillis, TimeUnit.MILLISECONDS) + : tasks.poll(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError( + "Thread interrupted while waiting for pending Signals tasks"); + } + if (task == null) { + if (!processedAny) { + LoggerFactory.getLogger(TestSignalEnvironment.class) + .debug("No pending Signals tasks found after waiting for {} {}", + maxWaitTime, unit); + } + break; + } + // Re-acquire the session lock before running the task so + // that DOM operations (which assert the lock is held) work + // correctly when the effect runs directly on the test + // thread instead of going through ui.access(). + if (hadLock) { + session.lock(); + } + try { + task.run(); + } finally { + if (hadLock) { + session.unlock(); + } + } + processedAny = true; } + return processedAny; } finally { if (hadLock) { session.lock(); } } - return true; } }