kotlin - Testing Kotlin Flow.combine with unending flows

Question

I have the following test

@Test
fun combineUnendingFlows() = runBlockingTest {

    val source1 = flow {
        emit("a")
        emit("b")
        suspendCancellableCoroutine { /* Never complete. */ }
    }

    val source2 = flowOf(1, 2, 3)

    val combinations = mutableListOf<String>()

    combine(source1, source2) { first, second -> "$first$second" }
        .onEach(::println)
        .onEach(combinations::add)
        .launchIn(this)

    advanceUntilIdle()

    assertThat(combinations).containsExactly("a1", "b1", "b2", "b3")
}

The assertion succeeds, however the test fails with the exception:

kotlinx.coroutines.test.UncompletedCoroutinesError: Test finished with active jobs

I know this is contrived, and we could easily make this pass by ensuring that source1 completes, but I'm wondering why it fails? Is runBlockingTest the wrong approach for testing never-ending flows?

(This is with coroutines 1.4.0).

Answer 1

The crux of the issue appears to be that runBlockingTest (probably rightly), expects there to be no running jobs by the time it goes to finish.

This would be fine if we could explicitly cancel the scope provided by runBlockingTest, however attempting to call .cancel() on that scope throws an exception.

There's more discussion on the problem in this GitHub issue.

In this contrived example, it would be simple enough get a reference to the Job created by launchIn() and cancel it before the test closes.

More elaborate scenarios could either catch UncompletedCoroutinesError and ignore it since they expect there to be unfinished coroutines, or they could create a new child CoroutineScope to explicitly cancel. In this example, that would look like

@Test
fun combineUnendingFlows() = runBlockingTest {
    val job = Job()
    val childScope = CoroutineScope(this.coroutineContext + job)

    val source1 = flow {
        emit("a")
        emit("b")
        suspendCancellableCoroutine { /* Never complete. */ }
    }

    val source2 = flowOf(1, 2, 3)

    val combinations = mutableListOf<String>()

    combine(source1, source2) { first, second -> "$first$second" }
        .onEach(::println)
        .onEach(combinations::add)
        .launchIn(childScope)

    advanceUntilIdle()
    assertThat(combinations).containsExactly("a1", "b1", "b2", "b3")
    job.cancel()
}