RxJava中类似TestCoroutineScope暂停和恢复的解决方法

Question

完整的源代码可在以下位置获得：https://github.com/AliRezaeiii/StarWarsSearch-RxPaging

这是我的本地单元测试，我在其中测试 ViewModel，同时使用协程进行网络连接：

@Test
fun givenServerResponse200_whenFetch_shouldReturnSuccess() {
    testCoroutineRule.runBlockingTest {
        `when`(api.fetchShowList()).thenReturn(emptyList())
    }

    val repository = ShowRepository(dao, api, context, TestContextProvider())

    testCoroutineRule.pauseDispatcher()

    val viewModel = MainViewModel(repository)

    assertThat(viewModel.shows.value, `is`(Resource.loading()))

    testCoroutineRule.resumeDispatcher()

    assertThat(viewModel.shows.value, `is`(Resource.success(emptyList())))
}

如您所知，我可以使用 TestCoroutineScope 暂停和恢复，因此我可以测试 liveData 何时处于加载或成功状态。

我想知道我们在使用 RxJava 进行测试时是否可以做同样的事情。

目前我只能验证成功状态:

@Test
fun givenServerResponse200_whenFetch_shouldReturnSuccess() {
    `when`(repository.getSpecie(anyString())).thenReturn(Single.just(specie))
    `when`(repository.getPlanet(anyString())).thenReturn(Single.just(planet))
    `when`(repository.getFilm(anyString())).thenReturn(Single.just(film))

    viewModel = DetailViewModel(schedulerProvider, character, 
               GetSpecieUseCase(repository), GetFilmUseCase(repository))

    viewModel.liveData.value.let {
        assertThat(it, `is`(notNullValue()))
        if (it is Resource.Success) {
            it.data?.let { data ->
                assertTrue(data.films.isNotEmpty())
                assertTrue(data.species.isNotEmpty())
            }
        }
    }
}

在 ViewModel 初始化块中，我发送了网络请求。您可以在下方查看 class。这可以在使用协程时使用暂停和恢复进行测试。 RxJava 怎么样？

open class BaseViewModel<T>(
    private val schedulerProvider: BaseSchedulerProvider,
    private val singleRequest: Single<T>
) : ViewModel() {

    private val compositeDisposable = CompositeDisposable()

    private val _liveData = MutableLiveData<Resource<T>>()
    val liveData: LiveData<Resource<T>>
        get() = _liveData

    init {
        sendRequest()
    }

    fun sendRequest() {
        _liveData.value = Resource.Loading
        singleRequest.subscribeOn(schedulerProvider.io())
            .observeOn(schedulerProvider.ui()).subscribe({
                _liveData.postValue(Resource.Success(it))
            }) {
                _liveData.postValue(Resource.Error(it.localizedMessage))
                Timber.e(it)
            }.also { compositeDisposable.add(it) }
    }

    override fun onCleared() {
        super.onCleared()
        compositeDisposable.clear()
    }
}

Answer 1

没有看到您的尝试，我只能猜测有两个可能的问题需要修复：

对所有提供程序方法使用相同的TestScheduler：

class ImmediateSchedulerProvider : BaseSchedulerProvider {

    val testScheduler = TestScheduler()

    override fun computation(): Scheduler = testScheduler

    override fun io(): Scheduler = testScheduler

    override fun ui(): Scheduler = testScheduler
}

单元测试并没有因为错误的状态而失败，所以即使代码没有，它们似乎也通过了运行:

@Test
fun givenServerResponse200_whenFetch_shouldReturnSuccess() {
    `when`(repository.getSpecie(anyString())).thenReturn(Single.just(specie))
    `when`(repository.getPlanet(anyString())).thenReturn(Single.just(planet))
    `when`(repository.getFilm(anyString())).thenReturn(Single.just(film))

    viewModel = DetailViewModel(schedulerProvider, character, GetSpecieUseCase(repository),
            GetPlanetUseCase(repository), GetFilmUseCase(repository))

    viewModel.liveData.value.let {
        assertThat(it, `is`(Resource.Loading))
    }

    schedulerProvider.testScheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS) // <-------------

    viewModel.liveData.value.let {
        assertThat(it, `is`(notNullValue()))
        if (it is Resource.Success) {
            it.data?.let { data ->
                assertTrue(data.films.isNotEmpty())
                assertTrue(data.species.isNotEmpty())
            }
        } else {
            fail("Wrong type " + it)  // <---------------------------------------------
        }
    }
}

RxJava中类似TestCoroutineScope暂停和恢复的解决方法

Solution to pause and resume in RxJava similar to TestCoroutineScope

android

unit-testing

rx-java

rx-java2

kotlin-coroutines