Turbine для Kotlin: полное руководство по тестированию Flow

1. Что это и зачем нужна

Библиотека Turbine — это современный инструмент для тестирования корутинных потоков данных (Flow) в языке Kotlin. Разработанная компанией Cash App, Turbine решает одну из самых болезненных проблем при написании unit-тестов для асинхронного кода: сложность проверки эмиссии значений из Flow, обработки ошибок и завершения потока. В отличие от стандартных подходов, таких как использование kotlinx-coroutines-test с ручным сбором элементов, Turbine предоставляет декларативный и интуитивно понятный API, который делает тесты читаемыми и надёжными.

Зачем нужна Turbine? Представьте, что вы пишете ViewModel, которая подписывается на репозиторий и трансформирует данные в UI-состояние. Без Turbine вам пришлось бы вручную создавать коллекторы, управлять Job и синхронизировать корутины. Turbine автоматизирует этот процесс: она создаёт виртуальную корутину, собирает все эмиссии в "турбину" (отсюда и название), а затем позволяет проверять каждое значение по порядку, игнорируя неважные промежуточные состояния. Это особенно полезно при тестировании BehaviourSubject-подобных потоков, где важно только финальное состояние.

Основные возможности библиотеки включают: проверку эмитированных значений в строгом порядке, игнорирование промежуточных значений с помощью skip(), тестирование ошибок через awaitError(), проверку завершения потока через awaitComplete(), а также работу с каналами (Channel) и StateFlow. Turbine интегрируется с любым фреймворком тестирования (JUnit, Kotest, Spek) и не требует дополнительных настроек — достаточно добавить зависимость и использовать testScope.

2. Установка

Для использования Turbine в вашем проекте добавьте следующую зависимость в файл build.gradle.kts (или build.gradle):

// build.gradle.kts
dependencies {
    // Для JVM проектов
    testImplementation("app.cash.turbine:turbine:1.1.0")
    
    // Если используете kotlinx-coroutines-test (рекомендуется)
    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.9.0")
}

// Для Android проектов (в android блоке)
android {
    testOptions {
        unitTests.isReturnDefaultValues = true
    }
}

Если вы используете Maven:

<dependency>
    <groupId>app.cash.turbine</groupId>
    <artifactId>turbine</artifactId>
    <version>1.1.0</version>
    <scope>test</scope>
</dependency>

Минимальные требования: Kotlin 1.9+, kotlinx-coroutines 1.8+. Библиотека поддерживает JVM, Android, JS и Native (через Kotlin Multiplatform).

3. Быстрый старт

Рассмотрим простейший пример: тестирование Flow, который эмитирует три числа.

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

class SimpleTurbineTest {

    @Test
    fun `test flow emits three values`() = runTest {
        val flow = flowOf(1, 2, 3)

        flow.test {
            // awaitItem() ожидает следующую эмиссию и возвращает её
            assertEquals(1, awaitItem())
            assertEquals(2, awaitItem())
            assertEquals(3, awaitItem())

            // awaitComplete() проверяет, что поток завершился без ошибок
            awaitComplete()
        }
    }
}

Запустите этот тест — он пройдёт успешно. Обратите внимание: test — это suspend-функция, которая запускает сбор потока в отдельной корутине. Все вызовы awaitItem() блокируют текущую корутину до появления следующего элемента. Если поток завершится раньше, чем мы ожидаем, тест упадёт с понятным сообщением об ошибке.

4. Основные методы/функции/классы

4.1. test — запуск тестирования потока

Сигнатура:

suspend fun <T> Flow<T>.test(
    timeout: Duration = DEFAULT_TIMEOUT,
    validate: suspend TestTurbine<T>.() -> Unit
)

Что делает: Запускает сбор потока в фоновой корутине и предоставляет объект TestTurbine для проверки эмиссий. Параметр timeout задаёт максимальное время ожидания каждого элемента (по умолчанию 1 секунда).

Пример:

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

4.2. awaitItem — ожидание следующего элемента

Сигнатура:

suspend fun <T> TestTurbine<T>.awaitItem(): T

Что делает: Приостанавливает корутину до тех пор, пока поток не эмитирует следующее значение. Возвращает это значение. Если поток завершится с ошибкой, будет выброшено исключение.

Пример:

flow {
    emit(10)
    emit(20)
}.test {
    assertEquals(10, awaitItem())
    assertEquals(20, awaitItem())
    awaitComplete()
}

4.3. awaitComplete — ожидание успешного завершения

Сигнатура:

suspend fun TestTurbine<*>.awaitComplete()

Что делает: Ожидает, что поток завершится без ошибок. Если поток эмитирует исключение, тест упадёт с ошибкой.

Пример:

flowOf(1).test {
    awaitItem()
    awaitComplete() // поток завершился успешно
}

4.4. awaitError — ожидание ошибки

Сигнатура:

suspend fun <T : Throwable> TestTurbine<*>.awaitError(): T

Что делает: Ожидает, что поток завершится с исключением. Возвращает само исключение для дальнейшей проверки.

Пример:

flow<Int> {
    throw RuntimeException("Boom!")
}.test {
    val error = awaitError()
    assertTrue(error is RuntimeException)
    assertEquals("Boom!", error.message)
}

4.5. cancelAndIgnoreRemainingEvents — отмена и игнорирование

Сигнатура:

suspend fun TestTurbine<*>.cancelAndIgnoreRemainingEvents()

