Is there something similar to RxJava

TestObserver

to test coroutines

Flow

?

I can't find anything on flow testing on that page

What exactly about flow testing do you need?

Since I'm dealing with infinite `Flow`s, I'd like to collect items one by one (the Rx equivalent would be

awaitCount(Int)

) and perform assertions on them. What are your recommendations for testing simpler flows ? Collect as

List

with

toList

and perform assertions on the list ?

Would be a great addition to the kotlinx-coroutines-test library:

fun <T> Flow<T>.test(): TestCollector<T>

and a new class

TestCollector<T>

. I'd open a PR or issue to add this to the library. (It would function like an Rx TestObserver/TestSubscriber)

Well I didnt find the rx test too helpful. What actually happening is the testobserver duplicating what assertion libraries already do

SQL Delight has some infrastructure for this: https://github.com/cashapp/sqldelight/blob/master/extensions/coroutines-extensions/src/commonTest/kotlin/com/squareup/sqldelight/runtime/coroutines/FlowAssert.kt Looks like:

@Test fun queryNotNotifiedAfterCancel() = runTest {
    db.createQuery(TABLE_EMPLOYEE, SELECT_EMPLOYEES, MAPPER)
        .asFlow()
        .test {
          expectItem().assert {
            hasRow("alice", "Alice Allison")
            hasRow("bob", "Bob Bobberson")
            hasRow("eve", "Eve Evenson")
          }

          cancel()

          db.employee(Employee("john", "John Johnson"))
          expectNoMoreEvents()
        }
  }

@Paul Woitaschek If the

flow

under test doesn’t complete, a terminating function on it never resumes. There is where something like a TestCollector or @jw’s example can help out.

It would be very nice if you could create an issue with a desired API shape and a bit of context (e.g. which particular problem this API helps to solve)

It's great to see https://github.com/cashapp/turbine just released. 👏

Already using that. Please incorporate that in the coroutines testing artifact 🙏