Is there something similar to RxJava `TestObserver` to test coroutines `Flow` ?

# coroutines

[tseisel]

Is there something similar to RxJava TestObserver to test coroutines Flow ?

[gildor]

https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-test/README.md

[Matej Drobnič]

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

[Matej Drobnič]

so basically answer is no?

[gildor]

What exactly about flow testing do you need?

[gildor]

For example things related to delay are there

[gildor]

there is no assert() extensions for example, but do you really need one? maybe you need some particular feature of TestObserver

[tseisel]

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 ?

[streetsofboston]

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)

[Paul Woitaschek]

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

[Paul Woitaschek]

I'd do the flow logic assertions using assert libraries:

runBlockingTest{
       val firstThreeItems = withTimeout(1000) {
         flowOf(1, 2, 3, 4).take(3).toList()
       }
        assertThat(firstThreeItems)...
      }

[jw]

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()
        }
  }

[streetsofboston]

@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.

[Vsevolod Tolstopyatov [JB]]

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)

[Marcelo Hernandez]

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

[Paul Woitaschek]

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