How to run an integration test for the Repository layer of an Android app with Firestore as a backend using Flow

I am currently trying to write an integration test for my repository layer that tests if I call a method, getExercises(), then it returns List<Exercise>, provided that the data is loaded into the local Firestore emulator ahead of time.

So far I got the local Firestore emulator to switch on and off at the beginning/end of a test run, respectively. I am able to populate my data into Firestore, and see the data in the local Firestore emulator via the web UI.

My problem is that my test assertion times out because the Task (an asynchronous construct the Firestore library uses), blocks the thread at the await() part in the repository method.

Test

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

Repository (System under test)

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

The output of this results in:

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

"Exercise count: 3" message never prints!

Note: I am using Robolectric 4.6.1, kotlinx-coroutines-playservices (1.5.0) to provide the await() extension function, and the Turbine testing library for flow assertions (0.6.1)

Perhaps of relevance is a superclass this test inherits that sets the main dispatcher to a test dispatcher.

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

Any help here would be greatly appreciate.

Edit I have opened an issue with the kotlin extensions team to get more visibility on how to go about testing this, including a repo demonstrating the problem.

This problem has been resolved in a new version of the kotlinx-coroutines package (1.6.0-RC). See my github compare across branches. Tests now pass as expected with this version.