handle = Arc.container().instance(WorkflowEventStore.class)) {
+ LOG.debug("Clearing WorkflowEventStore before test method: {}", context.getTestMethod().getName());
+ if (handle.isAvailable()) {
+ handle.get().clear();
+ }
+ }
+ }
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/assertions/AsyncFlowAssertions.java b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/AsyncFlowAssertions.java
new file mode 100644
index 000000000..009b50ace
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/AsyncFlowAssertions.java
@@ -0,0 +1,445 @@
+package io.quarkiverse.flow.testing.assertions;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+import org.assertj.core.api.Assertions;
+
+import io.quarkiverse.flow.testing.WorkflowEventStore;
+import io.quarkiverse.flow.testing.events.EventType;
+import io.quarkiverse.flow.testing.events.RecordedWorkflowEvent;
+import io.serverlessworkflow.impl.WorkflowModel;
+
+public class AsyncFlowAssertions implements ConfigurableAssertions {
+
+ private final WorkflowEventStore eventStore;
+ private Duration timeout = Duration.ofSeconds(5);
+ private Duration pollInterval = Duration.ofMillis(100);
+ private String instanceIdFilter = null;
+ private RecordedWorkflowEvent lastWaitedEvent = null;
+ private boolean strictlyOrdered = false;
+ private int lastMatchedIndex = -1;
+
+ private AsyncFlowAssertions(WorkflowEventStore eventStore) {
+ this.eventStore = Objects.requireNonNull(eventStore, "eventStore cannot be null");
+ }
+
+ public static AsyncFlowAssertions assertWith(WorkflowEventStore store) {
+ return new AsyncFlowAssertions(store);
+ }
+
+ /**
+ * Transitions from async waiting to synchronous fluent assertions over the events
+ * recorded so far in the store.
+ *
+ * Call this after all required events have been waited for, then chain
+ * {@link ConfigurableAssertions} methods to make structural assertions.
+ *
+ *
+ * AsyncFlowAssertions.assertWith(store)
+ * .workflowCompleted()
+ * .andAssert()
+ * .strictly()
+ * .workflowStarted()
+ * .workflowCompleted();
+ *
+ *
+ * @return a {@link ConfigurableAssertions} backed by the current snapshot of the store
+ */
+ public ConfigurableAssertions andAssert() {
+ return AsyncFlowAssertions.assertWith(eventStore);
+ }
+
+ public AsyncFlowAssertions timeout(Duration timeout) {
+ if (timeout == null || timeout.isNegative() || timeout.isZero()) {
+ throw new IllegalArgumentException("timeout must be positive");
+ }
+ this.timeout = timeout;
+ return this;
+ }
+
+ public AsyncFlowAssertions pollInterval(Duration pollInterval) {
+ if (pollInterval == null || pollInterval.isNegative() || pollInterval.isZero()) {
+ throw new IllegalArgumentException("Poll interval must be positive");
+ }
+ this.pollInterval = pollInterval;
+ return this;
+ }
+
+ public AsyncFlowAssertions workflowStarted() {
+ waitForEvent(EventType.WORKFLOW_STARTED, e -> true);
+ return this;
+ }
+
+ @Override
+ public AsyncFlowAssertions workflowCompleted() {
+ waitForEvent(EventType.WORKFLOW_COMPLETED, e -> true);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowStartedEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_STARTED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowCompletedEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_COMPLETED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowFailedEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_FAILED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .as("Expected %d WORKFLOW_FAILED events", expected)
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowCanceledEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_CANCELED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .as("Expected %d WORKFLOW_CANCELED events", expected)
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowSuspendedEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_SUSPENDED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .as("Expected %d WORKFLOW_SUSPENDED events", expected)
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowResumedEventCount(int expected) {
+ Assertions.assertThat(waitForEventsPolling(EventType.WORKFLOW_RESUMED)
+ .stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .count())
+ .as("Expected %d WORKFLOW_RESUMED events", expected)
+ .isEqualTo(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasTaskStartedEventCount(int expected) {
+ List allSatisfied = waitForEventsPolling(EventType.TASK_STARTED);
+ Assertions.assertThat(allSatisfied)
+ .as("Expected %d TASK_COMPLETED events but found %d.", expected, allSatisfied.size())
+ .hasSize(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions hasTaskCompletedEventCount(int expected) {
+ List allSatisfied = waitForEventsPolling(EventType.TASK_COMPLETED);
+ Assertions.assertThat(allSatisfied)
+ .as("Expected %d TASK_COMPLETED events but found %d.", expected, allSatisfied.size())
+ .hasSize(expected);
+ return this;
+ }
+
+ @Override
+ public WorkflowAssertions workflowCompletedWithin(Duration duration) {
+ workflowStarted();
+ workflowCompleted();
+ return andAssert().workflowCompletedWithin(duration);
+ }
+
+ @Override
+ public WorkflowAssertions allEventsForInstance(String id) {
+ return andAssert().allEventsForInstance(id);
+ }
+
+ @Override
+ public TaskCompletionOrderAssertions assertTask(String taskName) {
+ taskCompleted(taskName);
+ return andAssert().assertTask(taskName);
+ }
+
+ @Override
+ public void withOutput(Consumer outputAssertion) {
+ if (lastWaitedEvent == null) {
+ throw new AssertionError("No event has been waited for yet. Call an event wait method first.");
+ }
+ WorkflowModel output = lastWaitedEvent.getOutput()
+ .orElseThrow(() -> new AssertionError(
+ "Event " + lastWaitedEvent.getType() + " does not have output"));
+ outputAssertion.accept(output);
+ }
+
+ @Override
+ public ConfigurableAssertions configure() {
+ return andAssert();
+ }
+
+ /**
+ * Waits for a workflow failed event.
+ *
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions workflowFailed() {
+ waitForEvent(EventType.WORKFLOW_FAILED, e -> true);
+ return this;
+ }
+
+ /**
+ * Waits for a workflow cancelled event.
+ *
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions workflowCancelled() {
+ waitForEvent(EventType.WORKFLOW_CANCELED, e -> true);
+ return this;
+ }
+
+ /**
+ * Waits for a workflow suspended event.
+ *
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions workflowSuspended() {
+ waitForEvent(EventType.WORKFLOW_SUSPENDED, e -> true);
+ return this;
+ }
+
+ /**
+ * Waits for a workflow resumed event.
+ *
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions workflowResumed() {
+ waitForEvent(EventType.WORKFLOW_RESUMED, e -> true);
+ return this;
+ }
+
+ /**
+ * Waits for a task started event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskStarted(String taskName) {
+ waitForEvent(EventType.TASK_STARTED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task completed event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskCompleted(String taskName) {
+ waitForEvent(EventType.TASK_COMPLETED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task failed event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskFailed(String taskName) {
+ waitForEvent(EventType.TASK_FAILED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task cancelled event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskCancelled(String taskName) {
+ waitForEvent(EventType.TASK_CANCELLED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task suspended event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskSuspended(String taskName) {
+ waitForEvent(EventType.TASK_SUSPENDED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task resumed event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskResumed(String taskName) {
+ waitForEvent(EventType.TASK_RESUMED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ /**
+ * Waits for a task retried event with the specified task name.
+ *
+ * @param taskName the task name to wait for
+ * @return this for method chaining
+ */
+ public AsyncFlowAssertions taskRetried(String taskName) {
+ waitForEvent(EventType.TASK_RETRIED,
+ e -> e.getTaskName().map(taskName::equals).orElse(false));
+ return this;
+ }
+
+ private void waitForEvent(EventType type, Predicate condition) {
+ Instant deadline = Instant.now().plus(timeout);
+ long pollMillis = pollInterval.toMillis();
+
+ while (Instant.now().isBefore(deadline)) {
+ List all = eventStore.getAll();
+
+ if (strictlyOrdered) {
+ // Only look at events that come after the last matched position
+ for (int i = lastMatchedIndex + 1; i < all.size(); i++) {
+ RecordedWorkflowEvent e = all.get(i);
+ if ((instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ && (type == null || e.getType() == type)
+ && condition.test(e)) {
+ lastMatchedIndex = i;
+ lastWaitedEvent = e;
+ return;
+ }
+ }
+ } else {
+ Optional event = all.stream()
+ .filter(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .filter(e -> type == null || e.getType() == type)
+ .filter(condition)
+ .findFirst();
+
+ if (event.isPresent()) {
+ lastWaitedEvent = event.get();
+ return;
+ }
+ }
+
+ try {
+ Thread.sleep(pollMillis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting for event", e);
+ }
+ }
+
+ String eventDescription = type != null ? type.toString() : "matching condition";
+ String instanceInfo = instanceIdFilter != null ? " for instance " + instanceIdFilter : "";
+ String orderInfo = strictlyOrdered ? " (strictly after position " + lastMatchedIndex + ")" : "";
+ throw new AssertionError(
+ String.format("Timeout waiting for event %s%s%s after %s. Current events: %d",
+ eventDescription, instanceInfo, orderInfo, timeout, eventStore.size()));
+ }
+
+ private List waitForEventsPolling(EventType type) {
+ Predicate composedFilter = buildFilter(type);
+ Instant deadline = Instant.now().plus(timeout);
+
+ while (true) {
+ List results = eventStore.getAll().stream()
+ .filter(composedFilter)
+ .toList();
+
+ if (!results.isEmpty()) {
+ return results;
+ }
+
+ if (Instant.now().isAfter(deadline)) {
+ return List.of();
+ }
+
+ LockSupport.parkNanos(pollInterval.toMillis() * 1_000_000L);
+ }
+ }
+
+ private Predicate buildFilter(EventType type) {
+ Predicate filter = event -> true;
+ return filter.and(e -> instanceIdFilter == null || instanceIdFilter.equals(e.getInstanceId()))
+ .and(e -> type == null || e.getType() == type);
+ }
+
+ /**
+ * Enables strict ordering for subsequent event waits.
+ * When active, each event is only matched if it appears after the position
+ * of the previously matched event in the store, enforcing the declared order.
+ *
+ * @return this for method chaining
+ */
+ @Override
+ public ConfigurableAssertions strictly() {
+ this.strictlyOrdered = true;
+ return this;
+ }
+
+ /**
+ * Restricts all subsequent event waits to the given workflow instance.
+ *
+ * @param instanceId the workflow instance ID to filter by; must not be null
+ * @return this for method chaining
+ */
+ @Override
+ public ConfigurableAssertions filteringBy(String instanceId) {
+ if (instanceId == null) {
+ throw new IllegalArgumentException("instanceId cannot be null");
+ }
+ this.instanceIdFilter = instanceId;
+ return this;
+ }
+
+ /**
+ * Resets all mutable state — instance filter, last waited event, timeout, and poll interval —
+ * back to their defaults so this instance can be reused for a new assertion chain.
+ *
+ * @return this for method chaining
+ */
+ @Override
+ public AsyncFlowAssertions reset() {
+ this.instanceIdFilter = null;
+ this.lastWaitedEvent = null;
+ this.strictlyOrdered = false;
+ this.lastMatchedIndex = -1;
+ this.timeout = Duration.ofSeconds(5);
+ this.pollInterval = Duration.ofMillis(100);
+ return this;
+ }
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/assertions/ConfigurableAssertions.java b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/ConfigurableAssertions.java
new file mode 100644
index 000000000..728d45209
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/ConfigurableAssertions.java
@@ -0,0 +1,26 @@
+package io.quarkiverse.flow.testing.assertions;
+
+/**
+ * Configurable assertions interface that allows setting up assertion behavior
+ * before executing assertions.
+ */
+public interface ConfigurableAssertions extends WorkflowAssertions {
+
+ /**
+ * Enables strict ordering mode for assertions.
+ *
+ * @return {@link ConfigurableAssertions} for further configurations
+ */
+ ConfigurableAssertions strictly();
+
+ /**
+ * Filters events to only include those for the specified workflow instance.
+ *
+ * @param id the workflow instance ID to filter by
+ * @return {@link ConfigurableAssertions} for further configurations
+ */
+ ConfigurableAssertions filteringBy(String id);
+
+ ConfigurableAssertions reset();
+
+}
\ No newline at end of file
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/assertions/FlowAssertions.java b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/FlowAssertions.java
new file mode 100644
index 000000000..3723f0a82
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/FlowAssertions.java
@@ -0,0 +1,360 @@
+package io.quarkiverse.flow.testing.assertions;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import org.assertj.core.api.Assertions;
+
+import io.quarkiverse.flow.testing.WorkflowEventStore;
+import io.quarkiverse.flow.testing.events.EventType;
+import io.quarkiverse.flow.testing.events.RecordedWorkflowEvent;
+import io.serverlessworkflow.impl.WorkflowInstance;
+import io.serverlessworkflow.impl.WorkflowModel;
+
+public class FlowAssertions implements ConfigurableAssertions {
+
+ private final List events;
+ private int current = 0;
+ private boolean strictly = false;
+ private RecordedWorkflowEvent currentVerifiedEvent = null;
+
+ FlowAssertions(List events) {
+ this.events = Collections.unmodifiableList(events);
+ }
+
+ public static ConfigurableAssertions assertWith(WorkflowEventStore eventStore) {
+ return new FlowAssertions(eventStore.getAll());
+ }
+
+ // ── ConfigurableAssertions ────────────────────────────────────────────────
+
+ @Override
+ public ConfigurableAssertions strictly() {
+ this.strictly = true;
+ return this;
+ }
+
+ @Override
+ public ConfigurableAssertions filteringBy(String id) {
+ Objects.requireNonNull(id, "instanceId must not be null");
+ List filtered = events.stream()
+ .filter(e -> id.equals(e.getInstanceId()))
+ .collect(Collectors.toList());
+ return new FlowAssertions(filtered);
+ }
+
+ @Override
+ public ConfigurableAssertions reset() {
+ this.current = 0;
+ this.strictly = false;
+ this.currentVerifiedEvent = null;
+ return this;
+ }
+
+ public FlowAssertions workflowStarted() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_STARTED);
+ } else {
+ Assertions.assertThat(events)
+ .as("At least one WORKFLOW_STARTED event")
+ .anySatisfy(e -> Assertions.assertThat(e.getType()).isEqualTo(EventType.WORKFLOW_STARTED));
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowCompleted() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_COMPLETED);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_COMPLETED)
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent).as("At least one WORKFLOW_COMPLETED event").isNotNull();
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowCompleted(WorkflowInstance instance) {
+ if (strictly) {
+ RecordedWorkflowEvent event = assertNextEventTypeIs(EventType.WORKFLOW_COMPLETED);
+ Assertions.assertThat(event.getInstanceId())
+ .as("Instance ID for WORKFLOW_COMPLETED").isEqualTo(instance.id());
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_COMPLETED && e.getInstanceId().equals(instance.id()))
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent)
+ .as("At least one WORKFLOW_COMPLETED for instance '%s'", instance.id()).isNotNull();
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowFailed() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_FAILED);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_FAILED)
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent).as("At least one WORKFLOW_FAILED event").isNotNull();
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowCancelled() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_CANCELED);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_CANCELED)
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent).as("At least one WORKFLOW_CANCELLED event").isNotNull();
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowSuspended() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_SUSPENDED);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_SUSPENDED)
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent).as("At least one WORKFLOW_SUSPENDED event").isNotNull();
+ }
+ return this;
+ }
+
+ public FlowAssertions workflowResumed() {
+ if (strictly) {
+ assertNextEventTypeIs(EventType.WORKFLOW_RESUMED);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_RESUMED)
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent).as("At least one WORKFLOW_RESUMED event").isNotNull();
+ }
+ return this;
+ }
+
+ // ── Task event assertions ─────────────────────────────────────────────────
+
+ public FlowAssertions taskStarted(String taskName) {
+ return assertTaskEvent(EventType.TASK_STARTED, taskName);
+ }
+
+ public FlowAssertions taskCompleted(String taskName) {
+ return assertTaskEvent(EventType.TASK_COMPLETED, taskName);
+ }
+
+ public FlowAssertions taskFailed(String taskName) {
+ return assertTaskEvent(EventType.TASK_FAILED, taskName);
+ }
+
+ public FlowAssertions taskCancelled(String taskName) {
+ return assertTaskEvent(EventType.TASK_CANCELLED, taskName);
+ }
+
+ public FlowAssertions taskSuspended(String taskName) {
+ return assertTaskEvent(EventType.TASK_SUSPENDED, taskName);
+ }
+
+ public FlowAssertions taskResumed(String taskName) {
+ return assertTaskEvent(EventType.TASK_RESUMED, taskName);
+ }
+
+ public FlowAssertions taskRetried(String taskName) {
+ return assertTaskEvent(EventType.TASK_RETRIED, taskName);
+ }
+
+ private FlowAssertions assertTaskEvent(EventType type, String taskName) {
+ if (strictly) {
+ RecordedWorkflowEvent event = assertNextEventTypeIs(type);
+ Assertions.assertThat(event.getTaskName())
+ .as("Task name for %s event", type).hasValue(taskName);
+ } else {
+ currentVerifiedEvent = events.stream()
+ .filter(e -> e.getType() == type)
+ .filter(e -> e.getTaskName().map(taskName::equals).orElse(false))
+ .findFirst().orElse(null);
+ Assertions.assertThat(currentVerifiedEvent)
+ .as("At least one %s event for task '%s'", type, taskName).isNotNull();
+ }
+ return this;
+ }
+
+ // ── Output / Error ────────────────────────────────────────────────────────
+
+ public void withOutput(Consumer outputAssertion) {
+ if (currentVerifiedEvent == null) {
+ throw new AssertionError("No event has been verified yet. Call an event assertion method first.");
+ }
+ WorkflowModel output = currentVerifiedEvent.getOutput()
+ .orElseThrow(() -> new AssertionError("Event " + currentVerifiedEvent.getType() + " has no output"));
+ outputAssertion.accept(output);
+ }
+
+ @Override
+ public ConfigurableAssertions configure() {
+ return this;
+ }
+
+ public FlowAssertions withError(Consumer errorAssertion) {
+ if (currentVerifiedEvent == null) {
+ throw new AssertionError("No event has been verified yet. Call an event assertion method first.");
+ }
+ Throwable error = currentVerifiedEvent.getError()
+ .orElseThrow(() -> new AssertionError("Event " + currentVerifiedEvent.getType() + " has no error"));
+ errorAssertion.accept(error);
+ return this;
+ }
+
+ // ── Count assertions ──────────────────────────────────────────────────────
+
+ @Override
+ public WorkflowAssertions hasWorkflowStartedEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_STARTED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowCompletedEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_COMPLETED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowFailedEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_FAILED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowCanceledEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_CANCELED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowSuspendedEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_SUSPENDED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasWorkflowResumedEventCount(int n) {
+ return hasEventTypeCount(EventType.WORKFLOW_RESUMED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasTaskStartedEventCount(int n) {
+ return hasEventTypeCount(EventType.TASK_STARTED, n);
+ }
+
+ @Override
+ public WorkflowAssertions hasTaskCompletedEventCount(int n) {
+ return hasEventTypeCount(EventType.TASK_COMPLETED, n);
+ }
+
+ public FlowAssertions hasEventTypeCount(EventType type, int expected) {
+ long count = events.stream().filter(e -> e.getType() == type).count();
+ Assertions.assertThat(count).as("Count of %s events", type).isEqualTo(expected);
+ return this;
+ }
+
+ // ── Structural assertions ─────────────────────────────────────────────────
+
+ public FlowAssertions workflowCompletedWithin(Duration duration) {
+ RecordedWorkflowEvent start = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_STARTED).findFirst()
+ .orElseThrow(() -> new AssertionError("No WORKFLOW_STARTED event found"));
+ RecordedWorkflowEvent end = events.stream()
+ .filter(e -> e.getType() == EventType.WORKFLOW_COMPLETED).findFirst()
+ .orElseThrow(() -> new AssertionError("No WORKFLOW_COMPLETED event found"));
+ Assertions.assertThat(Duration.between(start.getTimestamp(), end.getTimestamp()))
+ .as("Workflow execution duration").isLessThanOrEqualTo(duration);
+ return this;
+ }
+
+ public FlowAssertions allEventsForInstance(String instanceId) {
+ Assertions.assertThat(events)
+ .as("All events should belong to instance %s", instanceId)
+ .allMatch(e -> instanceId.equals(e.getInstanceId()));
+ return this;
+ }
+
+ public FlowAssertions allEventsForWorkflow(String workflowId) {
+ Assertions.assertThat(events)
+ .as("All events should belong to workflow %s", workflowId)
+ .allMatch(e -> workflowId.equals(e.getWorkflowId()));
+ return this;
+ }
+
+ // ── Task completion order ─────────────────────────────────────────────────
+
+ public TaskCompletionOrderAssertions assertTask(String taskName) {
+ return new TaskCompletedAssertions(findTaskCompletedEvent(taskName), taskName, this);
+ }
+
+ public final class TaskCompletedAssertions implements TaskCompletionOrderAssertions {
+ private final RecordedWorkflowEvent subject;
+ private final String subjectName;
+ private final FlowAssertions parent;
+
+ private TaskCompletedAssertions(RecordedWorkflowEvent subject, String subjectName, FlowAssertions parent) {
+ this.subject = subject;
+ this.subjectName = subjectName;
+ this.parent = parent;
+ }
+
+ public FlowAssertions completedBefore(String other) {
+ Assertions.assertThat(subject.getTimestamp())
+ .as("'%s' should complete before '%s'", subjectName, other)
+ .isBefore(findTaskCompletedEvent(other).getTimestamp());
+ return parent;
+ }
+
+ public FlowAssertions completedBeforeOrEqualTo(String other) {
+ Assertions.assertThat(subject.getTimestamp())
+ .as("'%s' should complete before or at the same time as '%s'", subjectName, other)
+ .isBeforeOrEqualTo(findTaskCompletedEvent(other).getTimestamp());
+ return parent;
+ }
+
+ public FlowAssertions completedAfter(String other) {
+ Assertions.assertThat(subject.getTimestamp())
+ .as("'%s' should complete after '%s'", subjectName, other)
+ .isAfter(findTaskCompletedEvent(other).getTimestamp());
+ return parent;
+ }
+
+ public FlowAssertions completedAfterOrEqualTo(String other) {
+ Assertions.assertThat(subject.getTimestamp())
+ .as("'%s' should complete after or at the same time as '%s'", subjectName, other)
+ .isAfterOrEqualTo(findTaskCompletedEvent(other).getTimestamp());
+ return parent;
+ }
+ }
+
+ // ── Internals ─────────────────────────────────────────────────────────────
+
+ private RecordedWorkflowEvent findTaskCompletedEvent(String taskName) {
+ return events.stream()
+ .filter(e -> e.getType() == EventType.TASK_COMPLETED)
+ .filter(e -> e.getTaskName().map(taskName::equals).orElse(false))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No TASK_COMPLETED event found for task: " + taskName));
+ }
+
+ private RecordedWorkflowEvent assertNextEventTypeIs(EventType expectedType) {
+ if (current >= events.size()) {
+ throw new AssertionError(String.format(
+ "Expected event %s at index %d, but only %d events were recorded",
+ expectedType, current, events.size()));
+ }
+ RecordedWorkflowEvent event = events.get(current);
+ Assertions.assertThat(event.getType()).as("Event type at index %d", current).isEqualTo(expectedType);
+ currentVerifiedEvent = event;
+ current++;
+ return event;
+ }
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/assertions/TaskCompletionOrderAssertions.java b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/TaskCompletionOrderAssertions.java
new file mode 100644
index 000000000..ab065660b
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/TaskCompletionOrderAssertions.java
@@ -0,0 +1,39 @@
+package io.quarkiverse.flow.testing.assertions;
+
+/**
+ * Fluent assertions for comparing the completion order of two tasks.
+ */
+public interface TaskCompletionOrderAssertions {
+
+ /**
+ * Asserts that the subject task completed strictly before {@code otherTask}.
+ *
+ * @param otherTask the task that must complete after the subject
+ * @return the parent {@link WorkflowAssertions} for further chaining
+ */
+ WorkflowAssertions completedBefore(String otherTask);
+
+ /**
+ * Asserts that the subject task completed before or at the same instant as {@code otherTask}.
+ *
+ * @param otherTask the task that must complete after or at the same time as the subject
+ * @return the parent {@link WorkflowAssertions} for further chaining
+ */
+ WorkflowAssertions completedBeforeOrEqualTo(String otherTask);
+
+ /**
+ * Asserts that the subject task completed strictly after {@code otherTask}.
+ *
+ * @param otherTask the task that must complete before the subject
+ * @return the parent {@link WorkflowAssertions} for further chaining
+ */
+ WorkflowAssertions completedAfter(String otherTask);
+
+ /**
+ * Asserts that the subject task completed after or at the same instant as {@code otherTask}.
+ *
+ * @param otherTask the task that must complete before or at the same time as the subject
+ * @return the parent {@link WorkflowAssertions} for further chaining
+ */
+ WorkflowAssertions completedAfterOrEqualTo(String otherTask);
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/assertions/WorkflowAssertions.java b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/WorkflowAssertions.java
new file mode 100644
index 000000000..d4ae910f8
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/assertions/WorkflowAssertions.java
@@ -0,0 +1,204 @@
+package io.quarkiverse.flow.testing.assertions;
+
+import java.time.Duration;
+import java.util.function.Consumer;
+
+import io.serverlessworkflow.impl.WorkflowModel;
+
+public interface WorkflowAssertions {
+
+ /**
+ * Asserts that a task cancelled event exists for the specified task name.
+ *
+ * @param taskName the expected task name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskCancelled(String taskName);
+
+ /**
+ * Asserts that a task suspended event exists for the specified task name.
+ *
+ * @param taskName the expected task name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskSuspended(String taskName);
+
+ /**
+ * Asserts that a task resumed event exists for the specified task name.
+ *
+ * @param taskName the expected task name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskResumed(String taskName);
+
+ /**
+ * Asserts that a task retried event exists for the specified task name.
+ *
+ * @param taskName the expected task name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskRetried(String taskName);
+
+ /**
+ * Asserts that a task with the name taskName was started.
+ *
+ * @param taskName the task's name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskStarted(String taskName);
+
+ /**
+ * Asserts that a task with the name taskName was completed.
+ *
+ * @param taskName the task's name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskCompleted(String taskName);
+
+ /**
+ * Asserts that a task with the name taskName was failed.
+ *
+ * @param taskName the task's name
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions taskFailed(String taskName);
+
+ /**
+ * Asserts that a workflow started event exists.
+ *
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions workflowStarted();
+
+ /**
+ * Asserts that a workflow cancelled event exists.
+ *
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions workflowCancelled();
+
+ /**
+ * Asserts that a workflow suspended event exists.
+ *
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions workflowSuspended();
+
+ /**
+ * Asserts that a workflow resumed event exists.
+ *
+ * @return {@link WorkflowAssertions} for further assertions
+ */
+ WorkflowAssertions workflowResumed();
+
+ /**
+ * Asserts that a workflow was completed.
+ *
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions workflowCompleted();
+
+ /**
+ * Asserts that a workflow was failed.
+ *
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions workflowFailed();
+
+ /**
+ * Asserts that exactly expected workflow started events were recorded.
+ *
+ * @param expected the expected number of workflow started events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowStartedEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected workflow completed events were recorded.
+ *
+ * @param expected the expected number of workflow completed events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowCompletedEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected workflow failed events were recorded.
+ *
+ * @param expected the expected number of workflow failed events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowFailedEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected workflow canceled events were recorded.
+ *
+ * @param expected the expected number of workflow canceled events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowCanceledEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected workflow canceled events were recorded.
+ *
+ * @param expected the expected number of workflow suspended events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowSuspendedEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected workflow resumed events were recorded.
+ *
+ * @param expected the expected number of workflow resumed events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasWorkflowResumedEventCount(int expected);
+
+ /**
+ * Entry point for fluent task-completion ordering assertions.
+ * Resolves the {@code TASK_COMPLETED} event for the given task and returns a
+ * {@link TaskCompletionOrderAssertions} that lets you compare its completion
+ * timestamp against another task.
+ *
+ *
+ * assertions.assertTask("taskA").completedBefore("taskB");
+ * assertions.assertTask("taskB").completedAfter("taskA");
+ *
+ *
+ * @param taskName the task whose completion event is the subject of the assertion
+ * @return a {@link TaskCompletionOrderAssertions} scoped to the resolved event
+ * @throws AssertionError if no {@code TASK_COMPLETED} event is found for {@code taskName}
+ */
+ TaskCompletionOrderAssertions assertTask(String taskName);
+
+ /**
+ * Asserts that exactly expected task started events were recorded.
+ *
+ * @param expected the expected number of task started events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasTaskStartedEventCount(int expected);
+
+ /**
+ * Asserts that exactly expected task completed events were recorded.
+ *
+ * @param expected the expected number of task completed events
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions hasTaskCompletedEventCount(int expected);
+
+ WorkflowAssertions allEventsForInstance(String id);
+
+ /**
+ * Asserts that the workflow completed within the given duration from
+ * the moment the first workflow started event was recorded.
+ *
+ * @param duration the maximum allowed elapsed time between workflow start and completion
+ * @return an instance of {@link WorkflowAssertions}
+ */
+ WorkflowAssertions workflowCompletedWithin(Duration duration);
+
+ void withOutput(Consumer outputAssertion);
+
+ ConfigurableAssertions configure();
+
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/events/EventType.java b/testing/src/main/java/io/quarkiverse/flow/testing/events/EventType.java
new file mode 100644
index 000000000..9af0555f3
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/events/EventType.java
@@ -0,0 +1,76 @@
+package io.quarkiverse.flow.testing.events;
+
+/**
+ * Enumeration of all workflow lifecycle event types that can be recorded during test execution.
+ */
+public enum EventType {
+ /**
+ * Workflow instance has started execution.
+ */
+ WORKFLOW_STARTED,
+
+ /**
+ * Workflow instance has completed successfully.
+ */
+ WORKFLOW_COMPLETED,
+
+ /**
+ * Workflow instance has failed with an error.
+ */
+ WORKFLOW_FAILED,
+
+ /**
+ * Workflow instance has been canceled.
+ */
+ WORKFLOW_CANCELED,
+
+ /**
+ * Workflow instance has been suspended.
+ */
+ WORKFLOW_SUSPENDED,
+
+ /**
+ * Workflow instance has been resumed from suspension.
+ */
+ WORKFLOW_RESUMED,
+
+ /**
+ * Workflow instance status has changed.
+ */
+ WORKFLOW_STATUS_CHANGED,
+
+ /**
+ * A task within the workflow has started execution.
+ */
+ TASK_STARTED,
+
+ /**
+ * A task within the workflow has completed successfully.
+ */
+ TASK_COMPLETED,
+
+ /**
+ * A task within the workflow has failed with an error.
+ */
+ TASK_FAILED,
+
+ /**
+ * A task within the workflow has been cancelled.
+ */
+ TASK_CANCELLED,
+
+ /**
+ * A task within the workflow has been suspended.
+ */
+ TASK_SUSPENDED,
+
+ /**
+ * A task within the workflow has been resumed from suspension.
+ */
+ TASK_RESUMED,
+
+ /**
+ * A task within the workflow has been retried after a failure.
+ */
+ TASK_RETRIED
+}
diff --git a/testing/src/main/java/io/quarkiverse/flow/testing/events/RecordedWorkflowEvent.java b/testing/src/main/java/io/quarkiverse/flow/testing/events/RecordedWorkflowEvent.java
new file mode 100644
index 000000000..36b99d6eb
--- /dev/null
+++ b/testing/src/main/java/io/quarkiverse/flow/testing/events/RecordedWorkflowEvent.java
@@ -0,0 +1,349 @@
+package io.quarkiverse.flow.testing.events;
+
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import io.serverlessworkflow.impl.WorkflowDefinitionId;
+import io.serverlessworkflow.impl.WorkflowModel;
+import io.serverlessworkflow.impl.lifecycle.TaskCancelledEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskCompletedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskFailedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskResumedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskRetriedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskStartedEvent;
+import io.serverlessworkflow.impl.lifecycle.TaskSuspendedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowCancelledEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowCompletedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowFailedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowResumedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowStartedEvent;
+import io.serverlessworkflow.impl.lifecycle.WorkflowSuspendedEvent;
+
+public class RecordedWorkflowEvent {
+
+ private final EventType type;
+ private final Instant timestamp;
+ private final WorkflowDefinitionId workflowId;
+ private final String instanceId;
+ private final Object originalEvent;
+ private final Map metadata;
+
+ private RecordedWorkflowEvent(Builder builder) {
+ this.type = builder.type;
+ this.timestamp = builder.timestamp;
+ this.workflowId = builder.workflowId;
+ this.instanceId = builder.instanceId;
+ this.originalEvent = builder.originalEvent;
+ this.metadata = new HashMap<>(builder.metadata);
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowStartedEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowStartedEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_STARTED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("workflowName", event.workflowContext().definition().workflow().getDocument().getName())
+ .addMetadata("workflowVersion",
+ event.workflowContext().definition().workflow().getDocument().getVersion())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowCompletedEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowCompletedEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_COMPLETED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("output", event.output())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowFailedEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowFailedEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_FAILED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("error", event.cause())
+ .addMetadata("errorMessage", event.cause() != null ? event.cause().getMessage() : null)
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowCancelledEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowCancelledEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_CANCELED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowSuspendedEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowSuspendedEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_SUSPENDED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a WorkflowResumedEvent.
+ */
+ public static RecordedWorkflowEvent from(WorkflowResumedEvent event) {
+ return builder()
+ .type(EventType.WORKFLOW_RESUMED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskStartedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskStartedEvent event) {
+ return builder()
+ .type(EventType.TASK_STARTED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskCompletedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskCompletedEvent event) {
+ return builder()
+ .type(EventType.TASK_COMPLETED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .addMetadata("output", event.taskContext().output())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskFailedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskFailedEvent event) {
+ return builder()
+ .type(EventType.TASK_FAILED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .addMetadata("error", event.cause())
+ .addMetadata("errorMessage", event.cause() != null ? event.cause().getMessage() : null)
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskCancelledEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskCancelledEvent event) {
+ return builder()
+ .type(EventType.TASK_CANCELLED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskSuspendedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskSuspendedEvent event) {
+ return builder()
+ .type(EventType.TASK_SUSPENDED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskResumedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskResumedEvent event) {
+ return builder()
+ .type(EventType.TASK_RESUMED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .build();
+ }
+
+ /**
+ * Creates a RecordedWorkflowEvent from a TaskRetriedEvent.
+ */
+ public static RecordedWorkflowEvent from(TaskRetriedEvent event) {
+ return builder()
+ .type(EventType.TASK_RETRIED)
+ .workflowId(event.workflowContext().definition().id())
+ .instanceId(event.workflowContext().instanceData().id())
+ .originalEvent(event)
+ .addMetadata("taskName", event.taskContext().taskName())
+ .addMetadata("taskId", event.taskContext().position().jsonPointer())
+ .build();
+ }
+
+ // Getters
+
+ public EventType getType() {
+ return type;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ public WorkflowDefinitionId getWorkflowId() {
+ return workflowId;
+ }
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public Object getOriginalEvent() {
+ return originalEvent;
+ }
+
+ public Map getMetadata() {
+ return new HashMap<>(metadata);
+ }
+
+ // Convenience methods for common metadata access
+
+ public Optional getTaskName() {
+ return Optional.ofNullable((String) metadata.get("taskName"));
+ }
+
+ public Optional getTaskId() {
+ return Optional.ofNullable((String) metadata.get("taskId"));
+ }
+
+ public Optional getOutput() {
+ return Optional.ofNullable((WorkflowModel) metadata.get("output"));
+ }
+
+ public Optional getError() {
+ return Optional.ofNullable((Throwable) metadata.get("error"));
+ }
+
+ public Optional getErrorMessage() {
+ return Optional.ofNullable((String) metadata.get("errorMessage"));
+ }
+
+ public boolean isWorkflowEvent() {
+ return type.name().startsWith("WORKFLOW_");
+ }
+
+ public boolean isTaskEvent() {
+ return type.name().startsWith("TASK_");
+ }
+
+ // Builder
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder {
+ private EventType type;
+ private Instant timestamp = Instant.now();
+ private WorkflowDefinitionId workflowId;
+ private String instanceId;
+ private Object originalEvent;
+ private Map metadata = new HashMap<>();
+
+ public Builder type(EventType type) {
+ this.type = type;
+ return this;
+ }
+
+ public Builder timestamp(Instant timestamp) {
+ this.timestamp = timestamp;
+ return this;
+ }
+
+ public Builder workflowId(WorkflowDefinitionId workflowId) {
+ this.workflowId = workflowId;
+ return this;
+ }
+
+ public Builder instanceId(String instanceId) {
+ this.instanceId = instanceId;
+ return this;
+ }
+
+ public Builder originalEvent(Object originalEvent) {
+ this.originalEvent = originalEvent;
+ return this;
+ }
+
+ public Builder addMetadata(String key, Object value) {
+ this.metadata.put(key, value);
+ return this;
+ }
+
+ public Builder metadata(Map metadata) {
+ this.metadata = new HashMap<>(metadata);
+ return this;
+ }
+
+ public RecordedWorkflowEvent build() {
+ if (type == null) {
+ throw new IllegalStateException("EventType is required");
+ }
+ if (workflowId == null) {
+ throw new IllegalStateException("WorkflowId is required");
+ }
+ if (instanceId == null) {
+ throw new IllegalStateException("InstanceId is required");
+ }
+ return new RecordedWorkflowEvent(this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "RecordedWorkflowEvent{" +
+ "type=" + type +
+ ", timestamp=" + timestamp +
+ ", workflowId='" + workflowId + '\'' +
+ ", instanceId='" + instanceId + '\'' +
+ ", taskName=" + getTaskName().orElse("N/A") +
+ '}';
+ }
+}
\ No newline at end of file
diff --git a/testing/src/main/resources/META-INF/beans.xml b/testing/src/main/resources/META-INF/beans.xml
new file mode 100644
index 000000000..e69de29bb
diff --git a/testing/src/main/resources/META-INF/services/io.quarkus.test.junit.callback.QuarkusTestBeforeEachCallback b/testing/src/main/resources/META-INF/services/io.quarkus.test.junit.callback.QuarkusTestBeforeEachCallback
new file mode 100644
index 000000000..b5643f866
--- /dev/null
+++ b/testing/src/main/resources/META-INF/services/io.quarkus.test.junit.callback.QuarkusTestBeforeEachCallback
@@ -0,0 +1 @@
+io.quarkiverse.flow.testing.WorkflowEventStoreResetExtension
\ No newline at end of file
diff --git a/testing/src/test/java/io/quarkiverse/flow/testing/EventWaiterTest.java b/testing/src/test/java/io/quarkiverse/flow/testing/EventWaiterTest.java
new file mode 100644
index 000000000..a1f7e30ee
--- /dev/null
+++ b/testing/src/test/java/io/quarkiverse/flow/testing/EventWaiterTest.java
@@ -0,0 +1,201 @@
+package io.quarkiverse.flow.testing;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+
+import io.quarkiverse.flow.testing.assertions.AsyncFlowAssertions;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.fluent.func.dsl.FuncDSL;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+
+public class EventWaiterTest {
+
+ @Test
+ void should_wait_for_workflow_started() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ CompletableFuture.runAsync(workflowInstance::start);
+
+ AsyncFlowAssertions.assertWith(store)
+ .workflowStarted()
+ .andAssert()
+ .workflowStarted();
+ }
+ }
+
+ @Test
+ void should_wait_for_task_started() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("myTask", (number) -> {
+ try {
+ Thread.sleep(100); // simulate some work
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return number + 1;
+ }, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ CompletableFuture.runAsync(workflowInstance::start);
+
+ AsyncFlowAssertions.assertWith(store).taskStarted("myTask");
+ }
+ }
+
+ @Test
+ void should_wait_for_task_completed() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ CompletableFuture.runAsync(workflowInstance::start);
+
+ AsyncFlowAssertions.assertWith(store)
+ .taskCompleted("task1");
+ }
+ }
+
+ @Test
+ void should_wait_for_workflow_completed() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for workflow to complete using new API
+ AsyncFlowAssertions.assertWith(store)
+ .workflowCompleted()
+ // .andAssert()
+ .workflowCompleted();
+
+ // Verify workflow completed
+ assertThat(store.getAll()).isNotEmpty();
+ }
+ }
+
+ @Test
+ void should_wait_for_multiple_events_in_sequence() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for events in sequence using new API
+ AsyncFlowAssertions.assertWith(store)
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+
+ // Verify all events were recorded
+ assertThat(store.getAll()).hasSizeGreaterThanOrEqualTo(6);
+ }
+ }
+
+ @Test
+ void should_handle_custom_timeout() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("slowTask", (number) -> {
+ try {
+ Thread.sleep(5000); // Very slow task
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return number + 1;
+ }, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow on a dedicated thread so interrupt propagates to Thread.sleep inside the task
+ Thread slowThread = new Thread(() -> workflowInstance.start().join());
+ slowThread.setDaemon(true);
+ slowThread.start();
+
+ // This should timeout because the task takes 5 seconds but we only wait 100ms
+ assertThatThrownBy(() -> AsyncFlowAssertions.assertWith(store)
+ .timeout(Duration.ofMillis(100))
+ .taskCompleted("slowTask"))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Timeout");
+
+ // Interrupt the slow thread so it does not bleed into subsequent tests
+ slowThread.interrupt();
+ }
+ }
+}
diff --git a/testing/src/test/java/io/quarkiverse/flow/testing/FlowAssertionsTest.java b/testing/src/test/java/io/quarkiverse/flow/testing/FlowAssertionsTest.java
new file mode 100644
index 000000000..2ee5cfc8a
--- /dev/null
+++ b/testing/src/test/java/io/quarkiverse/flow/testing/FlowAssertionsTest.java
@@ -0,0 +1,663 @@
+package io.quarkiverse.flow.testing;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.lang3.exception.UncheckedInterruptedException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import io.quarkiverse.flow.testing.assertions.AsyncFlowAssertions;
+import io.quarkiverse.flow.testing.assertions.ConfigurableAssertions;
+import io.quarkiverse.flow.testing.assertions.FlowAssertions;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.fluent.func.dsl.FuncDSL;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowError;
+import io.serverlessworkflow.impl.WorkflowException;
+import io.serverlessworkflow.impl.WorkflowInstance;
+
+public class FlowAssertionsTest {
+
+ private static final String VALIDATE_MARVEL_VILLAINS = "validateVillain";
+ private static final String DR_OCTOPUS = "Dr. Octopus";
+ private static final String BRUTUS = "Brutus";
+ private static final String HULK_FRIEND = "Hulk";
+ private static final String FLASH = "Flash";
+
+ private final Workflow getVillainWorkflow = FuncWorkflowBuilder.workflow("spider", "man", "2.0.0")
+ .tasks(FuncDSL.function(VALIDATE_MARVEL_VILLAINS, input -> {
+
+ if (BRUTUS.equals(input)) {
+ throw new IllegalStateException("Brutus is from Popeye not from Marvel");
+ }
+
+ if (HULK_FRIEND.equals(input)) {
+ return Boolean.FALSE;
+ }
+
+ if (FLASH.equals(input)) {
+ try {
+ Thread.sleep(500);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new UncheckedInterruptedException(e);
+ }
+ }
+
+ return Boolean.TRUE;
+ }, String.class)).build();
+
+ @Test
+ @DisplayName("Workflow should start and complete successfully")
+ void test_workflow_started_and_completed() {
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ @DisplayName("Asserting workflowFailed() on a successfully completed workflow should throw AssertionError")
+ void test_workflow_failed_assertion_throws_when_workflow_completed_successfully() {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+
+ Assertions.assertThrows(AssertionError.class, () -> {
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .workflowFailed(); // fails because the workflow completed successfully
+ });
+ }
+ }
+
+ @Test
+ @DisplayName("Workflow and the validateVillain task should start and complete successfully")
+ void test_workflow_and_task_started_and_completed() {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .workflowCompleted()
+ .taskStarted(VALIDATE_MARVEL_VILLAINS)
+ .taskCompleted(VALIDATE_MARVEL_VILLAINS);
+ }
+ }
+
+ @Test
+ @DisplayName("Workflow and the validateVillain task should fail")
+ void test_workflow_and_the_validate_villain_task_should_fail() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance("Brutus");
+
+ CompletableFuture.runAsync(instance::start);
+
+ Thread.sleep(200); // wait for the workflow to process the task and fail
+
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .taskFailed(VALIDATE_MARVEL_VILLAINS)
+ .workflowFailed();
+ }
+ }
+
+ @Test
+ @DisplayName("Asserting taskFailed() on a successfully completed workflow should throw AssertionError")
+ void test_task_failed_assertion_throws_when_workflow_completed_successfully() {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .taskFailed(VALIDATE_MARVEL_VILLAINS));
+ }
+ }
+
+ @Test
+ @DisplayName("Should have 1 WORKFLOW_STARTED event")
+ void should_have_one_workflow_started_event() {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+
+ FlowAssertions.assertWith(store)
+ .hasWorkflowStartedEventCount(1);
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowStartedEventCount(10));
+ }
+ }
+
+ @Test
+ @DisplayName("Should have 1 WORKFLOW_COMPLETED event")
+ void should_have_one_workflow_completed_event() {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(DR_OCTOPUS);
+ instance.start().join();
+
+ FlowAssertions.assertWith(store)
+ .hasWorkflowCompletedEventCount(1);
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowCompletedEventCount(10));
+ }
+ }
+
+ @Test
+ @DisplayName("Should have 1 WORKFLOW_FAILED event")
+ void should_have_one_workflow_failed_event() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(BRUTUS);
+
+ CompletableFuture.runAsync(instance::start);
+
+ Thread.sleep(200); // wait for the workflow to process the task and fail
+
+ FlowAssertions.assertWith(store)
+ .hasWorkflowFailedEventCount(1);
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowFailedEventCount(10));
+ }
+ }
+
+ @Test
+ @DisplayName("Should have 1 WORKFLOW_CANCELLED event")
+ void should_have_one_workflow_cancelled_event() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(FLASH);
+
+ CompletableFuture.runAsync(instance::start);
+
+ Thread.sleep(200); // wait for the workflow to enter the Flash task's 1-second sleep
+
+ instance.cancel();
+
+ Thread.sleep(200); // wait for the cancellation event to be recorded
+
+ FlowAssertions.assertWith(store)
+ .hasWorkflowCanceledEventCount(1);
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowCanceledEventCount(10));
+ }
+ }
+
+ @Test
+ @DisplayName("Should have 1 WORKFLOW_SUSPENDED and WORKFLOW_RESUMED event")
+ void should_have_one_workflow_suspended_and_resumed_event() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(FLASH);
+
+ CompletableFuture.runAsync(instance::start);
+
+ Thread.sleep(200);
+
+ instance.suspend();
+
+ instance.resume();
+
+ Thread.sleep(200);
+
+ FlowAssertions.assertWith(store)
+ .hasWorkflowSuspendedEventCount(1)
+ .hasWorkflowResumedEventCount(1);
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowSuspendedEventCount(10));
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .hasWorkflowResumedEventCount(10));
+ }
+ }
+
+ @Test
+ @DisplayName("workflowCancelled() asserts a WORKFLOW_CANCELLED event exists")
+ void should_assert_workflow_cancelled_event() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(FLASH);
+
+ CompletableFuture.runAsync(instance::start);
+ Thread.sleep(200);
+ instance.cancel();
+ Thread.sleep(200);
+
+ FlowAssertions.assertWith(store)
+ .workflowCancelled();
+
+ Assertions.assertThrows(AssertionError.class,
+ () -> FlowAssertions.assertWith(WorkflowEventStore.createInstance())
+ .workflowCancelled());
+ }
+ }
+
+ @Test
+ @DisplayName("workflowSuspended() and workflowResumed() assert the corresponding events exist")
+ void should_assert_workflow_suspended_and_resumed_events() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(FLASH);
+
+ CompletableFuture.runAsync(instance::start);
+ Thread.sleep(200);
+ instance.suspend();
+ Thread.sleep(100);
+ instance.resume();
+ Thread.sleep(300);
+
+ FlowAssertions.assertWith(store)
+ .workflowSuspended()
+ .workflowResumed();
+ }
+ }
+
+ @Test
+ @DisplayName("taskCancelled() asserts a TASK_CANCELLED event exists for the given task")
+ void should_assert_task_cancelled_event() throws InterruptedException {
+ // A listen task registers itself as a cancelable CompletableFuture,
+ // so cancelling the workflow properly fires TASK_CANCELLED (not TASK_FAILED).
+ Workflow listenWorkflow = FuncWorkflowBuilder.workflow("listen", "cancel", "1.0.0")
+ .tasks(
+ FuncDSL.listen("waitingForEvent", FuncDSL.toOne("never-comes-event")))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(listenWorkflow);
+ WorkflowInstance instance = def.instance("input");
+
+ CompletableFuture.runAsync(instance::start);
+ Thread.sleep(200);
+ instance.cancel();
+ Thread.sleep(200);
+
+ FlowAssertions.assertWith(store)
+ .taskCancelled("waitingForEvent");
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .taskCancelled("nonExistentTask"));
+ }
+ }
+
+ @Test
+ @DisplayName("taskSuspended() and taskResumed() throw AssertionError because the SDK does not fire task-level suspend/resume events")
+ void task_suspended_and_resumed_assertions_throw_when_no_task_events_fired() throws InterruptedException {
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ WorkflowDefinition def = app.workflowDefinition(getVillainWorkflow);
+ WorkflowInstance instance = def.instance(FLASH);
+
+ CompletableFuture.runAsync(instance::start);
+ Thread.sleep(200);
+ instance.suspend();
+ Thread.sleep(100);
+ instance.resume();
+ Thread.sleep(300);
+
+ // Workflow-level suspend/resume events ARE fired
+ FlowAssertions.assertWith(store)
+ .workflowSuspended()
+ .workflowResumed();
+
+ // Task-level suspend/resume events are NOT fired by the current SDK implementation
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .taskSuspended(VALIDATE_MARVEL_VILLAINS));
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .taskResumed(VALIDATE_MARVEL_VILLAINS));
+ }
+ }
+
+ @Test
+ @DisplayName("taskRetried() asserts a TASK_RETRIED event exists for the given task")
+ void should_assert_task_retried_event() {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ Workflow retryWorkflow = FuncWorkflowBuilder.workflow("retry", "test", "1.0.0")
+ .tasks(
+ FuncDSL.tryCatch("retryBlock", tryTask -> tryTask
+ .tryHandler(FuncDSL.function("retriableTask", input -> {
+ if (attempts.getAndIncrement() < 2) {
+ throw new WorkflowException(
+ WorkflowError.error("https://example.com/error/transient", 503).build());
+ }
+ return "success";
+ }, String.class))
+ .catchHandler(catchBlock -> catchBlock
+ .retry(retry -> retry
+ .backoff(b -> {
+ })
+ .delay(d -> d.milliseconds(10))
+ .limit(limit -> limit
+ .attempt(attempt -> attempt.count(3)))))))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+ try (WorkflowApplication app = buildWorkflowAppWith(store)) {
+ app.workflowDefinition(retryWorkflow)
+ .instance("go")
+ .start()
+ .join();
+
+ FlowAssertions.assertWith(store)
+ .taskRetried("retriableTask");
+
+ Assertions.assertThrows(AssertionError.class, () -> FlowAssertions.assertWith(store)
+ .taskRetried("nonExistentTask"));
+ }
+ }
+
+ @Test
+ @DisplayName("Should use ordered assertions with strictly() mode")
+ void should_use_ordered_assertions_with_inOrder() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore workflowEventStore = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(workflowEventStore))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(5L);
+ workflowInstance.start().join();
+
+ // Strict ordering - events must occur in exact sequence
+ FlowAssertions.assertWith(workflowEventStore)
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ @DisplayName("Should use unordered assertions by default")
+ void should_use_unordered_assertions_by_default() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("taskA", (number) -> number + 1, Long.class),
+ FuncDSL.function("taskB", (number) -> number * 2, Long.class),
+ FuncDSL.function("taskC", (number) -> number - 3, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // No inOrder() - just verify events exist, order doesn't matter
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .taskStarted("taskC") // Can check in any order
+ .taskStarted("taskA")
+ .taskCompleted("taskB")
+ .taskStarted("taskB")
+ .taskCompleted("taskA")
+ .taskCompleted("taskC")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ @DisplayName("Should count specific event types")
+ void should_count_specific_event_types() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class),
+ FuncDSL.function("task3", (number) -> number - 5, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // Count specific event types
+ FlowAssertions.assertWith(store)
+ .workflowStarted() // Must call assertion first to get FluentEventAssertions
+ .hasWorkflowStartedEventCount(1)
+ .hasWorkflowCompletedEventCount(1)
+ .hasTaskStartedEventCount(3)
+ .hasTaskCompletedEventCount(3);
+ }
+ }
+
+ @Test
+ @DisplayName("Should verify task execution order")
+ void should_verify_task_execution_order() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("firstTask", (number) -> number + 1, Long.class),
+ FuncDSL.function("secondTask", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // Verify one task completed before another
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .assertTask("firstTask").completedBeforeOrEqualTo("secondTask");
+ }
+ }
+
+ @Test
+ @DisplayName("Should verify workflow completion time")
+ void should_verify_workflow_completion_time() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("quickTask", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // Verify workflow completed within a time limit
+ FlowAssertions.assertWith(store)
+ .workflowStarted() // Must call assertion first
+ .workflowCompletedWithin(Duration.ofSeconds(5));
+ }
+ }
+
+ @Test
+ @DisplayName("Should verify all events for specific instance")
+ void should_verify_all_events_for_specific_instance() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // Verify all events belong to the same instance
+ FlowAssertions.assertWith(store)
+ .workflowStarted() // Must call assertion first
+ .allEventsForInstance(workflowInstance.id());
+ }
+ }
+
+ @Test
+ @DisplayName("Should verify output of completed workflow")
+ void should_verify_output_of_completed_workflow() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("doubleIt", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore workflowEventStore = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(workflowEventStore))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(5L);
+ workflowInstance.start().join();
+
+ // Verify workflow output
+ FlowAssertions.assertWith(workflowEventStore)
+ .strictly()
+ .workflowStarted()
+ .taskStarted("doubleIt")
+ .taskCompleted("doubleIt")
+ .workflowCompleted()
+ .withOutput(output -> {
+ assertThat(output.asNumber().orElseThrow()).isEqualTo(10L);
+ });
+ }
+ }
+
+ @Test
+ @DisplayName("Should reset and reuse assertions")
+ void should_reset_and_reuse_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore workflowEventStore = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(workflowEventStore))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ ConfigurableAssertions configurableAssertions = FlowAssertions.assertWith(workflowEventStore);
+
+ // First pass - ordered assertions
+ configurableAssertions.strictly()
+ .workflowStarted()
+ .taskStarted("task1");
+
+ // Reset and verify again from the beginning
+ configurableAssertions.reset()
+ .strictly()
+ .reset()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ @DisplayName("Should combine event waiter with fluent assertions")
+ void should_combine_event_waiter_with_fluent_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore workflowEventStore = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(workflowEventStore))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for specific events and verify the complete sequence in one chain
+ AsyncFlowAssertions.assertWith(workflowEventStore)
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .workflowCompleted()
+ .andAssert()
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+ }
+ }
+
+ private static WorkflowApplication buildWorkflowAppWith(WorkflowEventStore eventStore) {
+ return WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(eventStore))
+ .build();
+ }
+
+}
diff --git a/testing/src/test/java/io/quarkiverse/flow/testing/InstanceIdFilteringTest.java b/testing/src/test/java/io/quarkiverse/flow/testing/InstanceIdFilteringTest.java
new file mode 100644
index 000000000..6fc355cf0
--- /dev/null
+++ b/testing/src/test/java/io/quarkiverse/flow/testing/InstanceIdFilteringTest.java
@@ -0,0 +1,160 @@
+package io.quarkiverse.flow.testing;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+
+import io.quarkiverse.flow.testing.assertions.AsyncFlowAssertions;
+import io.quarkiverse.flow.testing.assertions.FlowAssertions;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.fluent.func.dsl.FuncDSL;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+
+/**
+ * Tests for filtering workflow events by instance ID.
+ */
+public class InstanceIdFilteringTest {
+
+ @Test
+ void should_filter_events_by_instance_id_in_fluent_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("inc", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ // Start two workflow instances
+ WorkflowInstance instance1 = def.instance(10L);
+ WorkflowInstance instance2 = def.instance(20L);
+
+ instance1.start().join();
+ instance2.start().join();
+
+ // Verify we have events from both instances
+ assertThat(store.size()).isGreaterThan(0);
+
+ // Filter and assert on instance1
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .strictly()
+ .workflowStarted()
+ .taskStarted("inc")
+ .taskCompleted("inc")
+ .workflowCompleted();
+
+ // Filter and assert on instance2
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .workflowStarted()
+ .taskStarted("inc")
+ .taskCompleted("inc")
+ .workflowCompleted();
+
+ // Verify each instance has exactly 4 events (started, task started, task completed, completed)
+ assertThat(store.filterByInstanceId(instance1.id())).hasSize(4);
+ assertThat(store.filterByInstanceId(instance2.id())).hasSize(4);
+ }
+ }
+
+ @Test
+ void should_filter_events_by_instance_id_in_async_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("inc", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ // Start two workflow instances asynchronously
+ WorkflowInstance instance1 = def.instance(10L);
+ WorkflowInstance instance2 = def.instance(20L);
+
+ CompletableFuture.runAsync(() -> instance1.start().join());
+ CompletableFuture.runAsync(() -> instance2.start().join());
+
+ // Wait for instance1 completion and assert
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .workflowCompleted()
+ .configure()
+ .filteringBy(instance1.id())
+ .workflowStarted()
+ .taskCompleted("inc")
+ .workflowCompleted();
+
+ // Wait for instance2 completion and assert
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .workflowCompleted()
+ .configure()
+ .filteringBy(instance2.id())
+ .workflowStarted()
+ .taskCompleted("inc")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ void should_filter_events_with_ordered_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (n) -> n + 1, Long.class),
+ FuncDSL.function("task2", (n) -> n * 2, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ WorkflowInstance instance1 = def.instance(5L);
+ WorkflowInstance instance2 = def.instance(10L);
+
+ instance1.start().join();
+ instance2.start().join();
+
+ // Verify ordered execution for instance1
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+
+ // Verify ordered execution for instance2
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+ }
+ }
+}
diff --git a/testing/src/test/java/io/quarkiverse/flow/testing/IntegratedWaitAndAssertTest.java b/testing/src/test/java/io/quarkiverse/flow/testing/IntegratedWaitAndAssertTest.java
new file mode 100644
index 000000000..594152f25
--- /dev/null
+++ b/testing/src/test/java/io/quarkiverse/flow/testing/IntegratedWaitAndAssertTest.java
@@ -0,0 +1,259 @@
+package io.quarkiverse.flow.testing;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.quarkiverse.flow.testing.assertions.AsyncFlowAssertions;
+import io.quarkiverse.flow.testing.assertions.FlowAssertions;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.fluent.func.dsl.FuncDSL;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+
+/**
+ * Test demonstrating the new AsyncFluentEventAssertions API for waiting and asserting.
+ * Shows the separation of concerns: AsyncFluentEventAssertions for waiting,
+ * FluentEventAssertions for asserting.
+ */
+public class IntegratedWaitAndAssertTest {
+
+ @Test
+ void should_wait_and_assert_in_single_chain() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ return number + 1;
+ }, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ workflowInstance.start().thenAccept(Assertions::assertNotNull);
+
+ // Wait for events using AsyncFluentEventAssertions, then assert
+ AsyncFlowAssertions.assertWith(store)
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted()
+ .andAssert()
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ void should_wait_with_custom_timeout() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("quickTask", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Configure timeout and wait
+ AsyncFlowAssertions.assertWith(store)
+ .timeout(Duration.ofSeconds(10))
+ .workflowCompleted()
+ // .andAssert()
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ void should_wait_for_specific_task_completion() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class),
+ FuncDSL.function("task2", (number) -> number * 2, Long.class),
+ FuncDSL.function("task3", (number) -> number - 5, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for specific task and verify
+ AsyncFlowAssertions.assertWith(store)
+ .taskCompleted("task2")
+ // .andAssert()
+ .taskCompleted("task2");
+ }
+ }
+
+ @Test
+ void should_combine_wait_and_unordered_assertions() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("taskA", (number) -> number + 1, Long.class),
+ FuncDSL.function("taskB", (number) -> number * 2, Long.class),
+ FuncDSL.function("taskC", (number) -> number - 3, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for completion, then verify all tasks ran (order doesn't matter)
+ AsyncFlowAssertions.assertWith(store)
+ .workflowCompleted()
+ // .andAssert()
+ .taskStarted("taskA")
+ .taskStarted("taskB")
+ .taskStarted("taskC")
+ .taskCompleted("taskA")
+ .taskCompleted("taskB")
+ .taskCompleted("taskC");
+ }
+ }
+
+ @Test
+ void should_wait_and_verify_output() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("multiply", (number) -> number * 3, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(5L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Wait for completion and verify output
+ AsyncFlowAssertions.assertWith(store)
+ .timeout(Duration.ofSeconds(5))
+ .workflowCompleted()
+ .andAssert()
+ .strictly()
+ .workflowStarted()
+ .taskStarted("multiply")
+ .taskCompleted("multiply")
+ .workflowCompleted()
+ .withOutput(output -> {
+ assertThat(output.asNumber()).hasValue(15L);
+ });
+ }
+ }
+
+ @Test
+ void should_use_both_creation_methods() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+ workflowInstance.start().join();
+
+ // Method 1: Create with WorkflowEventStore (standard assertions)
+ FlowAssertions.assertWith(store)
+ .workflowStarted()
+ .taskStarted("task1");
+
+ // Method 2: Create with events list (no waiting support)
+ FlowAssertions.assertWith(store)
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ void should_use_polling_mode_explicitly() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance(); // Use shared storage for async
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+ WorkflowInstance workflowInstance = def.instance(10L);
+
+ // Start workflow asynchronously
+ CompletableFuture.runAsync(() -> workflowInstance.start().join());
+
+ // Use polling mode explicitly with custom poll interval
+ AsyncFlowAssertions.assertWith(store)
+ .pollInterval(Duration.ofMillis(10))
+ .timeout(Duration.ofSeconds(5))
+ .workflowCompleted()
+ // .andAssert()
+ .workflowCompleted();
+ }
+ }
+}
diff --git a/testing/src/test/java/io/quarkiverse/flow/testing/SequentialInstanceWaitTest.java b/testing/src/test/java/io/quarkiverse/flow/testing/SequentialInstanceWaitTest.java
new file mode 100644
index 000000000..79bce2351
--- /dev/null
+++ b/testing/src/test/java/io/quarkiverse/flow/testing/SequentialInstanceWaitTest.java
@@ -0,0 +1,190 @@
+package io.quarkiverse.flow.testing;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.junit.jupiter.api.Test;
+
+import io.quarkiverse.flow.testing.assertions.AsyncFlowAssertions;
+import io.quarkiverse.flow.testing.assertions.FlowAssertions;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.fluent.func.FuncWorkflowBuilder;
+import io.serverlessworkflow.fluent.func.dsl.FuncDSL;
+import io.serverlessworkflow.impl.WorkflowApplication;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+
+/**
+ * Tests for sequentially waiting for events from different workflow instances.
+ * Demonstrates using waitFor() for one instance, then another instance.
+ */
+public class SequentialInstanceWaitTest {
+
+ @Test
+ void should_wait_for_first_instance_then_second_instance() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("inc", (number) -> number + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ // Start two workflow instances asynchronously
+ WorkflowInstance instance1 = def.instance(10L);
+ WorkflowInstance instance2 = def.instance(20L);
+
+ CompletableFuture.runAsync(() -> instance1.start().join());
+ CompletableFuture.runAsync(() -> instance2.start().join());
+
+ // Wait for instance1 to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .workflowCompleted()
+ .configure()
+ .filteringBy(instance1.id())
+ .workflowStarted()
+ .taskCompleted("inc")
+ .workflowCompleted();
+
+ // Then wait for instance2 to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .workflowCompleted()
+ .configure()
+ .filteringBy(instance2.id())
+ .workflowStarted()
+ .taskCompleted("inc")
+ .workflowCompleted();
+
+ // Verify both instances completed
+ assertThat(store.filterByInstanceId(instance1.id())).isNotEmpty();
+ assertThat(store.filterByInstanceId(instance2.id())).isNotEmpty();
+ }
+ }
+
+ @Test
+ void should_wait_for_multiple_instances_with_different_tasks() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("task1", (n) -> n + 1, Long.class),
+ FuncDSL.function("task2", (n) -> n * 2, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ WorkflowInstance instance1 = def.instance(5L);
+ WorkflowInstance b = def.instance(10L);
+ WorkflowInstance instance3 = def.instance(15L);
+
+ // Start all instances asynchronously
+ CompletableFuture.runAsync(() -> instance1.start().join());
+ CompletableFuture.runAsync(() -> b.start().join());
+ CompletableFuture.runAsync(() -> instance3.start().join());
+
+ // Wait for instance1 task1 completion
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .taskCompleted("task1")
+ .configure()
+ .filteringBy(instance1.id())
+ .taskStarted("task1")
+ .taskCompleted("task1");
+
+ // Wait for b task2 completion
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(b.id())
+ .taskCompleted("task2")
+ .configure()
+ .filteringBy(b.id())
+ .taskCompleted("task1")
+ .taskCompleted("task2");
+
+ // Wait for instance3 workflow completion
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance3.id())
+ .workflowCompleted()
+ .configure()
+ .reset()
+ .strictly()
+ .workflowStarted()
+ .taskStarted("task1")
+ .taskCompleted("task1")
+ .taskStarted("task2")
+ .taskCompleted("task2")
+ .workflowCompleted();
+ }
+ }
+
+ @Test
+ void should_wait_for_specific_events_across_multiple_instances() {
+ Workflow workflow = FuncWorkflowBuilder.workflow()
+ .tasks(
+ FuncDSL.function("process", (n) -> n + 1, Long.class))
+ .build();
+
+ WorkflowEventStore store = WorkflowEventStore.createInstance();
+
+ try (WorkflowApplication app = WorkflowApplication.builder()
+ .withListener(new TestWorkflowExecutionListener(store))
+ .build()) {
+
+ WorkflowDefinition def = app.workflowDefinition(workflow);
+
+ WorkflowInstance instance1 = def.instance(100L);
+ WorkflowInstance instance2 = def.instance(200L);
+
+ CompletableFuture.runAsync(() -> instance1.start().join());
+ CompletableFuture.runAsync(() -> instance2.start().join());
+
+ // Wait for instance1 to start
+ AsyncFlowAssertions.assertWith(store).filteringBy(instance1.id())
+ .workflowStarted();
+
+ // Wait for instance2 to start
+ AsyncFlowAssertions.assertWith(store).filteringBy(instance2.id())
+ .workflowStarted();
+
+ // Wait for instance1 task to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .taskCompleted("process");
+
+ // Wait for instance2 task to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .taskCompleted("process");
+
+ // Wait for instance1 to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .workflowCompleted();
+
+ // Wait for instance2 to complete
+ AsyncFlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .workflowCompleted();
+
+ // Assert both completed successfully
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance1.id())
+ .workflowCompleted();
+
+ FlowAssertions.assertWith(store)
+ .filteringBy(instance2.id())
+ .workflowCompleted();
+ }
+ }
+}