Testing Coroutines and Flows with kotlinx-coroutines-test

Testing suspending code with real wall-clock time is a recipe for flaky, slow tests. If a coroutine calls delay(5_000), you do not want your test to actually sit there for five seconds, and you certainly do not want timing races deciding whether an assertion passes. The kotlinx-coroutines-test artifact solves this with a virtual time scheduler: delays are skipped instantly while the relative ordering of events is preserved. Add it to your build with testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1"), matching the version of your coroutines-core dependency.

The entry point is the runTest { ... } builder. It runs your test body inside a TestScope backed by a TestDispatcher, drives the scheduler until all child coroutines complete, and re-throws any uncaught exceptions so failures surface as normal test failures. Crucially, runTest auto-advances virtual time: when every coroutine is suspended on a delay, the scheduler fast-forwards to the next scheduled resumption. That means a function that delays for an hour finishes in microseconds, yet code that depends on ordering still behaves correctly. Always use runTest instead of runBlocking in tests so you inherit this virtual-time machinery.

Two TestDispatcher implementations control how coroutines are scheduled. StandardTestDispatcher (the default inside runTest) queues new coroutines rather than running them eagerly, so they execute only when you yield control or explicitly advance the scheduler. UnconfinedTestDispatcher runs new coroutines eagerly until their first suspension point, which is convenient when you simply want launched work to start immediately. Pick StandardTestDispatcher when you need precise control over interleaving, and UnconfinedTestDispatcher when eager execution makes the test read more naturally. Both share the same TestCoroutineScheduler, which you can hoist out and pass to several dispatchers so they advance in lockstep.

When you do not want auto-advance to run everything to completion at once, you drive the clock manually. advanceTimeBy(duration) pushes virtual time forward by a fixed amount, resuming any coroutines whose delays have elapsed; runCurrent() then executes tasks scheduled for the now-current instant. advanceUntilIdle() runs everything until no further work remains. You can also read currentTime to assert how much virtual time a piece of work consumed. This lets you test debounce, timeout, retry-with-backoff, and polling logic precisely: schedule the work, advance time step by step, and assert the intermediate state at each tick.

A subtle but important rule: code under test must not hard-code Dispatchers.Default or Dispatchers.IO, because those real dispatchers ignore the test scheduler and break virtual time. Inject a CoroutineDispatcher (or CoroutineContext) into your classes so tests can substitute a TestDispatcher. As a fallback for code you cannot change, Dispatchers.setMain(testDispatcher) replaces the Main dispatcher, paired with Dispatchers.resetMain() in teardown. Reusing one scheduler across the injected dispatcher and the test keeps delays unified, so a delay inside a repository and a delay inside the test observe the same virtual clock.

Flow testing builds on the same foundation. The simplest and most reliable assertion is to collect a finite flow into a list with toList() and compare it to the expected sequence. Because toList() suspends until the flow completes, it naturally drains every emission, and inside runTest any delays between emissions are skipped. This works beautifully for flows built from flowOf, map, filter, or a flow { } block that emits a known number of items, letting you assert exact values and ordering in one line.

For flows with intermediate delays you combine toList() with manual time control, or you assert the timing of each emission. A classic example is debounce: feed values into the flow, advance virtual time past the debounce window, and verify that only the final value survives. Because the scheduler is deterministic, the test produces the same result on every run and every machine. When you need to inspect emissions as they happen rather than after completion, the Turbine library (app.cash.turbine) offers a test { awaitItem() } DSL, but for finite flows plain toList() keeps dependencies minimal and intent obvious.

Two pitfalls deserve a final word. First, an infinite or hot flow (such as a MutableStateFlow or a flow that never completes) will make toList() hang forever; bound it with take(n) before collecting, or use Turbine. Second, work launched in a separate scope or on an injected real dispatcher will not be awaited by runTest, leaving assertions racing against unfinished coroutines; route that work through the same TestScope or TestDispatcher so the scheduler can track it. Internally runTest enforces this by failing the test if coroutines are still running after a timeout, which is a helpful signal that a dispatcher leaked out of test control.

Put together, these tools turn notoriously timing-dependent asynchronous code into fast, deterministic unit tests. Inject dispatchers so production code is testable, wrap test bodies in runTest, advance the virtual clock to exercise time-based logic, and assert flow output with toList(). The result is a suite that runs in milliseconds regardless of how many simulated seconds, minutes, or hours your coroutines pretend to wait.

import kotlinx.coroutines.delay
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class GreeterTest {
    suspend fun greetAfterDelay(name: String): String {
        delay(10_000) // 10 seconds of virtual time, skipped instantly
        return "Hello, $name"
    }

    @Test
    fun returnsGreeting() = runTest {
        val result = greetAfterDelay("Kotlin")
        assertEquals("Hello, Kotlin", result)
        // currentTime is a Long reporting the virtual time consumed
        assertEquals(10_000L, currentTime)
    }
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class TimedTest {
    @Test
    fun advancesStepByStep() = runTest {
        val log = mutableListOf<String>()
        launch {
            delay(1_000); log += "one"
            delay(1_000); log += "two"
        }
        // Nothing has run yet with StandardTestDispatcher
        assertEquals(emptyList(), log)

        advanceTimeBy(1_001)   // past the first delay
        runCurrent()
        assertEquals(listOf("one"), log)

        advanceUntilIdle()     // run the rest
        assertEquals(listOf("one", "two"), log)
    }
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class FlowTest {
    fun numbers(): Flow<Int> = flow {
        for (i in 1..3) {
            delay(500) // skipped by virtual time
            emit(i)
        }
    }

    @Test
    fun emitsDoubledValues() = runTest {
        val result = numbers().map { it * 2 }.toList()
        assertEquals(listOf(2, 4, 6), result)
    }
}

import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.withContext
import org.springframework.stereotype.Service

@Service
class PriceService(
    private val repo: PriceRepository,
    private val dispatcher: CoroutineDispatcher // injected, not Dispatchers.IO
) {
    suspend fun latestPrice(id: Long): Int = withContext(dispatcher) {
        repo.findPrice(id)
    }
}

// Test: substitute a TestDispatcher backed by the test scheduler
// @Test fun price() = runTest {
//     val service = PriceService(fakeRepo, StandardTestDispatcher(testScheduler))
//     assertEquals(42, service.latestPrice(1))
// }