kotlinx.coroutines.test 1.6.x + StateFlowの状態遷移を検証するためのヘルパー

TestScopeExtensions.kt

package co.uzzu.coroutines.testing

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.advanceUntilIdle

@OptIn(ExperimentalCoroutinesApi::class)
inline fun <T> TestScope.withCollectState(
    stateFlow: StateFlow<T>,
    destination: MutableList<T>,
    block: () -> Unit,
) {
    withCollectState(
        flowSets = arrayOf(TestingStateFlowSet(stateFlow, destination)),
        block = block,
    )
}

@OptIn(ExperimentalCoroutinesApi::class)
inline fun TestScope.withCollectState(
    vararg flowSets: TestingStateFlowSet<*>,
    block: () -> Unit,
) {
    val scopes = mutableListOf<CoroutineScope>()
    val dispatcher = UnconfinedTestDispatcher()
    try {
        flowSets.forEach {
            val coroutineScope = CoroutineScope(dispatcher)
            it.collectIn(coroutineScope)
            scopes.add(coroutineScope)
        }
        block()
        advanceUntilIdle()
    } finally {
        scopes.forEach { it.cancel() }
    }
}


class TestingStateFlowSet<T>(
    private val stateFlow: StateFlow<T>,
    private val destination: MutableList<T>,
) {
    fun collectIn(scope: CoroutineScope): Job =
        stateFlow.onEach { destination.add(it) }
            .launchIn(scope)
}