How to unit test your RxJava code in Kotlin

In a typical MVP architecture you have different layers: a repository/network client to fetch data, a view with attached a presenter that will call an interactor to proxy the request to the repository/network layer. And if you use RxJava you probably will return an observable from the repository layer, subscribe to it in the presenter layer and the interactor will decide which thread will the request be executed.

That’s pretty common, I would also argue that it’s better to have two different models: one for the response from the network and one for the UI. The benefit of two separated models is usually to prevent changes across the entire code base: if the server changes the response you can just map the new one to the UI model and the entire view/presenter layer doesn’t need to change at all. It’s also nice to have different name fields between the two layers so you don’t need to choose the names based on the API definition. For example I think image is a better name than backdrop_path for the example that I’m gonna show you soon.

The main point of this post is to show you how to write an interactor that fetches data from a client on a thread and returns an observable on the main thread.
Let’s imagine to have a network client that fetches a list of movies, the response from the network is:

class MoviesResponse(val page: Int, 
                     val total_results: Int, 
                     val total_pages: Int, 
                     val results: List<MovieResponse>)

class MovieResponse(val id: String, 
                    val title: String, 
                    val backdrop_path: String, 
                    val overview: String, 
                    val release_date: String)

It’s pretty simple, a MoviesResponse contains the number of results, the number of pages, the current page and a list of movies; each movie contains an id, a title, a backdrop path, an overview and a release date.

On the UI side we are just interested in showing a list of movies in a list, so we are expecting a title, a description and an image:

data class Movie(val title: String, val text: String, val _image: String){
    val image: String
        get() = "http://www.base.url/$_image"
}

So let’s write an interactor that call the network client, retrieve the MoviesResponse and returns an observable with a list of Movie objects.

class MoviesInteractor(val client: MoviesClient) {

    fun requestTopMovies(): Observable<List<Movie>> {
        return client.fetchMovies()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map {
                    it.results.map {
                        Movie(it.title, it.overview, it.backdrop_path)
                    }
                }
    }

}

The MoviesInteractor is expecting a MoviesClient, which has one function:

fun fetchMovies(): Observable<MoviesResponse>

If you are using Retrofit, it supports Observable out of the box (adding a plugin), so I will not focus on that right now. Assuming that client.fetchMovies() returns an observable of MoviesResponse, then the interactor can map this response to the list of Movies with this code:

.map {
    it.results.map {
        Movie(it.title, it.overview, it.backdrop_path)
    }
}

It’s very easy with Kotlin to achieve this, because the map operator of RxJava returns the instance of MoviesResponse which has a property ‘results’ that contains a list of MovieResponse. So for each MovieResponse we need to create a correspondent Movie object and luckily Kotlin has a built-in function map that let you map the object MovieResponse to a list of movies, so the ‘it’ instance inside the map function refers to a MovieResponse.

The rest of the interactor defines to observeOn the main thread and subscribe on another thread to avoid blocking the main thread.
Let’s see how to test this:

    @Mock
    lateinit var client: MoviesClient

    internal lateinit var underTest: MoviesInteractor

    @Test
    fun `when movies are requested, should call client and return response`() {
        val movieResponse = MoviesResponse(0, 0, 0, listOf(
                MovieResponse("id1", "title1", "/backdrop1", "Overview1", "release date1"),
                MovieResponse("id2", "title2", "/backdrop2", "Overview2", "release date2")))
        Mockito.`when`(client.fetchMovies()).thenReturn(Observable.just(movieResponse))

        val result = underTest.requestTopMovies()

        val testObserver = TestObserver<List<Movie>>()
        result.subscribe(testObserver)
        testObserver.assertComplete()
        testObserver.assertNoErrors()
        testObserver.assertValueCount(1)
        val listResult = testObserver.values()[0]
        assertThat(listResult.size, `is`(2))
        assertThat(listResult[0].title, `is`("title1"))
        assertThat(listResult[0].image, `is`("http://www.base.url/backdrop1"))
        assertThat(listResult[0].text, `is`("Overview1"))
        assertThat(listResult[1].title, `is`("title2"))
        assertThat(listResult[1].image, `is`("http://www.base.url/backdrop2"))
        assertThat(listResult[1].text, `is`("Overview2"))
    }

In the test we are mocking the response of the client, when the client is called then we return a fake observable that contains a MoviesResponse with two movies. To test the observable we can use the TestObserver class which let you test the content of your subscription: we can check that is complete, no errors occurred and that the value is a list of movies.

If you try to run that test you will likely to run into this error:

Caused by: java.lang.RuntimeException: Method getMainLooper in android.os.Looper not mocked. See http://g.co/androidstudio/not-mocked for details.

