android - 如何使用 Flow 为以 Firestore 作为后端的 Android 应用程序的存储库层运行集成测试

Question

我目前正在尝试为我的存储库层编写一个集成测试，以测试我是否调用了一个方法 getExercises() ，然后返回 List<Exercise> ，前提是数据提前加载到本地 Firestore 模拟器中。
到目前为止，我让本地 Firestore 模拟器分别在测试运行的开始/结束时打开和关闭。我能够将我的数据填充到 Firestore，并通过 Web UI 在本地 Firestore 模拟器中查看数据。
我的问题是我的测试断言超时，因为 Task (Firestore 库使用的异步构造)，阻塞 await() 处的线程存储库方法的一部分。
测试

package com.example.fitness.data

import androidx.test.ext.junit.runners.AndroidJUnit4
import app.cash.turbine.test
import com.example.fitness.Constants.EXERCISES_REF
import com.example.fitness.FirebaseEmulatorTest
import com.google.android.gms.tasks.Tasks
import dagger.hilt.android.testing.HiltAndroidRule
import dagger.hilt.android.testing.HiltAndroidTest
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import org.hamcrest.CoreMatchers.`is`
import org.hamcrest.MatcherAssert.assertThat
import org.junit.Before
import org.junit.Rule
import org.junit.Test
import org.junit.runner.RunWith
import javax.inject.Inject
import kotlin.time.Duration
import kotlin.time.ExperimentalTime

@HiltAndroidTest
@RunWith(AndroidJUnit4::class)
class ExerciseRepositoryTest : FirebaseEmulatorTest() {
    @get:Rule
    var hiltRule = HiltAndroidRule(this)

    @Inject
    lateinit var subject: ExerciseRepository

    @Before
    fun setup() {
        hiltRule.inject()
    }

    @ExperimentalTime
    @Test
    fun `#getExercises returns a flow of exercises`() = runBlocking {
        val exercises = mutableListOf<Exercise>().apply {
            add(Exercise("a", "pushups"))
            add(Exercise("b", "pull-ups"))
            add(Exercise("c", "sit-ups"))
        }

        runBlocking(Dispatchers.IO) {
            val task1 = firestoreInstance.collection(EXERCISES_REF).add(exercises.first())
            val task2 = firestoreInstance.collection(EXERCISES_REF).add(exercises[1])
            val task3 = firestoreInstance.collection(EXERCISES_REF).add(exercises.last())

            Tasks.await(task1)
            Tasks.await(task2)
            Tasks.await(task3)

            println("Done with tasks: task1: ${task1.isComplete}. task2: ${task2.isComplete}. task3: ${task3.isComplete}.")
        }

        println("About to get exercises")

        subject.getExercises().test(timeout = Duration.seconds(5)) {
            println("test body")

            assertThat(awaitItem().size, `is`(4)) // Just checking that it passes for the right reasons first. This number should be 3
        }
    }
}

存储库(被测系统)

package com.example.fitness.data

import com.example.fitness.Constants.EXERCISES_REF
import com.google.firebase.firestore.CollectionReference
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.tasks.await
import javax.inject.Inject
import javax.inject.Named
import javax.inject.Singleton

@Singleton
class ExerciseRepository @Inject constructor(
    @Named(EXERCISES_REF) private val exerciseCollRef: CollectionReference
) {
    fun getExercises() = flow<List<Exercise>> {
        println("beginning of searchForExercise")

        val exercises = exerciseCollRef.limit(5).get().await() // NEVER FINISHES!!
        println("Exercise count: ${exercises.documents}")

        emit(exercises.toObjects(Exercise::class.java))
    }
}

其输出结果为:

Done with tasks: task1: true. task2: true. task3: true.
About to search for exercises
beginning of searchForExercise
test body

Timed out waiting for 5000 ms
kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 5000 ms

“锻炼次数:3”消息永远不会打印!
注意:我使用的是 Robolectric 4.6.1， kotlinx-coroutines-playservices (1.5.0) 提供 await()扩展函数，以及用于流断言的涡轮测试库 (0.6.1)
也许相关的是这个测试继承的一个父类(super class)，它将主调度程序设置为一个测试调度程序。

package com.example.fitness

import androidx.arch.core.executor.testing.InstantTaskExecutorRule
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.test.*
import org.junit.After
import org.junit.Before
import org.junit.Rule

abstract class CoroutineTest {
    @Rule
    @JvmField
    val rule = InstantTaskExecutorRule()

    protected val testDispatcher: TestCoroutineDispatcher = TestCoroutineDispatcher()
    private val testCoroutineScope = TestCoroutineScope(testDispatcher)

    @Before
    fun setupViewModelScope() {
        Dispatchers.setMain(testDispatcher)
    }

    @After
    fun cleanupViewModelScope() {
        Dispatchers.resetMain()
    }

    @After
    fun cleanupCoroutines() {
        testDispatcher.cleanupTestCoroutines()
        testDispatcher.resumeDispatcher()
    }

    fun runBlockingTest(block: suspend TestCoroutineScope.() -> Unit) =
        testCoroutineScope.runBlockingTest(block)
}

这里的任何帮助将不胜感激。
编辑
我开了一个 issue与 kotlin 扩展团队一起了解如何进行测试，包括 repo证明问题。

Answer 1

此问题已在 kotlinx-coroutines 的新版本中得到解决。包(1.6.0-RC)。 See my github compare跨分支。测试现在按预期通过此版本。