0Pricing
Kotlin Academy · Урок

Тестирование StateFlow и SharedFlow с Turbine

Пишите модульные тесты для состояния на основе Flow с помощью библиотеки тестирования Turbine.

«Тестирование StateFlow и SharedFlow с Turbine» — бесплатный урок Kotlin Academy на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения Kotlin Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс Kotlin Academy содержит 4 уроков всего.

Зачем нужен Turbine

Turbine — это библиотека для тестирования Kotlin Flow, которая упрощает проверку выдаваемых значений. Без неё для тестирования потоков требуется вручную собирать значения из канала с ограничениями времени.

// Add to test dependencies:
// testImplementation("app.cash.turbine:turbine:1.0.0")
// testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.7.3")
import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun main() { println("Turbine + runTest = clean flow testing") }

Базовое тестирование потоков с помощью test()

flow.test { } собирает поток и позволяет проверять каждое выданное значение с помощью awaitItem(), awaitComplete() и awaitError().

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testBasicFlow() = runTest {
    val flow = flowOf(1, 2, 3)
    flow.test {
        assertEquals(1, awaitItem())
        assertEquals(2, awaitItem())
        assertEquals(3, awaitItem())
        awaitComplete()
    }
}

Тестирование StateFlow

StateFlow всегда выдаёт текущее значение при начале сбора. awaitItem() библиотеки Turbine получает это начальное значение.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testStateFlow() = runTest {
    val state = MutableStateFlow("loading")
    state.test {
        assertEquals("loading", awaitItem())  // initial value
        state.value = "success"
        assertEquals("success", awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

Тестирование событий SharedFlow

SharedFlow с replay=0 выдаёт только будущие события. Сначала начните сбор, затем выдайте событие.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testSharedFlow() = runTest {
    val events = MutableSharedFlow<String>()
    events.test {
        events.emit("LoginSuccess")
        assertEquals("LoginSuccess", awaitItem())
        events.emit("Navigate")
        assertEquals("Navigate", awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

runTest для виртуального времени

runTest использует виртуальное время: вызовы delay() мгновенно продвигают виртуальное время, поэтому тесты с задержками выполняются быстро.

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
import app.cash.turbine.*
fun testDelayedFlow() = runTest {
    val flow = flow {
        delay(1000)  // virtual — runs instantly
        emit("done")
    }
    flow.test {
        assertEquals("done", awaitItem())
        awaitComplete()
    }
}

awaitError для тестирования исключений

awaitError() проверяет, что поток завершился с исключением. Используйте его вместе с ожидаемым типом исключения.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testFlowError() = runTest {
    val flow = flow {
        emit(1)
        throw RuntimeException("stream error")
    }
    flow.test {
        assertEquals(1, awaitItem())
        val error = awaitError()
        assertEquals("stream error", error.message)
    }
}

expectNoEvents и cancelAndIgnoreRemainingEvents

expectNoEvents() проверяет, что не было выдано ни одного значения. cancelAndIgnoreRemainingEvents() отменяет сбор, игнорируя ожидающие выдачи.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testNoEvents() = runTest {
    val flow = MutableSharedFlow<Int>()
    flow.test {
        expectNoEvents()  // nothing emitted yet
        flow.emit(1)
        assertEquals(1, awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

Тестирование ViewModel с Turbine

Тестируйте StateFlow ViewModel, внедряя имитированный репозиторий и проверяя переходы состояния.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
sealed class State { object Loading : State(); data class Data(val v: String) : State() }
class FakeRepo { suspend fun fetch() = "result" }
class VM(repo: FakeRepo, scope: kotlinx.coroutines.CoroutineScope) {
    private val _s = MutableStateFlow<State>(State.Loading)
    val state = _s.asStateFlow()
    init { scope.launch { _s.value = State.Data(repo.fetch()) } }
}
fun testVM() = runTest {
    val vm = VM(FakeRepo(), backgroundScope)
    vm.state.test {
        assertEquals(State.Loading, awaitItem())
        val data = awaitItem() as State.Data
        assertEquals("result", data.v)
        cancelAndIgnoreRemainingEvents()
    }
}

Несколько сборщиков с turbineScope

turbineScope { } позволяет одновременно тестировать несколько потоков, используя отдельный экземпляр Turbine для каждого.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testMultipleFlows() = runTest {
    val a = MutableStateFlow(1)
    val b = MutableStateFlow("x")
    turbineScope {
        val aTurbine = a.testIn(backgroundScope)
        val bTurbine = b.testIn(backgroundScope)
        assertEquals(1, aTurbine.awaitItem())
        assertEquals("x", bTurbine.awaitItem())
        a.value = 2
        assertEquals(2, aTurbine.awaitItem())
        aTurbine.cancelAndIgnoreRemainingEvents()
        bTurbine.cancelAndIgnoreRemainingEvents()
    }
}

Тестирование операторов потоков

Тестируйте поведение собственных операторов потоков и преобразований изолированно с помощью Turbine.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testOperators() = runTest {
    (1..5).asFlow()
        .filter { it % 2 == 0 }
        .map { it * 10 }
        .test {
            assertEquals(20, awaitItem())
            assertEquals(40, awaitItem())
            awaitComplete()
        }
}

Тестирование с продвижением условного времени

Используйте testScheduler.advanceTimeBy(), чтобы точно управлять виртуальным временем в тестах, содержащих delay.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testTimedFlow() = runTest {
    val flow = flow {
        emit("start")
        delay(500)
        emit("end")
    }
    flow.test {
        assertEquals("start", awaitItem())
        testScheduler.advanceTimeBy(500)
        assertEquals("end", awaitItem())
        awaitComplete()
    }
}

Распространённые ошибки

Всегда отменяйте Turbine в конце тестов горячих потоков. Если забыть вызвать cancelAndIgnoreRemainingEvents(), тесты зависнут в ожидании новых событий.

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testGoodPractice() = runTest {
    val state = MutableStateFlow(0)
    state.test {
        awaitItem()  // initial 0
        state.value = 1
        awaitItem()  // 1
        // Always cancel hot flows:
        cancelAndIgnoreRemainingEvents()
    }
}

Быстрая проверка

Какая функция Turbine получает следующий выданный элемент и выбрасывает исключение, если поток сначала завершается или завершается с ошибкой?

Повторение

Turbine упрощает тестирование Flow с помощью awaitItem(), awaitComplete() и awaitError(). Используйте runTest для виртуального времени. Всегда отменяйте тесты горячих потоков с помощью cancelAndIgnoreRemainingEvents().

Часто задаваемые вопросы

Урок «Тестирование StateFlow и SharedFlow с Turbine» бесплатный?

Да — полный текст урока «Тестирование StateFlow и SharedFlow с Turbine» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс Kotlin Academy, подпишись на CoddyKit PRO. Курс Kotlin Academy содержит 4 уроков всего.

Чему я научусь в уроке «Тестирование StateFlow и SharedFlow с Turbine»?

Пишите модульные тесты для состояния на основе Flow с помощью библиотеки тестирования Turbine. Ты практикуешь Kotlin Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать Kotlin Academy?

Предыдущий опыт не требуется. Kotlin Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.

Сколько времени занимает урок «Тестирование StateFlow и SharedFlow с Turbine»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке Kotlin Academy?

Да. Каждый урок Kotlin Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. StateFlow: горячий контейнер состояния для интерфейса
  2. SharedFlow: шины событий и одноразовые события
  3. Преобразование холодного Flow в горячий с shareIn и stateIn
  4. Тестирование StateFlow и SharedFlow с Turbine
← Назад к Kotlin Academy