Coroutine Testing

Testing flows in your app can be confusing, especially when they’re interconnected with multiple data sources and emissions coming from multiple places. In this multi-part series we’ll cover how to effectively test flows and avoid flakiness in your coroutine testing. In the first part we’ll start by getting a better understanding of how Kotlin handles testing coroutines. Then in the second part we’ll look deeper into how Turbine works with different types of flows. Then we’ll look at how testing coroutines changes in KMP, and finally I’ll cover how this can be made into a skill for AIs to use when testing your code.

How Does Time Work?

First to best understand how to test let’s have a refresher on how tests actually operate. During tests there are two times, the real time and a virtual time. For example calling delay() in a flow doesn’t take real time, but Thread.sleep() does actually take real time (plus it blocks the thread). Similarly currentTimeMillis() uses the real time. With delay() the fake clock jumps ahead while the real clock stays the same.

With this in mind let’s look at the test example to get a good understanding of the different scheduling classes and how they work in a test.

class MyViewModel(dispatcher: CoroutineDispatcher = Dispatchers.Default) {
    private val scope = CoroutineScope(dispatcher)

    private val _state = MutableSharedFlow<String>()
    val state: SharedFlow<String> = _state.asSharedFlow()

    fun doThing(message: String) {
        scope.launch {
            delay(1000)
            _state.emit(message)
        }
    }

    fun close() {
        scope.cancel()
    }
}

@Test
fun myTest() = runTest {
    val dispatcher = StandardTestDispatcher(testScheduler)
    val viewModel = MyViewModel(dispatcher)

    viewModel.state.test {
        viewModel.doThing("hello")
        assertEquals("hello", awaitItem())
    }
}

Here you can see we’re doing a simple test for our ViewModel. This is what happens in the test under the hood:

First, runTest is called, which creates a testScheduler. The testScheduler uses the virtual clock, and schedules tasks based on that time. Then a CoroutineDispatcher, which puts work on the scheduler, is created and passed to the ViewModel (Another example of a dispatcher is Dispatchers.IO). From there the ViewModel uses that to create a Scope, which determines how long the work is alive for.

Additionally the testScheduler has a TimeSource which tells the code how to read the clock.

When doThing is called the state is updated in virtual time. Then when Turbine’s awaitItem() is called, Turbine suspends in real time until an event is available, or a timeout occurs. While the test body is suspended in awaitItem(), runTest runs the queued work and jumps the virtual clock ahead, so no advanceUntilIdle() is needed here. When the new state is emitted 1000ms of virtual time will have passed, and ~0ms of real time will have passed.

Included in the coroutine testing library are three functions to help drive the scheduler during the test:

• advanceUntilIdle - Runs all other coroutines on the scheduler until there is nothing left in the queue.
• advanceTimeBy - Advances virtual time by the given amount and runs any coroutines scheduled to run strictly before that point in virtual time. Meaning if you have a task that runs at 2 seconds and you advance time by exactly 2 seconds it will not run.
• runCurrent - Runs coroutines that are scheduled at the current virtual time.

Real threads are one of the biggest sources of flakiness in coroutine testing. In my tests, two flows updated from separate real threads and combined produced 49 different emission sequences across 300 runs. The testScheduler and StandardTestDispatcher are designed to use the virtual time, however if instead a ViewModel takes in a real CoroutineDispatcher then that work will be done in real time. This may work for simple tests such as this one with a short emission and Turbine awaiting, however for other tests this can cause issues.

Let’s go over some recommendations for when working with Dispatchers.

Dispatchers

A CoroutineDispatcher determines what thread or threads the corresponding coroutine uses for its execution. The coroutine dispatcher can confine coroutine execution to a specific thread, dispatch it to a thread pool, or let it run unconfined.

Recommendation 1: Use StandardTestDispatcher

