Test passes for 'runBlocking' but fails for 'runBlockingTest'

[MiSikora]

The test below passes when runBlocking is used but fails with runBlockingTest.

sealed class Event {
  data class A(val id: Long) : Event()
  data class B(val id: Long) : Event()
}

class Test {
  @Test fun blocking() = runBlocking {
    val events = flowOf(Event.A(0), Event.B(0)).broadcastIn(this).asFlow()

    val a = events.filterIsInstance<Event.A>().map { "$it" }
    val b = events.filterIsInstance<Event.B>().map { "$it" }

    val strings = flowOf(a, b).flattenMerge().toList()

    Assert.assertEquals(strings, listOf(Event.A(0), Event.B(0)).map { "$it" })
  }

  @Test fun blockingTest() = runBlockingTest {
    val events = flowOf(Event.A(0), Event.B(0)).broadcastIn(this).asFlow()

    val a = events.filterIsInstance<Event.A>().map { "$it" }
    val b = events.filterIsInstance<Event.B>().map { "$it" }

    val strings = flowOf(a, b).flattenMerge().toList()

    Assert.assertEquals(strings, listOf(Event.A(0), Event.B(0)).map { "$it" })
  }
}

Events are missing from the expected result.

java.lang.AssertionError: 
Expected :[A(id=0)]
Actual   :[A(id=0), B(id=0)]

Kotlin: 1.3.50.
Coroutines: 1.3.2.

[luis-cortes]

Should a test that passes using runBlocking always pass using runBlockingTest? I'm running into this and in order to make the runBlockingTest version work I have to add a lot more code.

[qwwdfsad]

Most of the time (except really long timeouts).
Current runTest is way too limited and being reworked right now

[Pozzoooo]

Hey, I'm facing the same issue on maybe a simpler scenario:

    @Test fun test1() {
        runBlockingTest {
            launch(Dispatchers.Default) {
                println("1")
                delay(1000L)
                println("2")
            }
            println("3")
        }
    }

    @Test fun test2() {
        runBlocking {
            launch(Dispatchers.Default) {
                println("1")
                delay(1000L)
                println("2")
            }
            println("3")
        }
    }

test1 fails with leaked coroutine, "3" and "1" is printed.
test2 passes with "3", "1" and "2" printed.

[MiSikora]

Another test case that fails for runBlockingTest but passes for runBlocking.

@Test fun test1() = runBlockingTest {
  val list = flowOf(1).onStart { emit(0) }
      .combine(flowOf("A")) { int, str -> "$str$int" }
      .toList()

  Assert.assertEquals(list, listOf("A0", "A1"))
}

Interestingly when onStart is used on the whole stream and not on parts of combine the test passes.

@Test fun test2() = runBlockingTest {
  val list = flowOf(1)
      .combine(flowOf("A")) { int, str -> "$str$int" }
      .onStart { emit("A0") }
      .toList()

  Assert.assertEquals(list, listOf("A0", "A1"))
}

[RBusarow]

Should a test that passes using runBlocking always pass using runBlockingTest?

I would say definitely not. runBlockingTest is not a catch-all function which can be applied to anything.

Code which uses delays will execute differently.
Code which launches but also continues code after the launch will execute differently.
Code which leaks coroutines (intentionally or not) will execute differently and fail. This is due to cleanupTestCoroutines().

[RBusarow]

I think we can more easily illustrate the disconnect here with a different look at this example:

@Test fun test1() = runBlockingTest {
  val list = flowOf(1).onStart { emit(0) }
      .combine(flowOf("A")) { int, str -> "$str$int" }
      .toList()

  Assert.assertEquals(list, listOf("A0", "A1"))
}

Let's reconstruct this a bit.

@Test fun test1() = runBlockingTest {

   val ints = flowOf(0, 1, 2, 3)
   val strs = flowOf("A", "B", "C")

   val list = combine(ints, strs) { int, str -> "$str$int" }
     .flowOn(TestCoroutineDispatcher())
     .toList()

   list shouldBe listOf("A0", "A1", "B1", "B2", "C2", "C3")
}

The result:

Expected :["A0", "A1", "B1", "B2", "C2", "C3"]
Actual   :["A3", "B3", "C3"]

When combine does its thing, both flows are converted into channelCoroutines and populated with sendFair(value).

The order of parameters matters here. The flow of Ints is processed completely before the flow of Strings is touched. This is because the TestCoroutineDispatcher in runBlockingTest { ... } is an immediate dispatcher. So the value of "3" is the latest, which is why it's the only one used by combine.

The same can be said for the tests using flattenMerge(). flattenMerge() uses launch(), which is different for TestCoroutineDispatcher than with a vanilla runBlocking { ... } or with Dispatchers.Unconfined.

In both cases, runBlockingTest looks like it's doing what it's supposed to.

Interestingly when onStart is used on the whole stream and not on parts of combine the test passes.

@Test fun test2() = runBlockingTest {
  val list = flowOf(1)
      .combine(flowOf("A")) { int, str -> "$str$int" }
      .onStart { emit("A0") }
      .toList()

  Assert.assertEquals(list, listOf("A0", "A1"))
}

This is because you're now emitting A0 to the flow before A1 is emitted. onStart { ... } is executed before the builder or combine.

Also, the order of your parameters in assertEquals() is backwards. It's expected, then actual. :)

[RBusarow]

    @Test fun test1() {
        runBlockingTest {
            launch(Dispatchers.Default) {
                println("1")
                delay(1000L)
                println("2")
            }
            println("3")
        }
    }

You're not actually using the dispatcher from runBlockingTest here. You're overriding it with Dispatchers.Default. If you were using the RBT dispatcher, it would automatically skip over that delay and the test would pass.

When overriding the dispatcher, you'd need to use .join() on the launch's Job to make sure execution had finished.

[Pozzoooo]

Hmmm, but how does that work for runBlocking?

If I need to have the same dispatcher, then should I encapsulate all Dispatchers to keep them under my control?
It looks like I'm gonna need to add a lot of dependencies for every case I need to use coroutines, my fear is to couple too much our code into the way coroutines works.

[RBusarow]

Hmmm, but how does that work for runBlocking?

If I need to have the same dispatcher, then should I encapsulate all Dispatchers to keep them under my control?
It looks like I'm gonna need to add a lot of dependencies for every case I need to use coroutines, my fear is to couple too much our code into the way coroutines works.

Yes, that's a common complaint. The typical solution is to wrap all references up into a single injectable object like so:

interface DispatcherProvider {
  val main: CoroutineDispatcher
  val mainImmediate: CoroutineDispatcher
  ...
}

A different solution is in the works for #1365 , where you'll be able to use Dispatchers.setDefault(...) to manipulate that singleton in lieu of all the injection, but now you're manipulating a shared state in your testing.

You can do things to mitigate this issue, such as only specifying the dispatcher when it's really necessary (for instance immediately before doing I/O in a repository). Typically, you shouldn't need to specify the dispatcher as the main thread is typically capable of much more than we give it credit for. Also, if you're using a third party library with coroutine support for I/O, such as Room or Retrofit, they already use their own dispatchers internally, so switching before calling a suspend function in their library is just wasteful.

Not to plug, but a final option would be my not-quite-complete little library DispatcherProvider, which uses the CoroutineContext itself to hold the interface described above. This way, the dispatchers are set upon creation of a CoroutineScope and are accessible from any coroutine or suspend function via the coroutineContext property.

[fblampe]

Are there any news or a timeline for this? Removing delays in tests via RunBlockingTest would be very helpful, but only if it works reliably without a lot of extra coding.