如何收集测试中 StateFlow 的所有发出？

Question

所以我想测试我的 viewModel 逻辑。我们使用MVI架构，所以我们有UI状态对象。当某个事件发生时，viewModel 会处理它并更新状态。就我而言，应该有 3 个发射。

您可以查看我的问题的示例：

视图模型：

import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class TestViewmodel: ViewModel() {
    private val _event = MutableSharedFlow<TestEvent>()

    private val _state = MutableStateFlow(TestState())
    val state = _state.asStateFlow()

    init {
        viewModelScope.launch {
            _event.collect{
                handleEvent(it)
            }
        }
    }

    private fun handleEvent(event: TestEvent){
        if(event == TestEvent.GetData){
            getTestData()
        }
    }

    private fun getTestData(){
        viewModelScope.launch(Dispatchers.IO){
            _state.update {
                it.copy(
                        loading = true
                )
            }
           // delay(100)

            val data = getApiData()

            _state.update {
                it.copy(
                        loading = false
                )
            }
           // delay(100)

            handleApiData(data)
        }
    }

    private fun handleApiData(list: List<String>){
        _state.update {
            it.copy(
                    data = list
            )
        }
    }

    private suspend fun getApiData(): List<String>{
        delay(1000)
        return listOf("ffefe","ffefe","fdfdfd")
    }

    fun setEvent(event: TestEvent){
        viewModelScope.launch {
            _event.emit(event)
        }
    }
}

data class TestState(
        val loading: Boolean = false,
        val data: List<String> = emptyList()
)

sealed interface TestEvent{
    data object GetData: TestEvent
}

测试类：

import io.kotest.matchers.shouldBe
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.flow.drop
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.test.resetMain
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.test.setMain
import org.junit.jupiter.api.AfterAll
import org.junit.jupiter.api.BeforeAll
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.TestInstance
import org.junit.jupiter.api.extension.ExtendWith

@OptIn(ExperimentalCoroutinesApi::class)
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
class TestViewModelTest(){
    private lateinit var viewModel: TestViewmodel

    @BeforeAll
    fun setUp(){
        Dispatchers.setMain(Dispatchers.Unconfined)
        viewModel = TestViewmodel()
    }

    @AfterAll
    fun afterAll(){
        Dispatchers.resetMain()
    }

    @Test
    fun testEmit() = runTest {
        val state = viewModel.state
        viewModel.setEvent(TestEvent.GetData)

        val result = state.drop(2).first()

        result.data.size shouldBe 3
    }

    @Test
    fun testEmitWithTurbine() = runTest {
        val state = viewModel.state

        state.test {
            viewModel.setEvent(TestEvent.GetData)

            awaitItem()
            awaitItem()
            awaitItem().data.size shouldBe 3
        }
    }

    @Test
    fun testEmitByGoogle() = runTest {
        val state = viewModel.state
        val result = mutableListOf<TestState>()

        val collectionJob = launch {
            state.toList(result)
        }

        viewModel.setEvent(TestEvent.GetData)
        
        result[2].data.size shouldBe 3

        collectionJob.cancel()
    }
}

所以我的问题是，当我尝试测试时，测试运行大约 1 分钟，然后它以 UncompletedCoroutinesError 退出。

我注意到，如果我在每次状态更新后都延迟，测试将会成功，但这不是一个好的解决方案。

我尝试了谷歌文档中的一些想法，但没有帮助。虽然测试没有因 UncompletedCoroutinesError 退出，但测试失败了。它失败了，因为状态属性只存储了第一个发射，至少看起来是这样。

我也尝试过使用Turbine库，但结果是一样的。

当我调试代码时，我注意到断言在第二次/第三次发出发生之前被调用，这解释了为什么我总是在列表中只有 1 项，但我不知道为什么会发生。

Answer 1

StateFlow是一个可观察的数据持有者，可以将其收集到 观察它随着时间的推移作为流保存的值。请注意，这 值流被合并，这意味着如果值设置在 StateFlow 速度很快，该 StateFlow 的收集器不能保证 接收所有中间值，仅接收最新的值。

您可以使用 FakeRepository，当用这个 fake 测试 ViewModel 时，您可以从 fake 中发出值来触发 ViewModel 的 StateFlow 中的更新，然后对更新后的值进行断言。

检查此链接。

更新 你应该像这样使用你的 TestState 类。

sealed class TestState {
    data class Success(val data: Any): TestState()
    data class Error(val error: String): TestState() //This is optional in your case.
    object Loading: TestState()
}

Answer 2

也许我找到了一个解决方案，我不会说这是最好的，但至少，它达到了我想要的效果。

class TestCollector<T>(
        scope: TestScope,
        numberOfEmit: Int,
        flow: StateFlow<T>,
        private val timeOut: Long,
        private val timeUnit: TimeUnit
) {
    private val values = mutableListOf<T>()
    private val countDownLatch = CountDownLatch(1)

    @OptIn(ExperimentalCoroutinesApi::class)
    private val job = scope.launch(UnconfinedTestDispatcher((scope.testScheduler))) {
        flow.collect {
            values.add(it)
            if (values.size == numberOfEmit) {
                countDownLatch.countDown()
                this.cancel()
            }
        }
    }

    fun getData(): List<T> {
        if (!countDownLatch.await(timeOut, timeUnit)) {
            job.cancel()
            throw TimeoutException("StateFlow value was never set.")
        }
        return values
    }
}

fun <T> StateFlow<T>.test(
        scope: TestScope,
        numberOfEmit: Int = 1,
        timeOut: Long = 10,
        timeUnit: TimeUnit = TimeUnit.SECONDS,
): TestCollector<T> {
    return TestCollector(
            scope = scope,
            numberOfEmit = numberOfEmit,
            timeOut = timeOut,
            timeUnit = timeUnit,
            flow = this
    )
}

用途：

    @Test
    fun testEmit() = runTest {
        val state = viewModel.state

        val testCollector = state.test(this, numberOfEmit = 4)
        viewModel.setEvent(TestEvent.GetData)
        val result = testCollector.getData()

        result.size shouldBe 4
    }

在我的例子中，结果大小为 4，因为它也收集了初始值。