For testing it’s recommended to use the StandardTestDispatcher because it makes timing deterministic. Nothing runs until you drive the scheduler using calls like runCurrent() and advanceUntilIdle(). The ordering of the emissions is also reproducible, which is helpful because Turbine collects unconfined. The collector runs eagerly on whatever thread emits, so it’s important to have the dispatcher and scheduler control the order of emissions. In my tests a ViewModel on a StandardTestDispatcher produced the same sequence in 500 out of 500 runs, and in 200 more runs under heavy CPU load.

Recommendation 2: Use the testScheduler

As mentioned the StandardTestDispatcher takes in a scheduler, which handles queuing of events and the virtual clock. Luckily TestScope includes a testScheduler variable that can be passed into the StandardTestDispatcher from within runTest. It’s important to use this in tests because a dispatcher created without it doesn’t share the test’s scheduler. If no scheduler is passed in, the dispatcher looks for one from Dispatchers.Main (when it has been set with setMain), and otherwise creates a brand-new scheduler. That scheduler isn’t the one runTest is using, so functions like advanceUntilIdle() never run its work and the test times out (in my test, a ViewModel built that way never ran at all). Sharing one scheduler also keeps all the work on a single virtual clock in one deterministic order.

Recommendation 3: Inject Dispatchers

It’s recommended to inject testing dispatchers into your classes for the best testing outcome.

class MyRepository(private val io: CoroutineDispatcher = Dispatchers.IO) : UserRepository {
    override suspend fun load(query: String) = withContext(io) { ... }
}

runTest {
    val dispatcher = StandardTestDispatcher(testScheduler)
    val vm = MyViewModel(MyRepository(dispatcher), ...)
    vm.search("x")
    advanceUntilIdle()
    assertEquals(UiState.Success("Ada"), vm.state.value)
}

By doing this you are asserting through your testing that the timing is controlled by the test and not by real threads and schedulers. Without the injected io the same test left state.value at Loading after advanceUntilIdle() in 300 out of 300 runs, because the scheduler can’t wait for a real dispatcher. With the dispatchers injected, the same test ran 300 times on each of 8 threads in parallel with 0 failures.

You can also do this through dependency injection like with Koin.

val appModule = module {
    single<CoroutineDispatcher>(named("io")) { Dispatchers.IO }
    viewModel { MyViewModel(get(named("io"))) }
}

@Test
fun myTest() = runTest {
    val testModule = module {
        single<CoroutineDispatcher>(named("io")) { StandardTestDispatcher(testScheduler) }
    }
    val koin = startKoin { modules(appModule, testModule) }.koin
    try {
        val viewModel: MyViewModel = koin.get()
        // ...
    } finally {
        stopKoin()
    }
}

Recommendation 4: Don’t use Dispatchers.setMain in parallel

If you need Dispatchers.Main in your code it’s an easy fix to just call Dispatchers.setMain in the @BeforeTest and call it a day, however this is not a good idea if you plan to run tests in parallel. This is because Dispatchers.Main is one global per process, so parallel tests overwrite each other’s main dispatcher and reset it from under each other. When I ran two tests in parallel that both used setMain, 241 out of 600 runs failed in one attempt and 598 out of 600 in another. It can corrupt Dispatchers.Main for the rest of the JVM so later, unrelated tests fail. Forgetting to reset it is a problem even without parallelism: the next runTest silently reuses the stale scheduler, and a queued task would run in the wrong test.

If you really do need to use Dispatchers.setMain be sure to call Dispatchers.resetMain after your test is done, and don’t run those tests in parallel.

val testDispatcher = UnconfinedTestDispatcher(testScheduler)
Dispatchers.setMain(testDispatcher)
try {
    ...
} finally {
    Dispatchers.resetMain()
}

Advanced Timing

A lot of coroutine testing involves launching a coroutine and asserting the result, but for more unique timing scenarios there are built-in options for handling certain scenarios.

Time Source

