Flow의 UnitTest를 손쉽게 만드는 방법

OKR

Objective

Kotlin의 Flow/StateFlow를 app.cash.turbine으로 안정적으로 테스트하고, Android 프로젝트에서 재사용 가능한 테스트 패턴을 확립한다.

Key Results

• Turbine의 기본 개념과 장점을 이해한다.
• Flow/StateFlow 테스트를 위한 핵심 API와 패턴을 습득한다.
• runTest·가상 시간·advanceUntilIdle와 함께 Turbine을 올바르게 사용한다.
• 실제 Android 코드(예: ViewModel 상태 스트림)를 검증하는 테스트 템플릿을 확보한다.

---

1) Turbine 개요

• Turbine은 Kotlin Flow를 테스트하기 위한 경량 라이브러리다.
• test { ... } 블록에서 스트림을 수집하고, awaitItem() 같은 API로 방출 값을 순차적으로 검증한다.
• 테스트가 끝나면 수집이 자동으로 취소되어 누수 없이 깔끔하다.

---

2) 의존성 추가

Gradle(Kotlin DSL) 예시:

dependencies {
    testImplementation("app.cash.turbine:turbine:<latest-version>")
    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:<latest-version>")
    testImplementation("junit:junit:<latest-version>")
}

---

3) 최소 예시: Flow 테스트

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

class FlowSampleTest {
    @Test
    fun emitsInOrder() = runTest {
        val target = flow {
            emit("A")
            emit("B")
            emit("C")
        }

        target.test {
            assertEquals("A", awaitItem())
            assertEquals("B", awaitItem())
            assertEquals("C", awaitItem())
            awaitComplete()
        }
    }
}

---

4) StateFlow 테스트 기본

StateFlow는 구독 즉시 최신 값을 1번 방출한다. 따라서 테스트 시작 시 초기 상태를 먼저 검증하자.

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

class StateFlowSampleTest {
    @Test
    fun stateFlowEmitsInitialThenUpdates() = runTest {
        val state = MutableStateFlow(0)

        state.test {
            assertEquals(0, awaitItem()) // 초기값
            state.value = 1
            assertEquals(1, awaitItem())
        }
    }
}

---

5) Android 실전 패턴: UI 상태 스트림 검증

ViewModel 또는 매니저가 내보내는 StateFlow를 Turbine으로 검증한다.

핵심은 “방출 순서”와 “여분 이벤트 없음”을 확인하는 것이다.

검색 화면의 상태 스트림을 Turbine으로 검증하는 예제다.

Idle → Loading → Success/Empty/Error 전이를 한 테스트에서 확인한다.

// Production-like code (간단화를 위해 한 파일에 배치)

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch

sealed interface SearchUiState {
    object Idle : SearchUiState
    object Loading : SearchUiState
    data class Success(val items: List<String>) : SearchUiState
    object Empty : SearchUiState
    data class Error(val message: String) : SearchUiState
}

interface SearchRepository {
    suspend fun search(query: String): List<String>
}

class SearchViewModel(
    private val repo: SearchRepository,
    private val scope: CoroutineScope
) {
    private val _state = MutableStateFlow<SearchUiState>(SearchUiState.Idle)
    val state: StateFlow<SearchUiState> = _state.asStateFlow()

    fun search(query: String) {
        scope.launch {
            _state.value = SearchUiState.Loading
            runCatching { repo.search(query) }
                .onSuccess { list ->
                    _state.value = if (list.isEmpty()) SearchUiState.Empty
                    else SearchUiState.Success(list)
                }
                .onFailure { e ->
                    _state.value = SearchUiState.Error(e.message ?: "unknown error")
                }
        }
    }
}

// Test code

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

class SearchViewModelTest {

    private class FakeRepo : SearchRepository {
        override suspend fun search(query: String): List<String> = when (query) {
            "kotlin" -> listOf("kotlin-101", "kotlin-coroutines")
            "empty" -> emptyList()
            "boom" -> error("boom")
            else -> listOf(query)
        }
    }

