Solution to pause and resume in RxJava similar to TestCoroutineScope

Question

Full source code is available at : https://github.com/AliRezaeiii/StarWarsSearch-RxPaging

Here is my local unit test where I test a ViewModel while I am using Coroutines for networking :

@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())))
}

As you know I can pause and resume using TestCoroutineScope, so I can test when liveData is in Loading or Success state.

I wonder if we can do the same thing when we test while we are using RxJava.

At the moment I just can verify Success state :

@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())
            }
        }
    }
}

in ViewModel init block, I send the network request. You can review it in the bellow class. That can be tested using pause and resume while using Coroutines. How about 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

Without seeing what you tried, I can only guess there were two possible issues that required fixing:

1. Use the same TestScheduler for all provider methods:

   class ImmediateSchedulerProvider : BaseSchedulerProvider {
   
       val testScheduler = TestScheduler()
   
       override fun computation(): Scheduler = testScheduler
   
       override fun io(): Scheduler = testScheduler
   
       override fun ui(): Scheduler = testScheduler
   }
   

2. The unit tests weren't failing for the wrong state so they appear to pass even when the code hasn't run:

   @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)  // <---------------------------------------------
           }
       }
   }