Testing Coroutines & Flow Explained
TESTING › Unit
Coroutine code leans on delay, launch, and dispatchers everywhere, and none of those play nicely with a naive test. Call a suspend function that does delay(5000) inside a normal test and you either block the thread for five real seconds or reach for Thread.sleep and pray. The kotlinx-coroutines-test library fixes this with runTest, a coroutine builder built specifically for tests. It runs your test body inside a TestScope backed by a TestCoroutineScheduler, which is a virtual clock rather than a real one. Delays inside that scope do not actually wait wall clock time, the scheduler just fast forwards past them, and runTest waits for everything launched in that scope to finish before returning. That's the property, virtual time, that turns coroutine tests from something that takes minutes to something that runs in milliseconds. By default, when you don't pass a dispatcher yourself, runTest installs a StandardTestDispatcher wired to that same scheduler.
@Test
fun computesTotal() = runTest {
val result = withDelay() // delay(5000) inside, but this test is instant
assertEquals(42, result)
}
That default, StandardTestDispatcher, does one important thing differently from what you might expect from ordinary launch: it queues newly launched coroutines rather than starting them immediately. If you call launch inside a test and assert right after, on StandardTestDispatcher that assertion still sees the pre-launch state, nothing has run yet. The coroutine sits on the scheduler until you explicitly drain it, most commonly with advanceUntilIdle, which we'll cover next. This is the single most common surprise in coroutine tests: a launch that looks synchronous when you read the code, but actually runs on a queue you have to pump yourself.
@Test
fun demonstratesQueuing() = runTest {
var saved = false
launch { saved = true } // queued, not run yet
assertFalse(saved)
advanceUntilIdle()
assertTrue(saved)
}
You drive that queue with a small set of tools. advanceUntilIdle runs everything queued, over and over, until the scheduler has nothing left to do, that's the blunt instrument you reach for by default. advanceTimeBy(n) is more surgical: it moves the virtual clock forward by n milliseconds and runs anything scheduled strictly before that new time. The word strictly matters. A coroutine that calls delay(1000) is scheduled to resume at exactly virtual time 1000, so advanceTimeBy(1000) moves the clock to 1000 but does not run that resume, because it was due at the boundary, not strictly before it. People trip over this constantly: they advance by exactly the delay's duration, expect the continuation to have run, and it hasn't. You need one more tick, or a call to runCurrent, to actually execute work scheduled at the instant you just arrived at.
launch {
delay(1_000)
done = true
}
advanceTimeBy(1_000) // runs work scheduled strictly BEFORE t=1000
assertFalse(done) // the resume AT exactly t=1000 hasn't run yet
runCurrent() // now it runs
assertTrue(done)
runCurrent is the third tool, and it's easy to conflate with the other two. It does not move the virtual clock at all, it just drains whatever is already scheduled to run at the current instant. That matters when a coroutine does some immediate work, then delays, then does more work: call runCurrent once and you get the immediate part, because it was already due, but the part after the delay stays pending, because that's scheduled for a future virtual time that runCurrent never advances to. If you want the whole thing to run, you need advanceTimeBy or advanceUntilIdle instead.
launch {
step1 = true // immediate
delay(10_000)
step2 = true // after delay
}
runCurrent() // runs tasks at t=0; clock stays at 0
assertTrue(step1)
assertFalse(step2) // still pending, delay hasn't elapsed
There's a second test dispatcher, UnconfinedTestDispatcher, and the difference is about eagerness. Where StandardTestDispatcher queues a newly launched coroutine and waits for you to pump the scheduler, UnconfinedTestDispatcher starts it immediately on the current thread and keeps running it until it hits its first suspension point, a delay, or a suspending call that actually suspends. Both dispatchers still share the same TestCoroutineScheduler underneath, so virtual time behaves identically once you're past that first suspension. UnconfinedTestDispatcher is handy for terser tests where you don't care about exact interleaving, but if the thing you're testing is precisely the order concurrent operations run in, its eagerness hides bugs that StandardTestDispatcher would expose.
// StandardTestDispatcher: must advance time manually
runTest(StandardTestDispatcher()) {
var result = 0
launch { result = 1 }
assertEquals(0, result) // not run yet
}
// UnconfinedTestDispatcher: runs eagerly to first suspension
runTest(UnconfinedTestDispatcher()) {
var result = 0
launch { result = 1 } // runs immediately
assertEquals(1, result)
}
Virtual time only helps if your production code is actually running on a dispatcher the test controls. A repository that hardcodes withContext(Dispatchers.IO) is running on a real thread pool, one with no connection to the TestCoroutineScheduler, so its timing is not deterministic and advancing virtual time does nothing to it. The fix is the same one you'd apply for any hard-to-test dependency: inject the dispatcher instead of hardcoding it, defaulting to Dispatchers.IO in production and passing a TestDispatcher tied to the shared scheduler in tests. This is exactly the kind of code an interviewer expects you to flag on sight, a hardcoded Dispatchers.IO or Dispatchers.Default inside a class under test is a testability smell before it's anything else.
// hardcoded: real IO thread pool bypasses the virtual-time scheduler
class BadRepo {
suspend fun fetch() = withContext(Dispatchers.IO) { /* ... */ }
}
// injected: production defaults to IO, tests swap it in
class GoodRepo(private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO) {
suspend fun fetch() = withContext(ioDispatcher) { /* ... */ }
}
val repo = GoodRepo(ioDispatcher = UnconfinedTestDispatcher(testScheduler))
viewModelScope has the same hardcoding problem, except it's baked into the platform: it's built on Dispatchers.Main, and Dispatchers.Main needs a real Android Looper to back it, which does not exist on a plain JVM unit test. Construct a ViewModel in a bare JUnit test and anything that touches viewModelScope throws immediately, an exception about a missing main dispatcher. The fix is to install a test dispatcher in Main's place before the test runs, with Dispatchers.setMain(testDispatcher), and remove it afterward with Dispatchers.resetMain(), so you don't leak a test dispatcher into other tests. Doing that by hand in every test class is tedious and easy to forget the teardown for, so the idiom is a JUnit rule, commonly called MainDispatcherRule, that wraps setMain and resetMain around every test automatically.
class MainDispatcherRule : TestWatcher() {
private val testDispatcher = StandardTestDispatcher()
override fun starting(d: Description?) = Dispatchers.setMain(testDispatcher)
override fun finished(d: Description?) = Dispatchers.resetMain()
}
class MyViewModelTest {
@get:Rule val mainRule = MainDispatcherRule()
@Test
fun test() = runTest {
val vm = MyViewModel() // viewModelScope now runs on testDispatcher
advanceUntilIdle()
}
}
One detail about setMain matters once you start mixing multiple test dispatchers in the same test: they all need to share one TestCoroutineScheduler, because that scheduler is the single virtual clock, and advancing it only affects coroutines tied to it. If you call Dispatchers.setMain(StandardTestDispatcher()) in setup, any StandardTestDispatcher you construct afterward without explicitly passing a scheduler automatically adopts the scheduler from whatever's currently installed as Main. That's convenient, it means you rarely have to thread testScheduler through by hand once Main is set. But if you ever construct a TestDispatcher with its own scheduler on purpose, or before setMain runs, you end up with two independent clocks, and advancing one leaves coroutines on the other frozen, a confusing failure mode if you don't know to look for it.
val mainDispatcher = StandardTestDispatcher()
Dispatchers.setMain(mainDispatcher)
// no scheduler passed, so it adopts Main's scheduler automatically
val secondDispatcher = StandardTestDispatcher()
assertSame(mainDispatcher.scheduler, secondDispatcher.scheduler)
Not every coroutine in a test is supposed to finish. A StateFlow collector, or anything watching a hot stream, runs forever by design, that's the point of it. Launch that kind of work in runTest's main TestScope and the test hangs, because runTest waits for its own scope to fully complete before returning, and a collector that never completes means it never will. TestScope.backgroundScope exists for exactly this case: anything launched there is cancelled automatically the moment the test body returns, so runTest never waits on it. The rule of thumb is simple, use the main scope for work that's supposed to finish during the test, and backgroundScope for anything open-ended that you're just observing.
@Test
fun collectHotStateFlow() = runTest {
val vm = MyViewModel()
val values = mutableListOf<UiState>()
backgroundScope.launch { // auto-cancelled when the test body ends
vm.uiState.toList(values) // never completes on its own; fine here
}
advanceUntilIdle()
assertEquals(UiState.Success, values.last())
}
That waiting behavior cuts both ways. If you accidentally launch something in the main TestScope that never completes, maybe it's awaiting a signal that's never sent, runTest doesn't hang forever and it doesn't silently pass either. It waits on its own scope, notices the coroutine still hasn't finished after the test body returns, and fails the test with an UncompletedCoroutinesError. That failure is your signal that some piece of work belongs in backgroundScope instead of the main scope, or that you're missing a call to complete a signal the coroutine is genuinely waiting on.
// bad: never-completing coroutine in the main scope
@Test
fun badTest() = runTest {
launch { awaitCancellation() } // fails: UncompletedCoroutinesError
}
// good: move it to backgroundScope, which is cancelled automatically
@Test
fun goodTest() = runTest {
backgroundScope.launch { awaitCancellation() }
}
SharingStarted.WhileSubscribed is a common way to expose a ViewModel's state as a StateFlow without paying for upstream work when nobody's watching, it only runs the upstream flow while at least one collector is active, then keeps it alive for a grace period after the last one leaves. That's exactly what trips people up in tests. If you construct the ViewModel, call advanceUntilIdle, and immediately read uiState.value expecting the loaded result, you'll often find it's still sitting at its initial value, because nothing has subscribed yet, so the upstream flow the WhileSubscribed policy is guarding never started. You have to start collecting first, commonly in backgroundScope, before advancing time and reading the value.
@Test
fun whileSubscribedNeedsCollector() = runTest {
val vm = MyViewModel() // uiState = stateIn(WhileSubscribed(5000), Initial)
advanceUntilIdle()
assertEquals(Initial, vm.uiState.value) // still Initial, nothing subscribed
backgroundScope.launch { vm.uiState.collect() }
advanceUntilIdle()
assertEquals(Success, vm.uiState.value)
}
Asserting Flow emissions by collecting into a mutable list and comparing at the end works, but it's clunky, and it's easy to get subtly wrong about ordering or completion. Turbine turns each emission into a one-liner: flow.test { ... } collects the flow inside the block and gives you awaitItem() to suspend for and return the next value, awaitComplete() to assert the flow finished normally with no exception, and awaitError() to assert it finished by throwing a Throwable instead. Reaching for the right one of those three tells the reader exactly what you expect to happen at the end of the stream, not just what values came out of it.
@Test
fun emitsLoadingThenSuccess() = runTest {
flowOf(Loading, Success(data)).test {
assertEquals(Loading, awaitItem())
assertEquals(Success(data), awaitItem())
awaitComplete()
}
}
Turbine is strict on purpose. If the test block ends while there's still an emitted item, or the completion event, sitting unread, it doesn't let that pass quietly, it throws a TurbineAssertionError listing exactly what was left over. That's a deliberate design choice, a flow that emitted something you never asserted on is usually a bug in the test, not a fact you get to ignore. When you genuinely don't care what's left, because you've already asserted the interesting part and just want to stop, you say so explicitly with cancelAndIgnoreRemainingEvents(), rather than the test failing to tell you it noticed something you didn't check.
// fails: "Hello" was emitted but never consumed
flow.test {
// forgot awaitItem()
} // TurbineAssertionError: Unconsumed events found: Item(Hello), Complete
// fine: explicitly discard whatever's left
flow.test {
assertEquals("Hello", awaitItem())
cancelAndIgnoreRemainingEvents()
}
Here's a case where Turbine's ordering guarantees actually save you from a real bug. A MutableSharedFlow with replay set to zero drops any emission that happens before a collector subscribes, that's normal hot flow behavior, there's no buffer holding onto it. If you wrote plain collect-then-emit code by hand, you could easily emit before your collector had actually started and lose the value with no error at all. Turbine's test { } avoids that trap by construction: it starts collecting first and only then hands control to your block, so an emit() call written inside the block is guaranteed to happen after the collector is already active, and nothing gets dropped.
val shared = MutableSharedFlow<Int>() // replay = 0
shared.test {
// Turbine guarantees the collector is active before this line runs,
// so the emit below is not dropped
shared.emit(42)
assertEquals(42, awaitItem())
cancelAndIgnoreRemainingEvents()
}
Everything so far comes together in one shape, the way you'd actually test a ViewModel in a real codebase. Inject a fake repository instead of a real network client, so the test controls exactly what data comes back and when. Install a MainDispatcherRule so viewModelScope lands on a test dispatcher instead of throwing on the missing Main. Run the test body in runTest, trigger whatever action loads the state, drive virtual time with advanceUntilIdle, and then assert on the resulting state, either by reading state.value directly or by collecting emissions with Turbine, launched in backgroundScope if the state never stops emitting. What's conspicuously absent from that list is a real network call, a Thread.sleep, or a physical device, that combination is exactly what makes coroutine-heavy ViewModel logic fast and deterministic to test.
class MyViewModelTest {
@get:Rule val mainRule = MainDispatcherRule()
private val fakeRepo = FakeUserRepository()
@Test
fun loadsUserIntoState() = runTest {
val vm = MyViewModel(fakeRepo)
vm.load(userId = 1)
advanceUntilIdle()
assertEquals(UiState.Success(user), vm.state.value)
}
}
When a coroutine fails in production, the stack trace tells you *what* threw, but in a big app with hundreds of concurrent coroutines, it rarely tells you *which* coroutine. Two tools fix that.
**1. CoroutineName** -- gives a human-readable label:
scope.launch(CoroutineName("sync-contacts")) {
fetchAndSave()
}
With the JVM flag -Dkotlinx.coroutines.debug enabled, the coroutine's name and an auto-assigned numeric ID appear in the thread name:
Thread: DefaultDispatcher-worker-2 @sync-contacts#47
Without the flag, you just see DefaultDispatcher-worker-2 -- no coroutine identity at all. The flag is lightweight enough to run in debug builds and staging environments. In production, most teams leave it off for the small performance overhead, but turn it on when chasing a specific bug.
**2. The debug flag also enables automatic IDs.** Even without CoroutineName, every coroutine gets a #N suffix so you can distinguish them:
DefaultDispatcher-worker-1 @coroutine#12
DefaultDispatcher-worker-3 @coroutine#13
The combination is powerful: add CoroutineName to your important coroutines (network sync, database writes, event processing), enable the debug flag in your test configuration, and every crash trace tells you exactly which logical operation failed.
For deeper debugging beyond thread names, the **kotlinx-coroutines-debug** library gives you a full picture of every live coroutine -- where it's suspended, what it's waiting for, and its parent-child tree.
Add it as a test dependency:
testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-debug:1.9.0")
The main entry point is DebugProbes:
@Before
fun setup() {
DebugProbes.install()
}
@After
fun teardown() {
DebugProbes.uninstall()
}
Once installed, DebugProbes.dumpCoroutines() prints every live coroutine with its state and creation stack trace:
Coroutine "sync-contacts#47", state: SUSPENDED
at UserRepo.fetchUsers(UserRepo.kt:34)
at SyncWorker.run(SyncWorker.kt:12)
(Coroutine creation stacktrace)
at SyncWorker.start(SyncWorker.kt:8)
This is invaluable for diagnosing hangs: if a test times out, dump the coroutines and you'll see exactly which one is stuck and where. You can also use DebugProbes.withDebugProbes { } in a test to install and uninstall automatically:
@Test
fun `find the stuck coroutine`() = DebugProbes.withDebugProbes {
runTest {
// ... test that might hang
DebugProbes.dumpCoroutines() // see what's alive
}
}
Important: DebugProbes adds significant overhead, so use it only in tests and local debugging, never in production.