[Android] Flow 단위 테스트 라이브러리 Turbine 사용법

단위 테스트 중 Flow를 테스트해야 하는 경우가 있는데, 디벨로퍼에선 이 때 Turbine이라는 라이브러리를 사용하는 예시를 보여주고 있다.

 

https://developer.android.com/kotlin/flow/test?hl=ko#turbine

Android에서 Kotlin 흐름 테스트  |  Android Developers

 

깃허브 링크는 아래를 확인한다. 오늘 기준 최신 버전은 1.2.1이다.

 

https://github.com/cashapp/turbine

GitHub - cashapp/turbine: A testing library for kotlinx.coroutines Flow

 

앱에 적용하려면 버전 카탈로그에 아래처럼 추가한다.

 

[versions]
turbine = "1.2.1"

[libraries]
turbine = { group = "app.cash.turbine", name = "turbine", version.ref = "turbine" }

 

앱 gradle에 추가한다.

 

testImplementation(libs.turbine)

 

이렇게 하면 Turbine을 사용할 수 있게 된다.

아래는 Turbine의 awaitItem(), awaitComplete()를 사용한 간단한 테스트다.

 

import app.cash.turbine.test
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.test.runTest
import org.junit.Assert.assertEquals
import org.junit.Test

class TurbineTest {

    @Test
    fun `순서대로 방출`() = runTest {
        val flow = flowOf("1", "2", "3")
        flow.test {
            assertEquals("1", awaitItem())
            assertEquals("2", awaitItem())
            assertEquals("3", awaitItem())
            awaitComplete()
        }
    }

}

 

awaitItem()은 작업을 일시 정지하고 아이템이 Turbine으로 전송될 때까지 대기한다. 즉 다음으로 방출되는 아이템을 기다렸다가 리턴하는 함수다.

awaitComplete()는 Turbine이 예외 없이 완료될 때까지 일시 정지한다. 즉 Flow가 완료될 때까지 기다리는 함수다.

 

위 테스트가 작동하는 방식을 보면 flowOf()를 통해 3개 문자열을 순차적으로 방출하는 Flow를 만든다.

이후 flow.test {} 를 호출해서 flowOf()로 생성된 Flow를 collect하기 시작한다. test가 호출되면 내부적으로 Flow를 구독하고, 방출되는 모든 값을 큐에 저장한다.

test()의 내부 구현은 아래와 같다. 이 구현이 포함된 파일은 아래 링크를 참고한다.

 

/**
 * Terminal flow operator that collects events from given flow and allows the [validate] lambda to
 * consume and assert properties on them in order. If any exception occurs during validation the
 * exception is rethrown from this method.
 *
 * ```kotlin
 * flowOf("one", "two").test {
 *   assertEquals("one", awaitItem())
 *   assertEquals("two", awaitItem())
 *   awaitComplete()
 * }
 * ```
 *
 * @param timeout If non-null, overrides the current Turbine timeout inside [validate]. See also:
 * [withTurbineTimeout].
 */
public suspend fun <T> Flow<T>.test(
  timeout: Duration? = null,
  name: String? = null,
  validate: suspend TurbineTestContext<T>.() -> Unit,
) {
  turbineScope(timeout) {
    collectTurbineIn(this, null, name).apply {
      TurbineTestContextImpl(this, currentCoroutineContext()).validate()
      cancel()
      ensureAllEventsConsumed()
    }
  }
}

private fun <T> Flow<T>.collectTurbineIn(scope: CoroutineScope, timeout: Duration?, name: String?): ReceiveTurbine<T> {
  // Use test-specific unconfined if test scheduler is in use to inherit its virtual time.
  @OptIn(ExperimentalCoroutinesApi::class) // UnconfinedTestDispatcher is still experimental.
  val unconfined = scope.coroutineContext[TestCoroutineScheduler]
    ?.let(::UnconfinedTestDispatcher)
    ?: Unconfined

  val output = Channel<T>(UNLIMITED)
  val job = scope.launch(unconfined, start = UNDISPATCHED) {
    try {
      collect { output.trySend(it) }
      output.close()
    } catch (e: Throwable) {
      output.close(e)
    }
  }

  return ChannelTurbine(output, job, timeout, name).also {
    scope.reportTurbine(it)
  }
}

 

