Flow 单元测试 (Turbine)

源：Turbine GitHub

虽然 runTest 配合 toList 可以测试简单的 Flow，但对于复杂的流（尤其是热流、错误处理），Google 官方推荐使用 CashApp 开源的 Turbine 库。

依赖配置

dependencies {
    testImplementation("app.cash.turbine:turbine:1.0.0")
}

核心 API: test

Turbine 提供了一个 test 扩展函数，将流的收集过程转变为类似“队列”的挂起操作。

@Test
fun testSimpleFlow() = runTest {
    flowOf("A", "B").test {
        // 依次断言发射的元素
        assertEquals("A", awaitItem())
        assertEquals("B", awaitItem())
        
        // 断言流正常结束
        awaitComplete()
    }
}

测试异常

@Test
fun testError() = runTest {
    flow {
        emit(1)
        throw RuntimeException("Boom!")
    }.test {
        assertEquals(1, awaitItem())
        
        // 断言异常
        val error = awaitError()
        assertEquals("Boom!", error.message)
    }
}

配合 Hot Flow (StateFlow)

测试 StateFlow 时，Turbine 特别有用，因为它可以跳过中间值（如果是 SharedFlow）或只关注最新的状态。

@Test
fun testViewModel() = runTest {
    val viewModel = MyViewModel()
    
    // 开始观察状态
    viewModel.uiState.test {
        // 初始状态
        assertEquals(UiState.Loading, awaitItem())
        
        // 触发业务逻辑
        viewModel.loadData()
        
        // 验证后续状态
        assertEquals(UiState.Success, awaitItem())
    }
}

忽略剩余元素

有时我们只关心前几个值，不想处理 awaitComplete。可以使用 cancelAndIgnoreRemainingEvents。

infiniteFlow.test {
    assertEquals(1, awaitItem())
    cancelAndIgnoreRemainingEvents() // 退出测试块
}