Что делает: Отменяет сбор потока и игнорирует все ещё не полученные элементы. Полезно, когда вы проверили только часть эмиссий и не хотите ждать завершения потока.

Пример:

flow {
    emit(1)
    emit(2)
    emit(3) // этот элемент будет проигнорирован
}.test {
    assertEquals(1, awaitItem())
    cancelAndIgnoreRemainingEvents()
}

4.6. skip — пропуск элементов

Сигнатура:

suspend fun <T> TestTurbine<T>.skip(count: Int)

Что делает: Пропускает указанное количество элементов, не проверяя их. Удобно, когда важны только последние состояния (например, при тестировании StateFlow с промежуточными значениями).

Пример:

flow {
    emit("loading")
    emit("data1")
    emit("data2") // нас интересует только последнее состояние
}.test {
    skip(2) // пропускаем "loading" и "data1"
    assertEquals("data2", awaitItem())
    awaitComplete()
}

4.7. expectNoEvents — проверка отсутствия эмиссий

Сигнатура:

suspend fun TestTurbine<*>.expectNoEvents()

Что делает: Проверяет, что в течение заданного таймаута поток не эмитировал никаких значений. Используется для проверки, что поток "молчит" в определённый момент.

Пример:

val coldFlow = flow<Int> {
    delay(100)
    emit(1)
}

coldFlow.test {
    expectNoEvents() // проверяем, что сразу после подписки ничего нет
    assertEquals(1, awaitItem())
    awaitComplete()
}

4.8. awaitSubscription — ожидание подписки

Сигнатура:

suspend fun TestTurbine<*>.awaitSubscription()

Что делает: Ожидает, что поток был подписан (коллектор начал работу). Полезно для тестирования холодных потоков или проверки, что подписка произошла до эмиссии.

Пример:

val sharedFlow = MutableSharedFlow<Int>()

// В реальном тесте вы бы запускали emission в другой корутине
sharedFlow.test {
    awaitSubscription() // ждём, пока тест подпишется
    sharedFlow.emit(42) // теперь можно эмитить
    assertEquals(42, awaitItem())
    cancelAndIgnoreRemainingEvents()
}

4.9. timeout — настройка таймаута

Сигнатура:

fun <T> TestTurbine<T>.timeout(duration: Duration): TestTurbine<T>

Что делает: Устанавливает таймаут для всех последующих операций ожидания в рамках текущего теста. По умолчанию используется 1 секунда.

Пример:

flow {
    delay(2000) // долгая операция
    emit("slow")
}.test {
    timeout(3.seconds) // увеличиваем таймаут до 3 секунд
    assertEquals("slow", awaitItem())
    awaitComplete()
}

5. Реальный практический пример

Рассмотрим полноценный пример: тестирование ViewModel, которая загружает данные из репозитория и трансформирует их в UI-состояние с помощью StateFlow.

import app.cash.turbine.test
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest
import org.junit.Before
import org.junit.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue

// --- Реальные классы приложения ---

data class User(val id: Int, val name: String)

sealed class UiState {
    object Loading : UiState()
    data class Success(val users: List<User>) : UiState()
    data class Error(val message: String) : UiState()
}

interface UserRepository {
    fun getUsers(): Flow<List<User>>
}

class UserViewModel(private val repository: UserRepository) {

    private val _state = MutableStateFlow<UiState>(UiState.Loading)
    val state: StateFlow<UiState> = _state

    suspend fun loadUsers() {
        repository.getUsers().collect { users ->
            _state.value = UiState.Success(users)
        }
    }
}

// --- Мок репозитория для тестов ---

class FakeUserRepository(private val shouldFail: Boolean = false) : UserRepository {
    override fun getUsers(): Flow<List<User>> = flow {
        if (shouldFail) {
            throw RuntimeException("Network error")
        }
        delay(100) // симуляция задержки
        emit(listOf(User(1, "Alice"), User(2, "Bob")))
    }
}

// --- Тесты ---

class UserViewModelTest {

    private lateinit var viewModel: UserViewModel

    @Before
    fun setup() {
        val repo = FakeUserRepository()
        viewModel = UserViewModel(repo)
    }

    @Test
    fun `initial state is Loading`() = runTest {
        // Проверяем начальное состояние без подписки
        assertEquals(UiState.Loading, viewModel.state.value)
    }

    @Test
    fun `loadUsers emits Loading then Success`() = runTest {
        // Подписываемся на StateFlow через Turbine
        viewModel.state.test {
            // Первое состояние должно быть Loading (уже эмитировано при создании)
            assertTrue(awaitItem() is UiState.Loading)

            // Запускаем загрузку
            viewModel.loadUsers()

            // После загрузки должно прийти Success
            val successState = awaitItem()
            assertTrue(successState is UiState.Success)
            assertEquals(2, (successState as UiState.Success).users.size)
            assertEquals("Alice", successState.users[0].name)

            // Поток StateFlow не завершается, поэтому отменяем
            cancelAndIgnoreRemainingEvents()
        }
    }

    @Test
    fun `loadUsers emits Error when repository fails`() = runTest {
        val failingRepo = FakeUserRepository(shouldFail = true)
        val failingViewModel = UserViewModel(failingRepo)

        failingViewModel.state.test {
            // Первое состояние Loading
            assertTrue(awaitItem() is UiState.Loading)

            // Запускаем загрузку, которая упадёт
            failingViewModel.loadUsers()

            //