It’s a quite cryptic error message, but the reason behind is that your interactor is trying to execute the client call in a different thread than the main thread. During a test execution we are actually not interested in this behaviour, we just want to make sure that the client is called and mapped back to our UI model.
To fix the error message you can add this in your setup method:

RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
RxJavaPlugins.setNewThreadSchedulerHandler { Schedulers.trampoline() }
RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

Those four lines of code let you specify which thread to use for the Io Scheduler, Computation Scheduler, new thread scheduler and the initial main thread scheduler for android; basically you are forcing all of them to use the same thread! But what is a trampoline? From the documentation:

/**
 * Creates and returns a {@link Scheduler} that queues work on the current thread to be executed after the
 * current work completes.
 *
 * @return a {@link Scheduler} that queues work on the current thread
 */

That exactly what we are interested in! We want to the work to be done synchronously because it’s not the job of this unit test to check that we are using different threads. I would also argue that you don’t need to test the different thread behaviour because you are using a library (RxJava) and from my point of view, testing a library is pointless.

That looks great, but you need to put those 4 lines of code in each setup method of each file and it seems quite ripetitive. I would suggest to create a rule that can be added in each tests that need this behaviour:

class RxImmediateSchedulerRule : TestRule {

    override fun apply(base: Statement, d: Description): Statement {
        return object : Statement() {
            @Throws(Throwable::class)
            override fun evaluate() {
                RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
                RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
                RxJavaPlugins.setNewThreadSchedulerHandler { Schedulers.trampoline() }
                RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

                try {
                    base.evaluate()
                } finally {
                    RxJavaPlugins.reset()
                    RxAndroidPlugins.reset()
                }
            }
        }
    }
}

Now instead of the four lines of code in the setup, we can just add a rule:

@Rule @JvmField var testSchedulerRule = RxImmediateSchedulerRule()

And that’s it! The complete implementation of the interactor and its test:

class MoviesInteractorTest {

    @Rule @JvmField
    val rule = MockitoJUnit.rule()!!

    @Rule @JvmField var testSchedulerRule = RxImmediateSchedulerRule()

    @Mock
    lateinit var client: MoviesClient

    internal lateinit var underTest: MoviesInteractor

    @Before
    fun setUp() {
        underTest = MoviesInteractor(client)
    }

    @Test
    fun `when top movies are requested, should call client and return response`() {
        val movieResponse = MoviesResponse(0, 0, 0, listOf(
                MovieResponse("id1", "title1", "/backdrop1", "Overview1", "release date1"),
                MovieResponse("id2", "title2", "/backdrop2", "Overview2", "release date2")))
        Mockito.`when`(client.fetchMovies()).thenReturn(Observable.just(movieResponse))

        val result = underTest.requestTopMovies()

        val testObserver = TestObserver<List<Movie>>()
        result.subscribe(testObserver)
        testObserver.assertComplete()
        testObserver.assertNoErrors()
        testObserver.assertValueCount(1)
        val listResult = testObserver.values()[0]
        assertThat(listResult.size, `is`(2))
        assertThat(listResult[0].title, `is`("title1"))
        assertThat(listResult[0].image, `is`("https://image.tmdb.org/t/p/w342/backdrop1"))
        assertThat(listResult[0].text, `is`("Overview1"))
        assertThat(listResult[1].title, `is`("title2"))
        assertThat(listResult[1].image, `is`("https://image.tmdb.org/t/p/w342/backdrop2"))
        assertThat(listResult[1].text, `is`("Overview2"))
    }
}

class RxImmediateSchedulerRule : TestRule {

    override fun apply(base: Statement, d: Description): Statement {
        return object : Statement() {
            @Throws(Throwable::class)
            override fun evaluate() {
                RxJavaPlugins.setIoSchedulerHandler { Schedulers.trampoline() }
                RxJavaPlugins.setComputationSchedulerHandler { Schedulers.trampoline() }
                RxJavaPlugins.setNewThreadSchedulerHandler { Schedulers.trampoline() }
                RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

                try {
                    base.evaluate()
                } finally {
                    RxJavaPlugins.reset()
                    RxAndroidPlugins.reset()
                }
            }
        }
    }
}

class MoviesInteractorImpl(val client: MoviesClient) {

    override fun requestTopMovies(): Observable<List<Movie>> {
        return client.fetchMovies()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map {
                    it.results.map {
                        Movie(it.title, it.overview, it.backdrop_path)
                    }
                }
    }

}

Blog

How to unit test your RxJava code in Kotlin

Daniele Bottillo

Join Our Newsletter. No Spam, Only the good stuff.

Related