Test methods of Room DAO with Kotlin Coroutines and Flow

Question

I am trying to migrate from LiveData to Flow in my Room Dao. App is working fine, but I have problems with testing behavior. When I run the test it is starting and running indefinately. I also tried to use kotlinx.coroutines.test runBlockingTest, but I had issue with "This job has not finished yet" like here. Can someone point me in right direction how to test behavior of my CoresDao?

@Dao
interface CoresDao {

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    suspend fun insertCores(cores: List<Core>)

    @Transaction
    suspend fun replaceCoresData(cores: List<Core>) {
        deleteAllCores()
        insertCores(cores)
    }

    @Query("SELECT * FROM cores_table")
    fun getAllCores(): Flow<List<Core>>

    @Query("DELETE FROM cores_table")
    suspend fun deleteAllCores()
}

@RunWith(AndroidJUnit4::class)
class CoresDaoTest {

    private lateinit var database: SpaceDatabase
    private lateinit var coresDao: CoresDao

    private val testDispatcher = TestCoroutineDispatcher()

    private val testCoresList = listOf(core2, core3, core1)

    @get:Rule
    var instantTaskExecutorRule = InstantTaskExecutorRule()

    @Before
    fun setup() {
        Dispatchers.setMain(testDispatcher)

        val context = InstrumentationRegistry.getInstrumentation().targetContext
        database = Room.inMemoryDatabaseBuilder(context, SpaceDatabase::class.java).build()
        coresDao = database.coresDao()
    }

    @After
    fun cleanup() {
        database.close()

        Dispatchers.resetMain()
        testDispatcher.cleanupTestCoroutines()
    }

    @Test
    fun testGetAllCores(): Unit = runBlocking {
        withContext(Dispatchers.Main) {
            runBlocking { coresDao.insertCores(testCoresList) }

            val coresList = mutableListOf<Core>()
            coresDao.getAllCores().collect { cores -> coresList.addAll(cores) }

            assertThat(coresList.size, equalTo(testCoresList.size))
        }
    }
}

Illustrational

Answer 1

To test Flow, the best APIs I found are .take(n).toList(). You can use runBlockingTest and you shouldn't need to use withContext to move the execution to another thread.

You can find an example of how it works here: https://github.com/manuelvicnt/MathCoroutinesFlow/blob/master/app/src/test/java/com/manuelvicnt/coroutinesflow/fibonacci/impl/NeverEndingFibonacciProducerTest.kt#L38

Subchloride

This may be a dumb question (I'm a Flow n00b), but is take(n) for how many items you want from the FLOW itself, or from a LIST you may have in the Flow? Like my DAO method returns a Flow<List<MyObject>>, so if I do take(1) would that just retrieve my List out of the Flow? – Spritsail

Answer 2

Since you are using the TestCoroutineDispatcher already, using runBlockingTest won't do anything in your example.. You will have to cancel the Flow after collection or the scope in which you launched the Flow

edit: an example for such a rule can be found here

In

The link is gone but the author has extracted out the rule into this library to test Flows. – Wonderland

Answer 3

Turns out that I wasn't properly handle Flow collection and cancelation and that was probably the cause of problem. Below is the code that works. More complex example can be found here.

@RunWith(AndroidJUnit4::class)
class CoresDaoTest {

    private lateinit var database: SpaceDatabase
    private lateinit var coresDao: CoresDao

    private val testDispatcher = TestCoroutineDispatcher()

    private val testCoresList = listOf(core2, core3, core1)

    @get:Rule
    var instantTaskExecutorRule = InstantTaskExecutorRule()

    @Before
    fun setup() {
        Dispatchers.setMain(testDispatcher)

        val context = InstrumentationRegistry.getInstrumentation().targetContext
        database = Room
            .inMemoryDatabaseBuilder(context, SpaceDatabase::class.java)
            .setTransactionExecutor(Executors.newSingleThreadExecutor())
            .build()
        coresDao = database.coresDao()
    }

    @After
    fun cleanup() {
        database.close()

        Dispatchers.resetMain()
        testDispatcher.cleanupTestCoroutines()
    }

    @Test
    fun testInsertAndGetAllCores() = runBlocking {
        coresDao.insertCores(testCoresList)

        val latch = CountDownLatch(1)
        val job = launch(Dispatchers.IO) {
            coresDao.getAllCores().collect { cores ->
                assertThat(cores.size, equalTo(testCoresList.size))
                latch.countDown()
            }
        }

        latch.await()
        job.cancel()
    }

Illustrational

While this works, the test will still run indefinitely if the code inside collect throws any error, like for example if the assert fails. – Wonderland

Thank You for pointing this out. Currently I am using method from accepted answer, which seems fine. – Illustrational

My test, just like yours, just keeps spinning. In the collect { } mine just asserts that the List isn't empty and then calls countDown(). Then after that the await() and cancel() calls. That method above is the same as what's in the more complex example, well one of the comlex example's functions anyway. One of them has a produceIn() call instead. No clue what the difference is or what I should use. I should just convert my app to just return List instead of this Flow crap. – Spritsail