Android ViewModel协程：为Dispatcher.IO写单元测试遇执行提前结束问题

核心问题：编写单元测试调用ViewModel的网络请求方法后，测试直接通过，DataSource中response.isSuccessful代码行从未执行，测试在异步任务完成前就结束了，无法验证LiveData的结果。

单元测试类 #

@RunWith(AndroidJUnit4::class)
class TestViewModel {

    private lateinit var mockWebServer: MockWebServer
    private val client = OkHttpClient.Builder()
        .connectTimeout(1, TimeUnit.SECONDS)
        .readTimeout(1, TimeUnit.SECONDS)
        .writeTimeout(1, TimeUnit.SECONDS)
        .build()

    private lateinit var api: TPSCallService

    @OptIn(ExperimentalCoroutinesApi::class)
    @Before
    fun setUp()  {
        mockWebServer = MockWebServer()
        api = Retrofit.Builder()
            .baseUrl(mockWebServer.url("/"))
            .client(client)
            .addConverterFactory(GsonConverterFactory.create())
            .build()
            .create(MyAPI::class.java)
    }

    @OptIn(ExperimentalCoroutinesApi::class)
    @Test
    fun testing() = runTest {
        val mockResponse = MockResponse()
        val inputStream = javaClass.classLoader?.getResourceAsStream("personalDetailsAPI/success.json")
        val source = inputStream?.let { inputStream.source().buffer() }
        mockResponse.setResponseCode(200).setBody(source!!.readString(StandardCharsets.UTF_8))
        mockWebServer.enqueue(mockResponse)

        val dataSource = MyDataSource(api)
        val vm = MainViewModel(dataSource)
        vm.getMyList()
        vm.dataList.observeForever() {
            Assert.assertEquals("",3, it.size)
        }
    }
}

ViewModel类 #

class MainViewModel @Inject constructor(): ViewModel() {

    constructor(myDataSource: MyDataSource): this() {
        this.storeDataSource = storeDataSource
    }

    companion object {
        private const val TAG = "MainViewModel_TAG"

        val job = SupervisorJob()
        var dispatcher = CoroutineScope(Dispatchers.IO + job)
    }

    private var myDataSource: MyDataSource = MyDataSource()

    private val _dataList = MutableLiveData<List<DataResponse>>()
    public val dataList: LiveData<List<DataResponse>>
        get() = _dataList

    public fun getMyList() {
        dispatcher.launch {
            myDataSource.getStoreList(Constants.DEFAULT_LATITUDE, Constants.DEFAULT_LONGITUDE) { dataLists, code ->
                _dataList.postValue(dataLists.getOrNull())// 从未执行到这里！
            }
        }
    }
}

class MainViewModelFactory() :
    ViewModelProvider.NewInstanceFactory() {
    override fun <T : ViewModel> create(modelClass: Class<T>): T {
        return modelClass.getConstructor()
            .newInstance()
    }
}

class MainViewModelFactoryTwo(val dataSource: MyDataSource) :
    ViewModelProvider.NewInstanceFactory() {
    override fun <T : ViewModel> create(modelClass: Class<T>): T {
        return modelClass.getConstructor(MyDataSource::class.java)
            .newInstance(dataSource)
    }
}

DataSource类 #

class MyDataSource {

    private lateinit var myAPI: MyAPI

    constructor() {
        myAPI = NetworkModule().provideMyAPI() // dagger提供Retrofit实例的方法
    }

    constructor(myAPI: MyAPI) {
        this.myAPI = myAPI
    }

    public suspend fun getStoreList(lat: Double, lng: Double, completion: (Result<List<StoreResponse>>, Int) -> Unit) {
        val response = myAPI.getMyList(lat, lng).awaitResponse()

        if (response.isSuccessful) { // 断点到不了这里！测试在执行前就结束了
            response.body()?.let {
                completion(Result.success(it), response.code())
            } ?: run {
                completion(Result.failure(Throwable()), response.code())
            }
        } else {
            completion(Result.failure(Throwable()), response.code())
        }
    }
}

1. 改造ViewModel，支持Dispatcher注入 #

ViewModel中硬编码的Dispatchers.IO导致测试无法控制协程执行时机，需将Dispatcher设计为可注入参数：

class MainViewModel @Inject constructor(
    private val myDataSource: MyDataSource,
    private val dispatcher: CoroutineDispatcher = Dispatchers.IO
): ViewModel() {
    // 移除companion中硬编码的dispatcher，改用ViewModel级别的协程作用域
    private val viewModelScope = CoroutineScope(SupervisorJob() + dispatcher)

    private val _dataList = MutableLiveData<List<DataResponse>>()
    public val dataList: LiveData<List<DataResponse>>
        get() = _dataList

    public fun getMyList() {
        viewModelScope.launch {
            myDataSource.getStoreList(Constants.DEFAULT_LATITUDE, Constants.DEFAULT_LONGITUDE) { dataLists, code ->
                _dataList.postValue(dataLists.getOrNull())
            }
        }
    }

    // 保留测试用构造函数
    constructor(myDataSource: MyDataSource) : this(myDataSource, Dispatchers.IO)
}

2. 用测试Dispatcher控制协程同步执行 #

在测试中使用UnconfinedTestDispatcher注入ViewModel，确保协程在测试线程同步执行，避免异步逃逸：

@OptIn(ExperimentalCoroutinesApi::class)
@Test
fun testing() = runTest {
    val mockResponse = MockResponse()
    val inputStream = javaClass.classLoader?.getResourceAsStream("personalDetailsAPI/success.json")
    val source = inputStream?.let { inputStream.source().buffer() }
    mockResponse.setResponseCode(200).setBody(source!!.readString(StandardCharsets.UTF_8))
    mockWebServer.enqueue(mockResponse)

    val testDispatcher = UnconfinedTestDispatcher(testScheduler)
    val dataSource = MyDataSource(api)
    val vm = MainViewModel(dataSource, testDispatcher)
    
    vm.getMyList()
    
    // 直接获取LiveData更新后的值
    val result = vm.dataList.getOrAwaitValue()
    Assert.assertEquals(3, result.size)
}

3. 添加LiveData测试扩展函数 #

如果需要等待LiveData异步更新，可添加以下扩展函数确保测试等待结果：

@OptIn(ExperimentalCoroutinesApi::class)
fun <T> LiveData<T>.getOrAwaitValue(
    time: Long = 2,
    timeUnit: TimeUnit = TimeUnit.SECONDS,
    afterObserve: () -> Unit = {}
): T {
    var data: T? = null
    val latch = CountDownLatch(1)
    val observer = object : Observer<T> {
        override fun onChanged(o: T?) {
            data = o
            latch.countDown()
            this@getOrAwaitValue.removeObserver(this)
        }
    }
    this.observeForever(observer)

    try {
        afterObserve.invoke()
        if (!latch.await(time, timeUnit)) {
            throw TimeoutException("LiveData value was never set.")
        }
    } finally {
        this.removeObserver(observer)
    }

    @Suppress("UNCHECKED_CAST")
    return data as T
}

4. 避免无等待的observeForever断言 #

原测试中observeForever内的断言可能永远不会执行，因为测试线程已提前结束。必须通过主动等待或同步协程执行的方式，确保LiveData更新后再做断言。

内容的提问来源于stack exchange，提问作者TheQ