Add consumeItemsUntilPredicate to check how it goes...

features/messages/impl/src/test/kotlin/io/element/android/features/messages/MessagesPresenterTest.kt

@@ -65,6 +65,7 @@ import io.element.android.libraries.mediapickers.test.FakePickerProvider
 import io.element.android.libraries.mediaupload.api.MediaSender
 import io.element.android.libraries.mediaupload.test.FakeMediaPreProcessor
 import io.element.android.libraries.textcomposer.MessageComposerMode
+import io.element.android.tests.testutils.consumeItemsUntilPredicate
 import io.element.android.tests.testutils.consumeItemsUntilTimeout
 import io.element.android.tests.testutils.testCoroutineDispatchers
 import io.mockk.mockk
@@ -490,11 +491,14 @@ class MessagesPresenterTest {
         }.test {
             val initialState = consumeItemsUntilTimeout().last()
             initialState.eventSink(MessagesEvents.InviteDialogDismissed(InviteDialogAction.Invite))
-            val remainingStates = consumeItemsUntilTimeout()
-            assertThat(remainingStates.size).isEqualTo(3)
-            assertThat(remainingStates[0].inviteProgress.isLoading()).isFalse()
-            assertThat(remainingStates[1].inviteProgress.isLoading()).isTrue()
-            assertThat(remainingStates[2].inviteProgress.isFailure()).isTrue()
+            val loadingState = consumeItemsUntilPredicate { state ->
+                state.inviteProgress.isLoading()
+            }.last()
+            assertThat(loadingState.inviteProgress.isLoading()).isTrue()
+            val failureState = consumeItemsUntilPredicate { state ->
+                state.inviteProgress.isFailure()
+            }.last()
+            assertThat(failureState.inviteProgress.isFailure()).isTrue()
         }
     }
 

tests/testutils/src/main/kotlin/io/element/android/tests/testutils/ReceiveTurbine.kt

@@ -24,17 +24,32 @@ import kotlin.time.Duration
 import kotlin.time.Duration.Companion.milliseconds
 
 /**
- * Consume all items until timeout is reached waiting for an event.
+ * Consume all items until timeout is reached waiting for an event or we receive terminal event.
  * The timeout is applied for each event.
  * @return the list of consumed items.
  */
 suspend fun <T : Any> ReceiveTurbine<T>.consumeItemsUntilTimeout(timeout: Duration = 100.milliseconds): List<T> {
+    return consumeItemsUntilPredicate(timeout) { false }
+}
+
+/**
+ * Consume items until predicate is true, or timeout is reached waiting for an event, or we receive terminal event.
+ * The timeout is applied for each event.
+ * @return the list of consumed items.
+ */
+suspend fun <T : Any> ReceiveTurbine<T>.consumeItemsUntilPredicate(
+    timeout: Duration = 100.milliseconds,
+    predicate: (T) -> Boolean,
+): List<T> {
     val items = ArrayList<T>()
     tryOrNull {
         while (true) {
             when (val event = withTurbineTimeout(timeout) { awaitEvent() }) {
                 is Event.Item<T> -> {
                     items.add(event.value)
+                    if (predicate(event.value)) {
+                        break
+                    }
                 }
                 else -> break
             }