Sometimes you need to test flows that only emit after elapsed times, such as a cache expiring or token timeouts. For those instances it’s recommended to inject the scheduler’s TimeSource during testing to test long stretches of time. Then call advanceTimeBy to test that these interactions occur.

class ExpiringCache(private val timeSource: TimeSource, private val ttl: kotlin.time.Duration) {
    private var value: String? = null
    private var storedAt = timeSource.markNow()
    fun put(v: String) { value = v; storedAt = timeSource.markNow() }
    fun get(): String? = value?.takeIf { storedAt.elapsedNow() < ttl }
}

val cache = ExpiringCache(testScheduler.timeSource, 5.minutes)
cache.put("Ada")
advanceTimeBy(4.minutes)
assertEquals("Ada", cache.get())
advanceTimeBy(2.minutes)
assertNull(cache.get())

Here you can see in order to test that the cache clears after 5 minutes the test is calling advanceTimeBy to move the virtual clock. This ensures that no real time elapses, which means slower test runners or issues with timing will not cause flakiness with this test.

Background Scope

Sometimes you need a flow to run for the entire test even if you’re not testing it, such as a connection status. In these scenarios you’ll want to run this flow in the backgroundScope, either by passing in the scope or calling it directly in the test. The backgroundScope is specifically designed for background work. It is automatically cancelled when the test finishes, and a failure in one of its coroutines doesn’t terminate the test body (it’s reported at the end of the test).

backgroundScope.launch { events.collect { received += it } }
runCurrent()   // start the collector before emitting
events.emit(1)

// or 

class ChatRepository(private val scope: CoroutineScope, source: Flow<String>) {
    // SharingStarted.Eagerly will need runCurrent() before the flow starts
    val messages: StateFlow<String> = source.stateIn(scope, SharingStarted.Eagerly, "") 
}
// test
val repo = ChatRepository(backgroundScope, source)

Don’t use backgroundScope for work whose result the test must wait for; use a normal child or await it instead. advanceUntilIdle() skips backgroundScope work, so to settle it use runCurrent() or advanceTimeBy().

App Level Scope

Sometimes you need an app-level scope that owns fire-and-forget work, such as synchronization. In tests these components must receive a scope you control, otherwise they hard-code the Dispatchers and race the test. You can either pass a backgroundScope like above, or an owned scope built with a SupervisorJob(). The SupervisorJob ensures that a failing child doesn’t cancel its siblings or the rest of the scope.

val errors = Collections.synchronizedList(mutableListOf<Throwable>())
val scope = CoroutineScope(SupervisorJob() + testDispatcher + CoroutineExceptionHandler { _, e -> errors += e })
scope.launch { error("first request failed") }
advanceUntilIdle()
scope.launch { state.value = 1 } // This still runs even though the scope launched an error previously
advanceUntilIdle()
scope.cancel()

Your choice depends on what the work should be doing. If the work is finite and you want advanceUntilIdle() to run it to completion, use a SupervisorJob scope. If it’s an endless observer, or you want automatic cleanup, use backgroundScope and settle it with runCurrent() or advanceTimeBy().

Using Real Threads

Sometimes your tests need to cover real threads or dispatchers, such as when a library owns its own threads or when you need to reproduce a race condition. In these cases we can’t control the timing virtually, so we need to structure our tests around making the tests independent of timing. Here are some tips if you really need to test real threads:

1. Keep as much as possible virtual - First off, consider if you really need the real thread. If you do, run as much as you can on a StandardTestDispatcher(testScheduler) and let only the part that needs a real thread use a real dispatcher.

2. Keep real-thread tests few and separate - If you do need real-thread tests then consider running them separately as a separate Gradle task. Make sure not to combine them with tests that call Dispatchers.setMain.

3. Don’t use fixed delays or sleeps to synchronize - Using a fixed wait is a guessing game that creates flakiness (waiting a fixed 3 ms for a real 3 ms job was wrong in 4 to 14 out of 200 runs, while join() never failed). Instead you should wait for a signal, such as join(), a CompletableDeferred, or Turbine’s awaitItem().