https://github.com/cashapp/turbine/blob/b2d8f7a695a14852f5871e98b1e5444d7f1065b4/src/commonMain/kotlin/app/cash/turbine/flow.kt

turbine/src/commonMain/kotlin/app/cash/turbine/flow.kt at b2d8f7a695a14852f5871e98b1e5444d7f1065b4 · cashapp/turbine

 

turbineScope로 테스트 전용 격리된 코루틴 스코프를 만들고 collectTurbineIn()으로 Flow 구독을 시작한다.

앞서 큐에 저장한다고 했는데 확인해 보면 Queue가 붙은 실제 큐를 사용하는 게 아닌 Channel<T>을 UNLIMITED로 설정해서 큐처럼 쓰고 있다. trySend()로 채널에 값을 추가하고 receive()로 채널에서 값을 꺼내는데 이 방식이 큐와 비슷해서 큐를 사용한다고 하는 것 같다.

 

아래는 예외를 검증하는 테스트다.

 

import app.cash.turbine.test
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
import org.junit.Test

class TurbineTest {

    @Test
    fun `에러 테스트`() = runTest {
        val errorFlow = flow<String> {
            emit("성공")
            throw RuntimeException("테스트 에러")
        }

        errorFlow.test {
            assertEquals("성공", awaitItem())
            val error = awaitError()
            assertTrue(error is RuntimeException)
            assertEquals("테스트 에러", error.message)
        }
    }

}

 

아래는 뷰모델을 테스트하는 예시다. 뷰모델 구성은 이렇다.

 

import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import com.example.regacyviewexample.data.model.Post
import com.example.regacyviewexample.domain.usecase.GetPostsUseCase
import dagger.hilt.android.lifecycle.HiltViewModel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import javax.inject.Inject

@HiltViewModel
class PostViewModel @Inject constructor(
    private val getPostsUseCase: GetPostsUseCase
) : ViewModel() {

    private val _uiState = MutableStateFlow(PostUiState())
    val uiState: StateFlow<PostUiState> = _uiState.asStateFlow()

    init {
        getPosts()
    }

    fun getPosts() = viewModelScope.launch {
        _uiState.value = _uiState.value.copy(isLoading = true, error = null)

        getPostsUseCase().fold(
            onSuccess = { posts ->
                _uiState.value = _uiState.value.copy(
                    isLoading = false,
                    posts = posts,
                    error = null
                )
            },
            onFailure = { exception ->
                _uiState.value = _uiState.value.copy(
                    isLoading = false,
                    error = exception.message ?: "알 수 없는 에러 발생"
                )
            }
        )
    }
}

data class PostUiState(
    val isLoading: Boolean = false,
    val posts: List<Post> = emptyList(),
    val error: String? = null
)

import androidx.arch.core.executor.testing.InstantTaskExecutorRule
import app.cash.turbine.test
import com.example.regacyviewexample.data.model.Post
import com.example.regacyviewexample.domain.usecase.GetPostsUseCase
import com.example.regacyviewexample.presentation.viewmodel.PostViewModel
import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.mockk
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.resetMain
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.test.setMain
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Rule
import org.junit.Test

@OptIn(ExperimentalCoroutinesApi::class)
class TurbineTest {

    @get:Rule
    val instantTaskExecutorRule = InstantTaskExecutorRule()

    private val testDispatcher = StandardTestDispatcher()
    private val getPostsUseCase: GetPostsUseCase = mockk()
    private lateinit var viewModel: PostViewModel

    @Before
    fun setup() {
        Dispatchers.setMain(testDispatcher)
    }

    @After
    fun tearDown() {
        Dispatchers.resetMain()
    }

    @Test
    fun `초기 상태에서 getPosts 호출 시 성공`() = runTest {
        // Given
        val mockPosts = listOf(
            Post(userId = 1, id = 1, title = "제목1", body = "내용1"),
            Post(userId = 1, id = 2, title = "제목2", body = "내용2")
        )
        coEvery { getPostsUseCase() } returns Result.success(mockPosts)

        // When & Then
        viewModel = PostViewModel(getPostsUseCase)

        viewModel.uiState.test {
            val initialState = awaitItem()
            assertFalse(initialState.isLoading)
            assertTrue(initialState.posts.isEmpty())
            assertNull(initialState.error)

            val loadingState = awaitItem()
            assertTrue(loadingState.isLoading)
            assertTrue(loadingState.posts.isEmpty())
            assertNull(loadingState.error)

            val successState = awaitItem()
            assertFalse(successState.isLoading)
            assertEquals(mockPosts, successState.posts)
            assertNull(successState.error)

            // 더 이상 이벤트가 없어야 함
            expectNoEvents()

            coVerify(exactly = 1) { getPostsUseCase() }
        }
    }

