Testing coroutines and flows in Modern Android & Kotlin
When you build real applications with coroutines and flows, your async code is just as critical as your business logic—but it's also just as easy to get wrong if you're not testing it properly. Testing coroutines and flows in Modern Android & Kotlin requires more than just running your app and hoping it works; you need deterministic async tests that run consistently, complete predictably, and surface race conditions before your users do.
Why async testing matters
Coroutines and flows simplify async code, but they introduce timing dependencies that make testing harder. A naive approach—wrapping tests in runBlocking or using arbitrary delays—leads to flaky tests that pass sometimes and fail others, or tests that take far too long to run.
We test coroutines to catch bugs early: race conditions, resource leaks, cancellation issues, and incorrect state transitions. The tools available in the Kotlin ecosystem make this possible without sacrificing speed or maintainability.
runTest from kotlinx-coroutines-test
runTest is the foundation of coroutine testing. It provides a special test dispatcher that runs your suspending code synchronously, advancing time as needed, so tests complete instantly instead of waiting for real delays.
Here's a practical example. Imagine a UserRepository that fetches a user with a 2-second simulated network delay:
class UserRepository(private val dispatcher: CoroutineDispatcher = Dispatchers.IO) {
suspend fun fetchUser(id: String): User = withContext(dispatcher) {
delay(2000)
User(id, "Alice")
}
}
class UserViewModel(private val repo: UserRepository) : ViewModel() {
private val _uiState = MutableStateFlow(UiState.Loading)
val uiState: StateFlow = _uiState.asStateFlow()
fun loadUser(id: String) {
viewModelScope.launch {
_uiState.value = UiState.Loading
try {
val user = repo.fetchUser(id)
_uiState.value = UiState.Success(user)
} catch (e: Exception) {
_uiState.value = UiState.Error(e.message ?: "Unknown error")
}
}
}
}Without runTest, you'd either add Thread.sleep(2500) (making tests slow) or hope timing doesn't cause failures. With runTest, you control time completely:
@Test
fun loadUser_success() = runTest {
val repo = UserRepository(dispatcher = StandardTestDispatcher(testScheduler))
val viewModel = UserViewModel(repo)
viewModel.loadUser("user123")
// Advance time past the 2-second delay
advanceUntilIdle()
val state = viewModel.uiState.value
assertTrue(state is UiState.Success)
assertEquals("Alice", (state as UiState.Success).user.name)
}Notice we pass StandardTestDispatcher(testScheduler) to the repository. This is key: the repository uses it internally via withContext, so all delays and suspensions run under our test's time control. advanceUntilIdle() fast-forwards past all pending operations instantly.
StandardTestDispatcher controls time
StandardTestDispatcher is your lever for deterministic testing. Instead of wall-clock time, it uses a virtual clock that only advances when you tell it to. This gives you three superpowers: speed (tests run in milliseconds), determinism (no flakiness), and control (you can test exact timing scenarios).
Let's build a more complex example: a feature that debounces user search input and retries failed requests.
class SearchViewModel(
private val searchService: SearchService,
private val dispatcher: CoroutineDispatcher = Dispatchers.IO
) : ViewModel() {
private val _results = MutableStateFlow>(emptyList())
val results: StateFlow> = _results.asStateFlow()
private val searchQuery = MutableStateFlow("")
init {
viewModelScope.launch {
searchQuery
.debounce(300)
.distinctUntilChanged()
.collect { query ->
if (query.isNotEmpty()) {
performSearch(query)
}
}
}
}
fun onQueryChanged(query: String) {
searchQuery.value = query
}
private suspend fun performSearch(query: String) = withContext(dispatcher) {
try {
val results = searchService.search(query)
_results.value = results
} catch (e: Exception) {
// retry logic here
}
}
}Testing the debounce behavior with virtual time:
@Test
fun search_debouncesInput() = runTest {
val service = FakeSearchService()
val viewModel = SearchViewModel(service, StandardTestDispatcher(testScheduler))
// Simulate rapid typing
viewModel.onQueryChanged("a")
advanceTimeBy(100)
viewModel.onQueryChanged("ab")
advanceTimeBy(100)
viewModel.onQueryChanged("abc")
// Search shouldn't happen yet (debounce is 300ms, we've only waited 200ms)
assertEquals(0, service.searchCallCount)
// Advance the remaining 100ms
advanceTimeBy(100)
// Now the search fires
assertEquals(1, service.searchCallCount)
assertEquals("abc", service.lastQuery)
}We control time with surgical precision: advanceTimeBy(100) moves the clock forward by exactly 100ms, advanceUntilIdle() runs everything pending, and currentTime lets you query where you are. This makes it trivial to test timing-sensitive behavior.
Turbine for testing flows
Flows emit sequences of values over time. Testing them means asserting on those sequences—first value, multiple values, completion, errors. That's where Turbine comes in: a small but powerful library that lets you consume flows in tests with clear, readable assertions.
Add it to your test dependencies:
testImplementation("app.cash.turbine:turbine:1.0.0")Here's a real scenario: a chat feature that loads messages incrementally, with a loading state at the start.
class ChatViewModel(
private val chatService: ChatService,
private val dispatcher: CoroutineDispatcher = Dispatchers.IO
) : ViewModel() {
fun loadMessages(roomId: String): Flow = flow {
emit(UiState.Loading)
try {
val messages = withContext(dispatcher) {
chatService.fetchMessages(roomId)
}
emit(UiState.Success(messages))
} catch (e: Exception) {
emit(UiState.Error(e.message ?: "Unknown error"))
}
}
}With Turbine, testing the flow is straightforward:
@Test
fun loadMessages_emitsLoadingThenSuccess() = runTest {
val service = FakeChatService()
val viewModel = ChatViewModel(service, StandardTestDispatcher(testScheduler))
viewModel.loadMessages("room1").test {
// First emission should be Loading
assertEquals(UiState.Loading, awaitItem())
// After fetching completes, we get Success
assertEquals(UiState.Success(listOf(Message("Hello"))), awaitItem())
// Flow should complete (no more items)
awaitComplete()
}
}</code><p></p>
<p>Turbine's <code>test</code> block gives you an intuitive API: <code>awaitItem()</code> waits for and returns the next emitted value (or fails if the timeout expires), <code>awaitComplete()</code> asserts the flow completed normally, and <code>awaitError()</code> asserts an exception was thrown. You can also use <code>expectNoEvents()</code> to verify nothing was emitted during a time window.</p>
<p>Testing error cases is just as clean:</p>
<pre><code class="language-kotlin">@Test
fun loadMessages_emitsError() = runTest {
val service = FakeChatService(shouldFail = true)
val viewModel = ChatViewModel(service, StandardTestDispatcher(testScheduler))
viewModel.loadMessages("room1").test {
assertEquals(UiState.Loading, awaitItem())
val error = awaitItem() as UiState.Error
assertTrue(error.message.isNotEmpty())
awaitComplete()
}
}Inject dispatchers into classes
All of this hinges on one principle: pass your CoroutineDispatcher as a constructor dependency. Don't hard-code Dispatchers.IO or Dispatchers.Main inside classes. This applies to all suspending code: repositories, view models, services.
The pattern is simple:
// Good: dispatcher is injectable
class DataRepository(private val dispatcher: CoroutineDispatcher = Dispatchers.IO) {
suspend fun getData(): Data = withContext(dispatcher) {
// network/database call
}
}
// Bad: dispatcher is hard-coded, impossible to test
class DataRepository {
suspend fun getData(): Data = withContext(Dispatchers.IO) {
// network/database call
}
}In production, you pass the default (real dispatcher). In tests, you pass StandardTestDispatcher(testScheduler). For view models, use viewModelScope as normal—the CoroutineScope will use your injected dispatcher when you call withContext or launch.
If you're using dependency injection (Hilt, Koin, etc.), define a qualifier for test dispatchers:
@Qualifier
@Retention(AnnotationRetention.BINARY)
annotation class TestDispatcher
@Module
@InstallIn(SingletonComponent::class)
object TestDispatcherModule {
@Provides
@TestDispatcher
fun provideTestDispatcher(): CoroutineDispatcher = StandardTestDispatcher()
}Then inject it:
class MyRepository(
@TestDispatcher private val dispatcher: CoroutineDispatcher
) { ... }Common mistakes
Not injecting dispatchers: Hard-coded Dispatchers.Main or Dispatchers.IO means tests can't control timing. Always inject.
Using runBlocking instead of runTest: runBlocking blocks the thread and defeats the purpose of coroutines. Use runTest exclusively for coroutine tests.
Forgetting to call advanceUntilIdle(): If your code has delays or async operations, you must advance the test scheduler past them. Otherwise, assertions run before work completes.
Not catching Turbine timeouts: Turbine waits a default timeout for items. If you expect an item and none arrives, the test times out with a confusing error. Use expectNoEvents() for cases where you expect silence.
Mixing real and test dispatchers: If part of your flow uses Dispatchers.IO and part uses your test dispatcher, you'll have non-deterministic behavior. Inject consistently across the entire call stack.
Key takeaways
runTestruns your coroutines synchronously using a virtual clock, making tests fast and deterministic.StandardTestDispatchergives you fine control over time viaadvanceTimeBy,advanceUntilIdle, andcurrentTime.- Turbine simplifies flow testing with readable assertions:
awaitItem(),awaitComplete(),awaitError(). - Always inject
CoroutineDispatcheras a constructor parameter with a sensible default; never hard-code it inside your logic. - Test both happy paths and edge cases: debouncing, retries, cancellation, and error handling all become trivial with these tools.
Comments
Loading comments…