StateFlow나 SharedFlow로 상태를 노출하는 클래스를 테스트하려고 하면 애매해지는 지점이 있습니다. flow.first()로는 첫 값만 확인할 수 있고, flow.toList()는 무한히 값을 방출하는 StateFlow에서는 영원히 끝나지 않습니다. 결국 CountDownLatch나 Job을 직접 다루면서 값을 리스트에 모으고 나중에 검증하는 코드를 짜게 되는데, 비동기 타이밍 문제로 테스트가 가끔 실패하는 일이 반복됩니다. Cash App이 만든 Turbine은 이 문제를 코루틴 테스트 확장 함수로 풀어줍니다.

왜 필요한가

Flow를 직접 collect해서 리스트에 담는 방식은 몇 가지 문제가 있습니다. 언제까지 collect할지 결정할 방법이 없어서 delay로 임의의 시간을 기다리거나 withTimeout으로 감싸야 하고, 값이 예상보다 늦게 오면 테스트가 timeout 없이 그냥 멈춰버립니다. StateFlow는 구독 시점에 현재 값을 즉시 방출하기 때문에, "초기값"과 "실제로 바뀐 값"을 구분해서 검증하는 코드도 매번 반복됩니다.

Turbine은 Flow<T>.test { } 확장 함수 하나로 이 문제를 해결합니다. 블록 안에서 awaitItem()을 호출할 때마다 다음 값이 도착할 때까지 코루틴이 대기하고, 정해진 시간(기본 3초) 안에 값이 오지 않으면 테스트를 실패시킵니다. kotlinx-coroutines-test의 가상 시간과도 맞물려서, delay가 섞여 있는 Flow도 실제로 몇 초씩 기다릴 필요가 없습니다.

핵심 개념

commonTest에 Turbine과 kotlinx-coroutines-test를 추가합니다.

// build.gradle.kts
kotlin {
    sourceSets {
        commonTest.dependencies {
            implementation(kotlin("test"))
            implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.9.0")
            implementation("app.cash.turbine:turbine:1.2.0")
        }
    }
}

Turbine은 순수 Kotlin 확장 함수라 JVM/Native/JS 어디서나 동일하게 동작합니다. androidTest나 iosTest를 따로 나눌 필요 없이 commonTest에 테스트를 하나만 작성하면 됩니다.

import app.cash.turbine.test
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class CounterTest {
    @Test
    fun `increment 호출 시 count가 1씩 늘어난다`() = runTest {
        val counter = Counter()

        counter.count.test {
            assertEquals(0, awaitItem())

            counter.increment()
            assertEquals(1, awaitItem())

            counter.increment()
            assertEquals(2, awaitItem())

            cancelAndIgnoreRemainingEvents()
        }
    }
}

test 블록은 내부적으로 Flow를 구독하는 코루틴을 하나 띄우고, 블록이 끝날 때 그 코루틴을 정리합니다. awaitItem()으로 소비하지 않은 이벤트가 남아 있으면 블록 종료 시점에 실패로 처리되기 때문에, StateFlow처럼 끝나지 않는 Flow는 마지막에 cancelAndIgnoreRemainingEvents()나 expectNoEvents()로 명시적으로 마무리해야 합니다.

실전 예시

목록 화면의 상태를 StateFlow<UiState>로 노출하는 클래스를 테스트해봅니다.

// commonMain/list/ItemListViewModel.kt
sealed interface UiState {
    data object Idle : UiState
    data object Loading : UiState
    data class Success(val items: List<String>) : UiState
    data class Error(val message: String) : UiState
}

class ItemListViewModel(
    private val repository: ItemRepository,
    scope: CoroutineScope,
) {
    private val _state = MutableStateFlow<UiState>(UiState.Idle)
    val state: StateFlow<UiState> = _state

    fun loadItems() {
        scope.launch {
            _state.value = UiState.Loading
            _state.value = try {
                UiState.Success(repository.fetchItems())
            } catch (e: Exception) {
                UiState.Error(e.message ?: "알 수 없는 오류")
            }
        }
    }
}