    @Test
    fun search_emits_Idle_Loading_Success_then_Empty_and_Error() = runTest {
        val vm = SearchViewModel(repo = FakeRepo(), scope = TestScope(this).backgroundScope)

        vm.state.test {
            // 초기 상태
            assertEquals(SearchUiState.Idle, awaitItem())

            // 1) 성공 케이스
            vm.search("kotlin")
            assertEquals(SearchUiState.Loading, awaitItem())
            advanceUntilIdle()
            val success = awaitItem()
            assertIs<SearchUiState.Success>(success)
            assertEquals(listOf("kotlin-101", "kotlin-coroutines"), success.items)

            // 2) Empty 케이스
            vm.search("empty")
            assertEquals(SearchUiState.Loading, awaitItem())
            advanceUntilIdle()
            assertEquals(SearchUiState.Empty, awaitItem())

            // 3) Error 케이스
            vm.search("boom")
            assertEquals(SearchUiState.Loading, awaitItem())
            advanceUntilIdle()
            val error = awaitItem()
            assertIs<SearchUiState.Error>(error)

            cancelAndIgnoreRemainingEvents()
        }
    }
}

---

6) 자주 쓰는 Turbine API (요약)

• test { ... } : Flow를 수집하며 테스트 블록 실행
• awaitItem() : 다음 아이템이 방출될 때까지 대기해 값을 반환
• awaitComplete() : 스트림 완료를 대기
• expectNoEvents() : 더 이상의 이벤트가 없어야 함을 검증
• cancelAndIgnoreRemainingEvents() : 남은 이벤트를 무시하고 종료
• expectMostRecentItem() : StateFlow의 최신 값을 즉시 획득
• awaitError() : 에러 방출을 대기하고 예외를 반환

예시:

flow.test {
    val first = awaitItem()
    expectNoEvents()
    cancelAndIgnoreRemainingEvents()
}

---

7) 코루틴 테스트와의 결합 포인트

• runTest 환경에서는 가상 시간이 사용된다.
• 비동기 작업(예: launch)이 완료되길 기다리려면 advanceUntilIdle()을 호출한다.
• 지연·타이밍 의존 코드가 있다면 TestScope.currentTime·advanceTimeBy(...) 등을 활용하자.

예시:

@Test
fun delayedEmission() = runTest {
    val target = someDelayedFlow() // 내부에서 delay 사용

    target.test {
        // 아직 방출 전
        expectNoEvents()

        // 지연을 가상 시간으로 스킵
        advanceUntilIdle()

        // 이제 값이 방출됨
        val v = awaitItem()
        cancelAndIgnoreRemainingEvents()
    }
}

---

8) 오류/취소 시나리오 검증

에러를 방출하는 Flow는 awaitError()로 단정할 수 있다.

import app.cash.turbine.test
import kotlin.test.Test
import kotlin.test.assertTrue
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest

class ErrorFlowTest {
    @Test
    fun emitsError() = runTest {
        val target = flow<Int> {
            throw IllegalStateException("boom")
        }

        target.test {
            val e = awaitError()
            assertTrue(e is IllegalStateException)
        }
    }
}

---

9) 상태 전이(Initial → Loading → Success/Empty/Error) 템플릿

Android UI 상태를 다루는 전형적인 전이 시퀀스를 한 번에 검증한다.

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

class UiStateMachineTest {
    @Test
    fun loadSequence() = runTest {
        val vm = /* ViewModel 생성 */

        vm.state.test {
            // Initial
            assertEquals(UiState.Initial, awaitItem())

            // Action
            vm.load()

            // Loading
            assertEquals(UiState.Loading, awaitItem())

            // 완료 후 최종 상태 (예: Success or Empty)
            advanceUntilIdle()
            val final = awaitItem()
            // when/then 분기별 검증
            // assertEquals(UiState.Success(data), final)
        }
    }
}

---

10) 실전 팁

• 테스트가 길어지면 “Given/When/Then” 블록을 주석으로 구분해 가독성을 높인다.
• StateFlow는 구독 즉시 1회 방출한다는 점을 항상 염두에 두고 첫 awaitItem()에서 초기 상태를 검증한다.
• “이 시점에 더 이상 이벤트가 나오면 안 된다”는 것을 expectNoEvents()로 자주 확인한다.
• 방출이 남아 있더라도 테스트를 종료하려면 cancelAndIgnoreRemainingEvents()로 깔끔히 마무리한다.
• Mockito/fixture와 함께 사용할 때는, 방출 순서와 호출 순서를 명시적으로 분리해 테스트를 안정화한다.

---

11) Turbine를 선택하는 이유

• 직관적인 DSL로 방출 순서를 명확히 검증할 수 있다.
• 수집/취소를 자동 관리하여 누수 위험이 낮다.
• kotlinx-coroutines-test와 자연스럽게 결합되어 가상 시간 기반 테스트가 쉽다.
• Android UI 상태 스트림(ViewModel의 StateFlow) 검증에 최적화되어 있다.

---