4. Be generous with the timeouts - In the virtual clock timeouts don’t really matter, but with real threads make sure to add large timeouts for Turbine and the runTest itself. This is especially true if you plan to run in CI.

   runTest(timeout = 30.seconds) {
       viewModel.state.test(timeout = 20.seconds) {}
   }
   

   • Note: The default timeout length for Turbine is 3 seconds.

5. Cancel every real scope you create - Leaked scopes can keep running after a test and can affect states used in your other tests.

   val scope = CoroutineScope(Dispatchers.Default)
   try { /* test body */ } finally { scope.cancel() }
   

6. Don’t use advanceUntilIdle to wait for real work - It only moves the virtual clock, so it won’t wait for real work (with a repository on Dispatchers.Default, state.value was still Loading in 300 out of 300 runs). Turbine’s expectNoEvents() and expectMostRecentItem() don’t wait either, so they can pass or return stale data (expectNoEvents() passed 300 out of 300 times even though the event arrived right after, and expectMostRecentItem() returned a stale value in 299 of 300 runs). Wait for a positive signal first, like awaitItem() or join().

7. Don’t rely on currentTimeMillis - currentTimeMillis reads the real clock, which virtual time doesn’t move, and a real timestamp stored in state broke an equality check in 4 out of 200,000 comparisons. Inject a TimeSource instead (see Time Source above).

8. Lastly: Assert final values and patterns, not exact sequences - With a real thread you can’t reliably assert all of the emissions, and that they are emitted in the same order. You want to check that the final result is correct, and that the intermediate emissions satisfy properties that hold on every run no matter how many emissions arrive. For example a Loading is never followed by an older state, or a Success never reverts to Idle. In my tests, combining two real-thread sources gave 3 to 6 emissions across 49 different sequences in 300 runs, yet “neither value ever goes backwards” and the final value held in every one of 200 runs.

Cover Failures That Land After the Test Ends or Outside It

Be sure to handle errors that might happen outside of the test scope, such as a detached scope. Code that launches work in a detached scope, like CoroutineScope(Dispatchers.Default), isn’t a child of the test so the failure might not be caught. If the exception lands while runTest is running it fails the test, however if it lands after the test has finished it is lost, so you get a false pass (this passed 20 out of 20 times in my tests). Additionally in a full run a late exception can fail a different, unrelated test. To combat this, inject the scope and dispatcher so the work runs on the test scheduler, run it to completion inside the test (runCurrent(), advanceUntilIdle(), or job.join()), and consider capturing and asserting failures on purpose.

val errors = Collections.synchronizedList(mutableListOf<Throwable>())
val scope = CoroutineScope(SupervisorJob() + testDispatcher + CoroutineExceptionHandler { _, e -> errors += e })
syncer.sync(); advanceUntilIdle()
assertEquals("sync failed", errors.single().message)

Here we’re capturing all of the errors using a CoroutineExceptionHandler and asserting the results with a synchronized list.

Also cancel every scope you create by calling scope.cancel() (in finally or @After) so nothing outlives the test.

Don’t Share State Between Tests

Anything shared between tests makes the result depend on test order, which can cause flakiness during testing. For example a singleton StateFlow could have one test set the value to 5, and the next test expects its first item to be 0 but gets 5 instead. You should create fresh instances in every test (a new ViewModel, repository, scope, etc.), and avoid singletons in the code under test.

If results are collected from several coroutines, use a thread-safe collection (a plain ArrayList written to from four coroutines had the wrong size in 49 out of 50 trials).

val results = Collections.synchronizedList(mutableListOf<Int>())

Conclusion

In this post we covered best practices for testing coroutines in Kotlin. Hopefully this gave you some context into how tests run and gave some guidance on how to best run coroutines in tests. In the next post we’ll go more into depth with Turbine, going over how to handle the different kinds of flows and how to reduce flaky tests.