성공 케이스는 Idle → Loading → Success 순서로 값이 바뀌는지 그대로 확인합니다.

// commonTest/list/ItemListViewModelTest.kt
class ItemListViewModelTest {
    @Test
    fun `loadItems 성공 시 상태가 순서대로 바뀐다`() = runTest {
        val repository = FakeItemRepository(items = listOf("우유", "계란"))
        val viewModel = ItemListViewModel(repository, scope = this)

        viewModel.state.test {
            assertEquals(UiState.Idle, awaitItem())

            viewModel.loadItems()

            assertEquals(UiState.Loading, awaitItem())
            assertEquals(UiState.Success(listOf("우유", "계란")), awaitItem())

            cancelAndIgnoreRemainingEvents()
        }
    }

    @Test
    fun `loadItems 실패 시 Error 상태로 바뀐다`() = runTest {
        val repository = FakeItemRepository(shouldFail = true)
        val viewModel = ItemListViewModel(repository, scope = this)

        viewModel.state.test {
            skipItems(1) // Idle

            viewModel.loadItems()

            assertEquals(UiState.Loading, awaitItem())
            assertEquals(UiState.Error("네트워크 오류"), awaitItem())

            expectNoEvents()
            cancelAndIgnoreRemainingEvents()
        }
    }
}

skipItems(1)은 값을 검증하지 않고 건너뛸 때 씁니다. 첫 번째 테스트처럼 모든 값을 다 확인하고 싶을 때는 awaitItem()을 그대로 쓰고, 초기값처럼 이번 테스트와 무관한 값은 skipItems로 건너뛰는 편이 어떤 값이 실제로 검증 대상인지 코드에서 더 잘 드러납니다.

상태 Flow와 별개로 "토스트 메시지 보여주기" 같은 1회성 이벤트를 Channel 기반 Flow로 노출하는 경우, state와 이벤트 Flow를 동시에 구독해야 할 때가 있습니다. test 블록을 두 개 순서대로 열면 첫 번째 블록이 끝날 때까지 두 번째 블록이 시작되지 않으므로, 두 Flow를 함께 검증하려면 turbineScope로 감싸고 각각 별도 코루틴에서 구독해야 합니다.

import app.cash.turbine.turbineScope

@Test
fun `저장 성공 시 상태와 이벤트가 함께 바뀐다`() = runTest {
    turbineScope {
        val stateTurbine = viewModel.state.testIn(backgroundScope)
        val eventTurbine = viewModel.event.testIn(backgroundScope)

        assertEquals(UiState.Idle, stateTurbine.awaitItem())

        viewModel.save()

        assertEquals(UiState.Saving, stateTurbine.awaitItem())
        assertEquals(SaveEvent.ShowToast("저장 완료"), eventTurbine.awaitItem())
    }
}

장단점 정리

장점

  • awaitItem/awaitError/awaitComplete만으로 Flow 값을 순서대로 검증하므로, 콜백이나 리스트 누적 코드 없이 테스트가 실제 실행 순서 그대로 읽힘
  • commonTest에서 그대로 동작해서 안드로이드/iOS 타겟을 따로 나눠 테스트를 작성할 필요가 없음
  • kotlinx-coroutines-test의 TestDispatcher와 맞물려서, delay가 섞인 Flow도 가상 시간으로 즉시 검증됨

단점

  • test 블록 종료 시점에 소비하지 않은 이벤트가 남아 있으면 실패하기 때문에, StateFlow처럼 끝나지 않는 Flow마다 cancelAndIgnoreRemainingEvents()를 매번 붙여줘야 함
  • 여러 Flow를 동시에 검증하려면 turbineScope와 testIn으로 따로 풀어써야 해서, 단일 Flow 테스트보다 코드가 한 단계 더 복잡해짐

StateFlow나 SharedFlow로 상태를 노출하는 코드가 늘어날수록 값 하나하나를 수동으로 리스트에 모아 검증하는 비용이 커집니다. Turbine을 넣으면 그 비용 대부분을 awaitItem 한 줄로 줄일 수 있습니다.