0Pricing
Kotlin Academy · 课时

使用 Turbine 测试 StateFlow 与 SharedFlow

使用 Turbine 测试库为基于 Flow 的状态编写单元测试。

使用 Turbine 测试 StateFlow 与 SharedFlow 是 CoddyKit 上的免费 Kotlin Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Kotlin Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Kotlin Academy 课程共包含 4 节课。

为什么使用 Turbine？

Turbine 是一个用于 Kotlin 流的测试库，可以轻松断言发出的值。没有它，测试流就需要手动收集通道并设置超时。

// 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 在收集时始终会发出当前值。Turbine 的 awaitItem() 会捕获这次初始发出。

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 事件

使用 replay=0 的 SharedFlow 只会发出之后发生的事件。请先开始收集，再发出事件。

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()
    }
}

使用 Turbine 测试 ViewModel

通过注入一个模拟仓库并断言状态转换，测试 ViewModel 的 StateFlow。

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()
        }
}

使用模拟时间推进进行测试

在涉及 delay 的测试中，使用 testScheduler.advanceTimeBy() 精确控制虚拟时间。

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 使用 awaitItem()、awaitComplete() 和 awaitError() 简化了 Flow 测试。使用 runTest 处理虚拟时间。始终使用 cancelAndIgnoreRemainingEvents() 取消热流测试。

常见问题解答

「使用 Turbine 测试 StateFlow 与 SharedFlow」课时是免费的吗？

是的 — 「使用 Turbine 测试 StateFlow 与 SharedFlow」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Kotlin Academy 课程的其余内容，请升级到 CoddyKit PRO。 Kotlin Academy 课程共包含 4 节课。

「使用 Turbine 测试 StateFlow 与 SharedFlow」这节课中我会学到什么？

使用 Turbine 测试库为基于 Flow 的状态编写单元测试。 你通过在浏览器中直接运行的动手代码来练习 Kotlin Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Kotlin Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Kotlin Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「使用 Turbine 测试 StateFlow 与 SharedFlow」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Kotlin Academy 课中编写并运行代码吗？

能。每节 Kotlin Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. StateFlow：用于界面的热状态容器
  2. SharedFlow：事件总线与一次性事件
  3. 使用 shareIn 与 stateIn 将冷 Flow 转换为热 Flow
  4. 使用 Turbine 测试 StateFlow 与 SharedFlow
← 返回 Kotlin Academy