    @Test
    fun `getPosts 호출 시 에러 발생`() = runTest {
        // Given
        val errorMessage = "네트워크 에러"
        coEvery { getPostsUseCase() } returns Result.failure(Exception(errorMessage))

        // When & Then
        viewModel = PostViewModel(getPostsUseCase)

        viewModel.uiState.test {
            val initialState = awaitItem()
            assertFalse(initialState.isLoading)
            assertTrue(initialState.posts.isEmpty())
            assertNull(initialState.error)

            val loadingState = awaitItem()
            assertTrue(loadingState.isLoading)
            assertNull(loadingState.error)

            val errorState = awaitItem()
            assertFalse(errorState.isLoading)
            assertTrue(errorState.posts.isEmpty())
            assertEquals(errorMessage, errorState.error)

            expectNoEvents()
            coVerify(exactly = 1) { getPostsUseCase() }
        }
    }

    @Test
    fun `에러 메시지가 null인 경우 기본 메시지 표시`() = runTest {
        // Given
        val exceptionWithNullMessage = object : Exception() {
            override val message: String? = null
        }
        coEvery { getPostsUseCase() } returns Result.failure(exceptionWithNullMessage)

        // When & Then
        viewModel = PostViewModel(getPostsUseCase)

        viewModel.uiState.test {
            awaitItem() // 초기 상태
            awaitItem() // 로딩 상태

            val errorState = awaitItem()
            assertEquals("알 수 없는 에러 발생", errorState.error)

            expectNoEvents()
        }
    }

}

 

첫 테스트는 awaitItem()을 3번 써서 초기 상태, 로딩 상태, 성공 상태일 때 뷰모델 안의 uiState 값을 각각 검증한다.

 

@Test
fun `초기 상태에서 getPosts 호출시 성공적으로 데이터 로드`() = runTest {
    // Given
    val mockPosts = listOf(
        Post(userId = 1, id = 1, title = "제목1", body = "내용1"),
        Post(userId = 1, id = 2, title = "제목2", body = "내용2")
    )
    coEvery { getPostsUseCase() } returns Result.success(mockPosts)

    // When & Then
    viewModel = PostViewModel(getPostsUseCase)

    viewModel.uiState.test {
        val initialState = awaitItem()
        assertFalse(initialState.isLoading)
        assertTrue(initialState.posts.isEmpty())
        assertNull(initialState.error)

        val loadingState = awaitItem()
        assertTrue(loadingState.isLoading)
        assertTrue(loadingState.posts.isEmpty())
        assertNull(loadingState.error)

        val successState = awaitItem()
        assertFalse(successState.isLoading)
        assertEquals(mockPosts, successState.posts)
        assertNull(successState.error)

        // 더 이상 이벤트가 없어야 함
        expectNoEvents()

        coVerify(exactly = 1) { getPostsUseCase() }
    }
}

 

expectNoEvents()는 이미 수신된 소비되지 않은 이벤트가 없음을 보장하는 함수다. 즉 더 이상 상태 변경이 없음을 확인하는 함수다. 만약 expectNoEvents()가 실행될 때 emit되는 다른 아이템이 있다면 이 테스트는 실패한다.

마지막으로 coVerify를 통해 getPostsUseCase가 1회 실행됐음을 검증하곤 테스트는 끝난다. 다른 실패, 기본 에러 메시지 표시 테스트도 검증 시나리오만 다르고 쓰는 함수는 비슷하다.

 

 

참고한 사이트)

 

https://medium.com/@sharmapraveen91/mastering-turbine-for-flow-testing-your-secret-weapon-for-2025-android-interviews-adb6e43de497

Mastering Turbine for Flow Testing: Your Secret Weapon for 2025 Android Interviews