Unit Test for RxJava and Retrofit

Question

I have This method that calls a Rest API and returns the result as an Observable (Single):

fun resetPassword(email: String): Single<ResetPassword> {
    return Single.create { emitter ->

        val subscription = mApiInterfacePanda.resetPassword(email)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe({ resetPasswordResponse ->
                when(resetPasswordResponse.code()) {
                    200 ->  {
                        resetPasswordResponse?.body()?.let { resetPassword ->
                            emitter.onSuccess(resetPassword)
                        }
                    }
                    else -> emitter.onError(Exception("Server Error"))
                }
            }, { throwable ->
                    emitter.onError(throwable)
            })

        mCompositeDisposable.add(subscription)

    }
}

Unit Test:

@Test
fun resetPassword_200() {
    val response = Response.success(200, sMockResetPasswordResponse)
    Mockito.`when`(mApiInterfacePanda.resetPassword(Mockito.anyString()))
        .thenReturn(Single.just(response))

    mTokenRepository.resetPassword(MOCK_EMAIL)

    val observer = mApiInterfacePanda.resetPassword(MOCK_EMAIL)
    val testObserver = TestObserver.create<Response<ResetPassword>>()
    observer.subscribe(testObserver)

    testObserver.assertSubscribed()
    testObserver.awaitCount(1)
    testObserver.assertComplete()
    testObserver.assertResult(response)
}

My Problem is only this line gets covered and the other lines won't run and that has a lot of impact on my total test coverage:

return Single.create { emitter ->

Stalkinghorse

1 2 3 – Anaximander

@sonnet thanks for sharing but I have searched the internet and I also visited some StackOverflow pages, but I haven't find any answer for my particular case. – Stalkinghorse

Is the problem not getting more test coverage or that the above code doesn't work? – Anaximander

@sonnet It passes the test, the line coverage is only on Single.create { } and it shows something is wrong how I handle my Unit Test. – Stalkinghorse

your val observer should be mTokenRepository.resetPassword(MOCK_EMAIL) since that's the thing you're observing and trying to test, not the details inside it which is mApiInterfacePanda.resetPassword(MOCK_EMAIL). Also, remove the Single.create { } construct. That's a bridge between reactive and callback styles. You can simply return mApiInterfacePanda.resetPassword(MOCK_EMAIL) from repository.resetPassword() func. I'd assume the emitter.onSuccess() is not being triggered somehow – Anaximander

Answer 1

There's more than one thing going on here if I'm not mistaken. Let's take it in parts.

First, your "internal" observer:

mApiInterfacePanda.resetPassword(email)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe({ resetPasswordResponse -> ... })

Is observing on the android main thread and executing on a background thread. To the best of my knowledge, in most cases, the test thread will end before your mApiInterfacePanda .resetPassword has a chance to finish and run. You didn't really post the test setup, so I'm not sure if this is an actual issue, but in any case it's worth mentioning. Here's 2 ways to fix this:

RxJavaPlugins and RxAndroidPlugins

RxJava already provides a way to change the schedulers that are provided. An example is RxAndroidPlugins.setMainThreadSchedulerHandler. Here's how it could help:

@Before
fun setUp() {
   RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }
   RxJavaPlugins.setInitIoSchedulerHandler { Schedulers.trampoline() }
}

The above methods make sure that everywhere you use the main thread scheduler and the io scheduler, it'll instead return the trampoline scheduler. This is a scheduler that guarantees that the code is executed in the same thread that was executing previously. In other words, it'll make sure you run it on the unit test main thread.

You will have to undo these:

@After
fun tearDown() {
   RxAndroidPlugins.reset()
   RxJavaPlugins.reset()
}

You can also change other schedulers.

Inject the schedulers

You can use kotlin's default arguments to help out with injecting schedulers:

fun resetPassword(
  email: String, 
  obsScheduler: Scheduler = AndroidSchedulers.mainThread(),
  subScheduler: Scheduler = Schedulers.io()
): Single<ResetPassword> {
   return Single.create { emitter ->

     val subscription = mApiInterfacePanda.resetPassword(email)
        .observeOn(obsScheduler)
        .subscribeOn(subScheduler)
        .subscribe({ resetPasswordResponse ->
            when(resetPasswordResponse.code()) {
                200 ->  {
                    resetPasswordResponse?.body()?.let { resetPassword ->
                        emitter.onSuccess(resetPassword)
                    }
                }
                else -> emitter.onError(Exception("Server Error"))
            }
        }, { throwable ->
                emitter.onError(throwable)
        })

    mCompositeDisposable.add(subscription)
  }
}

At test time you can just call it like resetPassword("[email protected]", Schedulers.trampoline(), Schedulers.trampoline() and for the application just pass in the email.

---

The other thing I see here is maybe not related to the problem, but I think it's still good to know. First, you're creating a single, but you don't need to do this.

Single.create is usually used when you don't have reactive code. However, mApiInterfacePanda.resetPassword(email) already returns a reactive component and although I'm not sure, let's just assume it's a single. If not, it should be fairly simple to convert it to something else.

You're also holding on to a disposable, which from what I can tell shouldn't be necessary.

Lastly, you're using retrofit according to your tags so you don't need to make the call return a raw response unless extremely necessary. This is true because retrofit checks the status code for you and will deliver the errors inside onError with an http exception. This is the Rx way of handling the errors.

With all this in mind, I'd rewrite the entire method like this:

fun resetPassword(email: String) = mApiInterfacePanda.resetPassword(email)

(note that resetPassword must not return a raw response, but Single<ResetPassword>

It actually shouldn't need anything else. Retrofit will make sure things end up in either onSuccess or onError. You don't need to subscribe to the result of the api here and handle disposables - let whoever is calling this code handle it.

You may also notice that if this is the case, then the solution for the schedulers is not needed. I guess this is true in this case, just remember some operators operate in some default schedulers and you may need to override them in some cases.

---

So how would I test the above method?

Personally I'd just check if the method calls the api with the right parameters:

@Test
fun resetPassword() {
   mTokenRepository.resetPassword(MOCK_EMAIL)

   verify(mApiInterfacePanda).resetPassword(MOCK_EMAIL)
}

I don't think there's much more needed here. There's no more logic I can see in the rewritten method.

Pincers

Still learning the ropes for testing Rx and this is great explanation. Much thanks to you, Fred. – Dynamometry