Testing Android ViewModels with Kotlin Flow, Turbine, and Coroutines
ViewModels are the core of Android architecture: they hold state, handle business logic, and expose data via Kotlin Flow. Testing them thoroughly — including loading states, error handling, and side effects — requires understanding how to control coroutines and observe Flow emissions in tests.
This guide covers ViewModel testing with TestDispatcher, Turbine (Flow testing library), and StateFlow/SharedFlow patterns.
Setup
// app/build.gradle.kts
dependencies {
testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.0")
testImplementation("app.cash.turbine:turbine:1.1.0")
testImplementation("io.mockk:mockk:1.13.10")
testImplementation("com.google.truth:truth:1.4.2")
testImplementation("junit:junit:4.13.2")
}Test Dispatcher Setup
Without test dispatcher control, coroutines started in viewModelScope don't complete synchronously:
// MainDispatcherRule.kt
class MainDispatcherRule(
val testDispatcher: TestDispatcher = UnconfinedTestDispatcher()
) : TestWatcher() {
override fun starting(description: Description?) {
Dispatchers.setMain(testDispatcher)
}
override fun finished(description: Description?) {
Dispatchers.resetMain()
}
}Use UnconfinedTestDispatcher (runs coroutines eagerly, good for most ViewModel tests) or StandardTestDispatcher (requires manual advanceUntilIdle(), good for testing timing).
Basic ViewModel Test
@HiltViewModel
class OrderListViewModel @Inject constructor(
private val repository: OrderRepository
) : ViewModel() {
sealed class UiState {
object Loading : UiState()
data class Success(val orders: List<Order>) : UiState()
data class Error(val message: String) : UiState()
}
private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
val uiState: StateFlow<UiState> = _uiState.asStateFlow()
init {
loadOrders()
}
fun loadOrders() {
viewModelScope.launch {
_uiState.value = UiState.Loading
try {
val orders = repository.getOrders()
_uiState.value = UiState.Success(orders)
} catch (e: Exception) {
_uiState.value = UiState.Error(e.message ?: "Unknown error")
}
}
}
}class OrderListViewModelTest {
@get:Rule
val mainDispatcherRule = MainDispatcherRule()
private val repository = mockk<OrderRepository>()
private lateinit var viewModel: OrderListViewModel
@Test
fun `initial load shows orders`() = runTest {
val orders = listOf(
Order("ORD-1", 50.0, OrderStatus.COMPLETED),
Order("ORD-2", 30.0, OrderStatus.PENDING),
)
coEvery { repository.getOrders() } returns orders
viewModel = OrderListViewModel(repository)
val state = viewModel.uiState.value
assertThat(state).isInstanceOf(OrderListViewModel.UiState.Success::class.java)
assertThat((state as OrderListViewModel.UiState.Success).orders).isEqualTo(orders)
}
@Test
fun `load failure shows error`() = runTest {
coEvery { repository.getOrders() } throws IOException("Network error")
viewModel = OrderListViewModel(repository)
val state = viewModel.uiState.value
assertThat(state).isInstanceOf(OrderListViewModel.UiState.Error::class.java)
assertThat((state as OrderListViewModel.UiState.Error).message).contains("Network error")
}
}Testing with Turbine
Turbine makes testing Flow emissions clean and readable:
class OrderListViewModelTest {
@get:Rule
val mainDispatcherRule = MainDispatcherRule(StandardTestDispatcher())
private val repository = mockk<OrderRepository>()
@Test
fun `loadOrders emits loading then success`() = runTest {
val orders = listOf(Order("ORD-1", 50.0, OrderStatus.COMPLETED))
coEvery { repository.getOrders() } returns orders
val viewModel = OrderListViewModel(repository)
viewModel.uiState.test {
assertThat(awaitItem()).isEqualTo(OrderListViewModel.UiState.Loading)
advanceUntilIdle()
assertThat(awaitItem())
.isEqualTo(OrderListViewModel.UiState.Success(orders))
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `loadOrders emits loading then error on failure`() = runTest {
coEvery { repository.getOrders() } throws RuntimeException("Server error")
val viewModel = OrderListViewModel(repository)
viewModel.uiState.test {
assertThat(awaitItem()).isEqualTo(OrderListViewModel.UiState.Loading)
advanceUntilIdle()
val errorState = awaitItem()
assertThat(errorState).isInstanceOf(OrderListViewModel.UiState.Error::class.java)
assertThat((errorState as OrderListViewModel.UiState.Error).message)
.contains("Server error")
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `refresh reloads orders`() = runTest {
val initialOrders = listOf(Order("ORD-1", 50.0, OrderStatus.COMPLETED))
val refreshedOrders = listOf(
Order("ORD-1", 50.0, OrderStatus.COMPLETED),
Order("ORD-2", 30.0, OrderStatus.PENDING),
)
coEvery { repository.getOrders() } returnsMany listOf(initialOrders, refreshedOrders)
val viewModel = OrderListViewModel(repository)
viewModel.uiState.test {
skipItems(2) // skip Loading + initial Success
viewModel.loadOrders()
advanceUntilIdle()
assertThat(awaitItem()).isEqualTo(OrderListViewModel.UiState.Loading)
assertThat(awaitItem())
.isEqualTo(OrderListViewModel.UiState.Success(refreshedOrders))
cancelAndIgnoreRemainingEvents()
}
}
}Testing SharedFlow (Side Effects / Events)
Use SharedFlow for one-time events (navigation, snackbar, error dialog):
class CheckoutViewModel @Inject constructor(
private val paymentService: PaymentService,
private val cartRepository: CartRepository
) : ViewModel() {
private val _uiState = MutableStateFlow(CheckoutUiState())
val uiState: StateFlow<CheckoutUiState> = _uiState.asStateFlow()
// One-time events
private val _events = MutableSharedFlow<CheckoutEvent>()
val events: SharedFlow<CheckoutEvent> = _events.asSharedFlow()
fun checkout() {
viewModelScope.launch {
_uiState.update { it.copy(isLoading = true) }
try {
val result = paymentService.processPayment(_uiState.value.cartId)
cartRepository.clear(_uiState.value.cartId)
_events.emit(CheckoutEvent.NavigateToConfirmation(result.orderId))
} catch (e: PaymentException) {
_events.emit(CheckoutEvent.ShowError(e.message ?: "Payment failed"))
} finally {
_uiState.update { it.copy(isLoading = false) }
}
}
}
}
sealed class CheckoutEvent {
data class NavigateToConfirmation(val orderId: String) : CheckoutEvent()
data class ShowError(val message: String) : CheckoutEvent()
}class CheckoutViewModelTest {
@get:Rule
val mainDispatcherRule = MainDispatcherRule()
private val paymentService = mockk<PaymentService>()
private val cartRepository = mockk<CartRepository>(relaxed = true)
@Test
fun `checkout success emits navigation event`() = runTest {
coEvery { paymentService.processPayment(any()) } returns PaymentResult(orderId = "ORD-999")
val viewModel = CheckoutViewModel(paymentService, cartRepository)
viewModel.events.test {
viewModel.checkout()
val event = awaitItem()
assertThat(event).isInstanceOf(CheckoutEvent.NavigateToConfirmation::class.java)
assertThat((event as CheckoutEvent.NavigateToConfirmation).orderId).isEqualTo("ORD-999")
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `checkout failure emits error event`() = runTest {
coEvery { paymentService.processPayment(any()) } throws PaymentException("Card declined")
val viewModel = CheckoutViewModel(paymentService, cartRepository)
viewModel.events.test {
viewModel.checkout()
val event = awaitItem()
assertThat(event).isInstanceOf(CheckoutEvent.ShowError::class.java)
assertThat((event as CheckoutEvent.ShowError).message).contains("Card declined")
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `checkout clears cart on success`() = runTest {
coEvery { paymentService.processPayment(any()) } returns PaymentResult(orderId = "ORD-999")
val viewModel = CheckoutViewModel(paymentService, cartRepository)
viewModel.checkout()
coVerify { cartRepository.clear(any()) }
}
@Test
fun `loading state is true during checkout then false after`() = runTest {
coEvery { paymentService.processPayment(any()) } returns PaymentResult(orderId = "ORD-999")
val viewModel = CheckoutViewModel(paymentService, cartRepository)
viewModel.uiState.test {
val initial = awaitItem()
assertThat(initial.isLoading).isFalse()
viewModel.checkout()
// Can't test loading=true in UnconfinedTestDispatcher (runs eagerly)
// Use StandardTestDispatcher for this:
val final = awaitItem()
assertThat(final.isLoading).isFalse() // loading cleared after completion
cancelAndIgnoreRemainingEvents()
}
}
}Testing Loading State Transitions
To test the loading state during execution, use StandardTestDispatcher:
class CheckoutViewModelLoadingTest {
@get:Rule
val mainDispatcherRule = MainDispatcherRule(StandardTestDispatcher())
@Test
fun `loading is true during checkout`() = runTest {
val paymentService = mockk<PaymentService>()
val cartRepository = mockk<CartRepository>(relaxed = true)
coEvery { paymentService.processPayment(any()) } coAnswers {
delay(1000) // simulate network delay
PaymentResult(orderId = "ORD-999")
}
val viewModel = CheckoutViewModel(paymentService, cartRepository)
viewModel.uiState.test {
assertThat(awaitItem().isLoading).isFalse() // initial state
viewModel.checkout()
advanceTimeBy(100) // start coroutine but don't finish
assertThat(awaitItem().isLoading).isTrue() // loading during payment
advanceUntilIdle() // complete the payment
assertThat(awaitItem().isLoading).isFalse() // done
cancelAndIgnoreRemainingEvents()
}
}
}Testing Flow Transformation in ViewModel
When ViewModel transforms a repository Flow:
class SearchViewModel @Inject constructor(
private val repository: ProductRepository
) : ViewModel() {
private val searchQuery = MutableStateFlow("")
val searchResults: StateFlow<List<Product>> = searchQuery
.debounce(300)
.flatMapLatest { query ->
if (query.isBlank()) flowOf(emptyList())
else repository.search(query)
}
.stateIn(
scope = viewModelScope,
started = SharingStarted.WhileSubscribed(5000),
initialValue = emptyList()
)
fun onSearch(query: String) {
searchQuery.value = query
}
}class SearchViewModelTest {
@get:Rule
val mainDispatcherRule = MainDispatcherRule(StandardTestDispatcher())
@Test
fun `search query returns results`() = runTest {
val repository = mockk<ProductRepository>()
val results = listOf(Product("Widget Pro", 49.99))
coEvery { repository.search("widget") } returns flowOf(results)
val viewModel = SearchViewModel(repository)
viewModel.searchResults.test {
assertThat(awaitItem()).isEmpty() // initial empty state
viewModel.onSearch("widget")
advanceTimeBy(400) // past the 300ms debounce
assertThat(awaitItem()).isEqualTo(results)
cancelAndIgnoreRemainingEvents()
}
}
@Test
fun `empty query clears results`() = runTest {
val repository = mockk<ProductRepository>()
coEvery { repository.search(any()) } returns flowOf(
listOf(Product("Widget", 10.0))
)
val viewModel = SearchViewModel(repository)
viewModel.searchResults.test {
skipItems(1) // skip initial empty
viewModel.onSearch("widget")
advanceTimeBy(400)
skipItems(1) // skip results
viewModel.onSearch("")
advanceTimeBy(400)
assertThat(awaitItem()).isEmpty()
cancelAndIgnoreRemainingEvents()
}
}
}Fake Repository vs MockK
For repositories with complex state, prefer fakes:
class FakeOrderRepository : OrderRepository {
private val orders = MutableStateFlow<List<Order>>(emptyList())
private var shouldThrow: Exception? = null
fun addOrder(order: Order) {
orders.value = orders.value + order
}
fun setError(e: Exception) {
shouldThrow = e
}
override fun observeOrders(): Flow<List<Order>> = orders
override suspend fun getOrders(): List<Order> {
shouldThrow?.let { throw it }
return orders.value
}
override suspend fun updateOrder(order: Order) {
orders.value = orders.value.map { if (it.id == order.id) order else it }
}
}Use fakes when: the repository has state that tests need to control (add orders, trigger updates). Use MockK when: you just need to stub return values and verify calls.
Common Pitfalls
StateFlow doesn't emit the same value twice. If your test sets the same value, Turbine's awaitItem() hangs. Check for distinctUntilChanged behavior.
SharingStarted.WhileSubscribed and tests: stateIn with WhileSubscribed only subscribes when there's a collector. In Turbine's test {} block, the subscription exists. Outside of it, it doesn't. Subscribe inside test {}.
UnconfinedTestDispatcher skips loading states: UnconfinedTestDispatcher runs coroutines synchronously before returning to the caller. Use StandardTestDispatcher when you need to assert intermediate states like isLoading = true.
Not canceling Turbine: Always end a test {} block with cancelAndIgnoreRemainingEvents(), cancelAndConsumeRemainingEvents(), or consuming all items. Turbine fails the test if you leave items unconsumed.
Summary
ViewModel testing with Flow requires three things: a MainDispatcherRule to control coroutines, Turbine for observing Flow emissions in order, and clear fake/mock collaborators. Use StandardTestDispatcher + advanceUntilIdle() when testing state transitions that involve timing. Use UnconfinedTestDispatcher for simpler tests where you only care about the final state. Test every ViewModel state: loading, success, error, and any side effects via SharedFlow events.
