Kotlin Flow：从不收集 emitAll

Question

我正在尝试为 networkBoundResource 的 kotlin 版本编写一个 UnitTest，可以在 serveral sources 和 several features 上找到它

这是我对以下问题的标记-cmets 版本。

inline fun <ResultType, RequestType> networkBoundResource(
    ...
    coroutineDispatcher: CoroutineDispatcher
) = flow {

    emit(Resource.loading(null)) // emit works!

    val data = queryDatabase().firstOrNull()

    val flow = if (shouldFetch(data)) {
        
        emit(Resource.loading(data)) // emit works!

        try {
            saveFetchResult(fetch())
            query().map { Resource.success(it) }
            
        } catch (throwable: Throwable) {
            onFetchFailed(throwable)
            query().map { Resource.error(throwable.toString(), it) }
            
        }
    } else {
        query().map { Resource.success(it) }
    }

    emitAll(flow) // emitAll does not work!

}.catch { exception ->
    emit(Resource.error("An error occurred while fetching data! $exception", null))

}.flowOn(coroutineDispatcher)

这是我对此代码的单元测试之一。对代码进行了一些编辑以专注于我的问题：

@get:Rule
val testCoroutineRule = TestCoroutineRule()

private val coroutineDispatcher = TestCoroutineDispatcher()

@Test
fun networkBoundResource_noCachedData_shouldMakeNetworkCallAndStoreUserInDatabase() = testCoroutineRule.runBlockingTest {
    ...

    // When getAuthToken is called
    val result = networkBoundResource(..., coroutineDispatcher).toList()

    result.forEach {
        println(it)
    }    
}

问题是println(it) 只打印Resource.loading(null) 排放量。但是如果你看一下flow {} 块的最后一行，你会发现应该有另一个val flow 的发射。但是这种排放永远不会到达我的单元测试中。为什么？

Answer 1

正如@MarkKeen 建议的那样，我现在创建了自己的实现，并且效果很好。与 SO 上的代码相比，这个版本现在注入了 coroutineDispatcher 以便于测试，它让流程负责错误处理，它不包含嵌套流程并且更容易阅读和理解。将更新的数据存储到数据库中仍然存在副作用，但我现在太累了，无法解决这个问题。

import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.flow.*

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType?>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline shouldFetch: (ResultType?) -> Boolean = { true },
    coroutineDispatcher: CoroutineDispatcher
) = flow<Resource<ResultType>> {
    
    // check for data in database
    val data = query().firstOrNull()
    
    if (data != null) {
        // data is not null -> update loading status
        emit(Resource.loading(data))
    }
    
    if (shouldFetch(data)) {
        // Need to fetch data -> call backend
        val fetchResult = fetch()
        // got data from backend, store it in database
        saveFetchResult(fetchResult)        
    }

    // load updated data from database (must not return null anymore)
    val updatedData = query().first()

    // emit updated data
    emit(Resource.success(updatedData))    
    
}.onStart { 
    emit(Resource.loading(null))
    
}.catch { exception ->
    emit(Resource.error("An error occurred while fetching data! $exception", null))
    
}.flowOn(coroutineDispatcher)

这个内联乐趣的一个可能的单元测试，用于AuthRepsitory：

@ExperimentalCoroutinesApi
class AuthRepositoryTest {

    companion object {
        const val FAKE_ID_TOKEN = "FAkE_ID_TOKEN"
    }

    @get:Rule
    val testCoroutineRule = TestCoroutineRule()

    private val coroutineDispatcher = TestCoroutineDispatcher()

    private val userDaoFake = spyk<UserDaoFake>()

    private val mockApiService = mockk<MyApi>()

    private val sut = AuthRepository(
        userDaoFake, mockApiService, coroutineDispatcher
    )

    @Before
    fun beforeEachTest() {
        userDaoFake.clear()
    }

    @Test
    fun getAuthToken_noCachedData_shouldMakeNetworkCallAndStoreUserInDatabase() = testCoroutineRule.runBlockingTest {
        // Given an empty database
        coEvery { mockApiService.getUser(any()) } returns NetworkResponse.Success(UserFakes.getNetworkUser(), null, HttpURLConnection.HTTP_OK)

        // When getAuthToken is called
        val result = sut.getAuthToken(FAKE_ID_TOKEN).toList()

        coVerifyOrder {
            // Then first try to fetch data from the DB
            userDaoFake.get()

            // Then fetch the User from the API
            mockApiService.getUser(FAKE_ID_TOKEN)

            // Then insert the user into the DB
            userDaoFake.insert(any())

            // Finally return the inserted user from the DB
            userDaoFake.get()
        }
        
        assertThat(result).containsExactly(
            Resource.loading(null),
            Resource.success(UserFakes.getAppUser())
        ).inOrder()
    }
}

Answer 2

我不太确定完整的行为，但本质上你想获得资源，并且当前流量都集中在 FlowCollector<T> 中，这使得推理和测试变得更加困难。

我以前从未使用或看过 Google 代码，老实说，我只是看了一眼。我的主要收获是它的封装很差，并且似乎打破了关注点的分离——它管理资源状态，并处理所有 io 工作一类。我希望有 2 个不同的类来分隔该逻辑并允许更轻松的测试。

作为简单的伪代码我会做这样的事情：

class ResourceRepository {

    suspend fun get(r : Request) : Resource {
        // abstract implementation details network request and io 
        // - this function should only fulfill the request 
        // can now be mocked for testing
        delay(3_000)
        return Resource.success(Any())
    }
}

data class Request(val a : String)

sealed class Resource {

    companion object {
        val loading : Resource get() = Loading
        fun success(a : Any) : Resource = Success(a)
        fun error(t: Throwable) : Resource = Error(t)
    }

    object Loading : Resource()

    data class Success(val a : Any) : Resource()

    data class Error(val t : Throwable) : Resource()
}

fun resourceFromRequest(r : Request) : Flow<Resource> =
    flow { emit(resourceRepository.get(r)) }
        .onStart { emit(Resource.loading) }
        .catch { emit(Resource.error(it)) }

这使您可以大大简化resourceFromRequest() 函数的实际测试，因为您只需要模拟存储库和一种方法。这使您可以抽象和处理其他地方的网络和 io 工作，而这些工作又可以单独进行测试。