15. Testing RxJava Code

Written by Alex Sullivan

First and foremost — you’re a hero for not skipping this chapter! Testing your code is at the heart of writing good software — RxJava comes with lots of nifty tricks for testing everything under the sun. In this chapter, you’ll use JUnit to write unit tests to test a few operators and this chapter’s app.

Getting started

You’re going to be working on an app named HexColor for this chapter. HexColor is a nifty app that lets you input a hex color string. The app then shows you that color and (if it’s within a set of known hex colors) tells you what the name of the color is. Open the starter project and run the app. You should see an app that looks like this:

Enter a full hex string to see the app in action. It wouldn’t be a color-based app if there wasn’t some product placement, so try to enter the Ray Wenderlich green color: #006636

You should see the following screen:

In the top-left, you can see the color broken up by RGB values. On the right, you can see the name of the color.

Fancy, right?

Now that you’re thoroughly impressed, take a look at the ColorViewModel class to see what’s going on inside. Most of the logic for the app is actually contained in the init block:

// Send the hex string to the activity
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .subscribe(hexStringLiveData::postValue)
  .addTo(disposables)

// Send the actual color object to the activity
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .map { if (it.length < 7) "#FFFFFF" else it }
  .map { colorCoordinator.parseColor(it) }
  .subscribe(backgroundColorLiveData::postValue)
  .addTo(disposables)

// Send over the color name "--" if the hex string is less than
// seven chars
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .filter { it.length < 7 }
  .map { "--" }
  .subscribe(colorNameLiveData::postValue)
  .addTo(disposables)

// If our color name enum contains the given hex string, send
// that color name over.
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .filter {
    hexString -> ColorName.values().map { it.hex }
                   .contains(hexString)
  }
  .map { hexString -> ColorName.values().first {
     it.hex == hexString }
  }
  .map { it.toString() }
  .subscribe(colorNameLiveData::postValue)
  .addTo(disposables)

// Send the RGB values of the color to the activity.
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .map {
    if (it.length == 7) {
      colorCoordinator.parseRgbColor(it)
      } else {
        RGBColor(255, 255, 255)
      }
    }
    .map { "${it.red},${it.green},${it.blue}" }
    .subscribe(rgbStringLiveData::postValue)
    .addTo(disposables)

The hexStringSubject property is a BehaviorSubject, which receives hex string digits from the user as they come in via the digitClicked method. At any given moment, hexStringSubject has the whole hex string that the user has entered.

Each block in the above code subscribes to the hexStringSubject behavior subject, interprets the current string, and sends some information to the several live data objects contained in ColorViewModel.

The app is complete in its functionality — it just needs a few tests to make it perfect!

Weirdly enough, whoever wrote this app actually created two test classes with some plumbing already set up. How convenient!

Before you start writing tests for ColorViewModel, you need some background on testing in RxJava. To do that, you’ll start by writing a few sample RxJava tests in the OperatorTest class.

Introduction to TestObserver

Have you ever tried to test asynchronous code? If you have, you probably know that it’s no cake walk. It can be (very) difficult to both test all aspects of your asynchronous code and keep your unit tests running quickly. RxJava provides an extremely convenient set of test utilities to make testing Observables easier — the first of which is the TestObserver class.

Open up OperatorTest.kt, and add the following code to the test concat method, which is already in the file:

@Test
fun `test concat`() {
  val observableA = Observable.just(1)
  val observableB = Observable.just(2)
  val observableC = observableA.concatWith(observableB)
}

Note: When offered an import for Observable, make sure to import io.reactivex.rxjava3.core.Observable, not the java.util version.

You’ve got two simple Observables that emit one integer and then complete. Then you’ve got a third Observable that uses the concatWith method to concatenate those two Observables together. As you’ve probably gathered from the name of the method, you want to test that that concatWith method returns what you’d expect — an Observable that emits two values: 1 followed by 2, and then it finishes.

You could subscribe to observableC and record what values are emitted and then assert against those values as they come in. But that would be messy and would quickly fall apart when you have more complicated streams.

You’ve got another option: the test() method.

Every RxJava type (Observable, Maybe, Completable, and so on) exposes a test() method that returns a TestObserver. You can use this TestObserver class to assert against different conditions on your Observable (or whatever other RxJava type you’re using).

Add the following code below the line declaring observableC:

observableC.test()
  .assertResult(1, 2)
  .assertComplete()

You’re using the test() method on observableC and asserting a couple of things: that the Observable returns two results, 1 and 2, and then completes.

Run this unit test by right-clicking the little green Play button in the left sidebar next to the test name:

You should see the test pass.

If you’re like me, you’re always skeptical of a test that passes on the first try. Try updating the assertResult statement to remove one of the values:

.assertResult(1)

Run the test again, and good news — it does indeed fail! 👍 for failure!

The TestObserver class that the test() method returns has many more uses. One of the most convenient things that it offers is an insight into what values the Observer has received so far. You can use the values() method on it to get a list of all the items that Observable has emitted. You’ll see a few more examples of what TestObserver can do as you progress through some other nifty testing utilities RxJava exposes.

Using a TestScheduler

In addition to TestObserver, the RxJava library exposes a special scheduler that you can use to control when your Observables emit items. That scheduler is called TestScheduler.

Add the following code below the concat test you wrote earlier:

@Test
fun `test amb`() {
  // 1
  val observableA = Observable.interval(1, TimeUnit.SECONDS)
    .take(3)
    .map { 5 * it }
  val observableB = Observable
    .interval(500, TimeUnit.MILLISECONDS)
    .take(3)
    .map { 10 * it }

  // 2
  val ambObservable = observableA.ambWith(observableB)

  // 3
  val testObserver = ambObservable.test()

  testObserver.assertValueCount(3)
  testObserver.assertResult(0L, 10L, 20L)
  testObserver.assertComplete()
}

Since that’s a lengthy block of code, here’s a breakdown to describe what’s going on so far:

1. You’re creating two Observables using the interval method. As a reminder, the interval factory method creates a new Observable that starts counting up at a frequency that you dictate. For observableA, that frequency is once every second. For observableB, that frequency is once every 500 milliseconds. You’re then taking the first three results from each Observable and doing math on it. For observableA, you’re multiplying that number by 5. For observableB, you’re multiplying the number by 10.

2. You’re then creating a new ambObservable using the ambWith method. The amb and ambWith methods are very handy — they take two or more Observables and combine them into a resulting Observable that mirrors the first Observable to fire. The other Observable is discarded. amb comes in very handy when you have two sources of information and you only care about whichever one is the fastest. Imagine reading from a database and pulling data from a network. You may only care about which source gets your users the information they care about fastest. Neat!

3. Finally, you’re using the test() method you learned about earlier to create a testObserver. You’re then asserting a few things about the ambObservable: it emitted three values, those values were longs with a value of 0, 10, and 20, and finally that the Observable completed after emitting those three values.

Note: If you’re confused about why 0, 10, and 20 are emitted by the ambObservable don’t worry! amb can be a confusing method to wrap your head around. Since observableA emits every second and observableB emits every half second, observableB will be the first one to emit between the two. Since amb only mirrors the first Observable to emit a value, observableB will win out. Then, since observableB uses the map method to multiply its values by 10, the resulting observable will emit 0 * 10, 1 * 10, and 2 * 10 - so 0, 10, and 20.

Run the unit test by clicking the Run test button, which shows up in the left sidebar next to the top of the test amb method. You should see the following:

java.lang.AssertionError: Value counts differ; Expected: 3, Actual: 0 (latch = 1, values = 0, errors = 0, completions = 0)

Uh, oh — that doesn’t look like a success at all! It looks like the assertValueCount(3) call failed. assertValueCount asserts that the Observable has emitted a certain number of values. The error message shows that ambObservable has emitted zero items by the time you start asserting.

The reason that call failed is because the testing code hasn’t waited enough time for the Observables to actually start emitting values. Remember that the first Observable only emits once a second (starting after the first second), and the second Observable only emits once every 500 milliseconds.

There are two options for handling these sorts of timing issues:

1. You could start adding Thread.sleep calls to make the test wait long enough for the Observable to start emitting. Good test design and sleeping threads are not two things that go together! In this case, you run the risk of hitting interrupted thread exceptions and it makes your tests take much longer to finish. Doesn’t sound like a great option.
2. You could use a TestScheduler and use it to control time yourself!

It may be shocking, especially with the title of this section, but the best path forward is option number 2: using a TestScheduler.

At the very top of test amb, create a TestScheduler before creating any of the Observables:

val scheduler = TestScheduler()

Then, update both Observable.interval calls to take a third parameter — the new TestScheduler:

val observableA = Observable.interval(1, TimeUnit.SECONDS, scheduler)
  .take(3)
  .map { 5 * it }
val observableB = Observable
  .interval(500, TimeUnit.MILLISECONDS, scheduler)
  .take(3)
  .map { 10 * it }

This may seem weird. Usually, you use schedulers in either the subscribeOn or observeOn operators. However, most RxJava operators and factory methods that deal with time can actually take a scheduler as a parameter.

That scheduler is then responsible for reporting the time back to that observable so it can figure out when to emit a new item. By default, the computation scheduler is used for this time logic. In the above example, the interval method will ask the TestScheduler you passed in for time information.

This is great news because TestScheduler allows you to control what time it reports back to the interval method!

Remove the existing three assertions at the end of test amb, and replace that code with the following:

scheduler.advanceTimeBy(500, TimeUnit.MILLISECONDS)
testObserver.assertValueCount(1)

You’re using the advanceTimeBy method on TestScheduler to advance the schedulers clock forward a certain amount of time; in this case, 500 milliseconds. Since observableB emits every 500 milliseconds, that means that after that call the ambObservable should have emitted one value.

Run the unit test again. You should see a successful test. Nice!

For completeness, add the following code below the assertValueCount call:

scheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS)

testObserver.assertValueCount(3)
testObserver.assertResult(0L, 10L, 20L)
testObserver.assertComplete()

You’re again advancing time (can anyone say time travel?). This time you’re moving forward one more second, which should give observableB the opportunity to emit two more items. Since you’re using the take method the Observable should finish after the three values are emitted.

Run the test, again, and you should see another success. Nice!

In addition to the advanceTimeBy method, TestScheduler exposes a method called triggerActions, which triggers any actions that are due to be run by that point in time. You’ll see an example later on.

Injecting schedulers

There will be many times in which you’re attempting to unit test classes that don’t directly expose an Observable. For example: Most of the ViewModel classes that you’ll see in this book don’t expose an Observable. Instead, they subscribe to those Observables internally and expose LiveData objects that work better with the Android lifecycle.

That makes for a great architecture, but it can make it more difficult to test that your Observables are doing what you expect.

Here’s an example Timer class that uses the interval method to count time:

class Timer() {
  var elapsedTime: Int = 0

  init {
    val intervalObservable = Observable
      .interval(1, TimeUnit.SECONDS)
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())

    intervalObservable  
      .subscribe {
        elapsedTime++
      }
  }
}

You can query the timer’s elapsedTime variable to see how much time has passed since it was first instantiated.

Now imagine you wanted to unit test this class. Since intervalObservable isn’t exposed, you can’t use the test() method or supply a TestScheduler to the subscribeOn or observeOn operators.

This is where Dependency Injection comes in to play. Injection?! That sounds painful!

The good news is that Dependency Injection is a fancy term for passing parameters, which supply dependencies to your classes rather than having those classes create them internally.

With Dependency Injection, you can rewrite the Timer class as follows:

class Timer(backgroundScheduler: Scheduler,
  mainThreadScheduler: Scheduler, timerScheduler: Scheduler) {
  var elapsedTime: Int = 0

  init {
    Observable.interval(1, TimeUnit.SECONDS, timerScheduler)
      .subscribeOn(backgroundScheduler)
      .observeOn(mainThreadScheduler)
      .subscribe {
        elapsedTime++
      }
  }
}

Now you can easily pass in a TestScheduler when you create an instance of Timer to unit test. Everyone wins!

Using Trampoline schedulers

Now that you’re injecting schedulers, there’s another scheduler that can be very helpful when running unit tests.

Often times, when unit testing an Observable, you want a scheduler that will force the work of the Observable to happen on the current thread. You can achieve some of this behavior by using TestScheduler. However, if you’re not working with Observables that interact with time it can be laborious to have to call advanceTimeBy or triggerActions all the time.

Instead, you can use the TrampolineScheduler class, which you saw in Chapter 13, “Intro to Schedulers.”

In case you missed that chapter or it’s been a while, here’s a quick refresher: TrampolineScheduler is a scheduler that schedules work on the current thread at the end of an internal queue it holds. It’s a great option when you’re injecting a scheduler and you don’t want to go through the ceremony of using a TestScheduler.

How about a quick example? Add the following unit test to the OperatorTest class:

@Test
fun `using trampoline schedulers`() {
  val observableA = Observable.just(1)
    .subscribeOn(TrampolineScheduler.instance())

  val observableB = Observable.just(1)
    .subscribeOn(Schedulers.io())
}

Here you see two Observables, both of which are using the just method to construct an Observable that emits the integer 1 then finishes.

observableA uses the subscribeOn operator with a TrampolineScheduler, whereas observableB uses the io scheduler.

If you were to run these two Observables, what do you think would happen?

Since observableA is using a TrampolineScheduler, it will be run on whatever the current thread is, thus blocking the method until it finishes. observableB, on the other hand, would run on a different thread and the test method would terminate before it finished!

Add the following code at the end of the method:

observableA.test().assertResult(1)
observableB.test().assertEmpty()

You’re asserting that observableA does indeed finish while observableB does not, since it’s run on a different thread and won’t have time to finish before the assertion is called.

Run the unit test. You should see a dazzling success. Well, a success anyways!

Using subjects with mocked data

One thing that can be very helpful is mocking data — that is, replacing one real piece of the puzzle with a different one that appears the same but which you have direct control over. This allows you to check that the other pieces of the puzzle work the way you expect them to when you feed them specified data.

As an example, think of trying to create a button that allows the user to repeatedly tap to add a photo to a list, with a maximum of five photos. If you were to use a ViewModel to control that, you would want to keep adding incoming photos until the list reached its maximum, and then disable the button the ViewModel controlled.

Within the test package, create a new Kotlin file called PhotoTest.kt. In this file, add the following code to bring this example to life:

// 1
class Photo

// 2
interface PhotoProvider {
  fun photoObservable(): Observable<Photo>
}

// 3
class PhotoViewModel(provider: PhotoProvider) {

  var disableButton = false
  private var photoList = arrayListOf<Photo>()

  init {
    // 4
    provider.photoObservable()
      .subscribe {
        photoList.add(it)
        if (photoList.size >= 5) {
          disableButton = true
        }
      }
  }
}

What’s happening here? Walking through this step by step:

1. Create the simplest possible Kotlin class — one that just has a name.
2. Declare an interface, which will provide an Observable that can be watched.
3. Create a ViewModel, which takes the PhotoProvider interface you just created as a parameter. Congratulations - you’re now using dependency injection!
4. Take the passed in PhotoProvider and subscribe to its Observable. When a new photo is added, you add it to the list and then determine if the button should be disabled.

Next, below the existing code, add a new test class and the beginnings of a test:

class PhotosTest {
  @Test
  fun `button disabled after 5 photos`() {
    val photoProviderMock = object: PhotoProvider {
      override fun photoObservable(): Observable<Photo> {
        TODO("Return some data")
      }
    }

    val viewModel = PhotoViewModel(photoProviderMock)

    Assert.assertFalse(viewModel.disableButton)
  }
}

What value should the photoObservable method return? You could provide a simple Observable that immediately returns five photos. But then you can’t accurately test the transition from the disableButton value going from false to true.

Instead, it’s often beneficial to return a PublishSubject that you can control in the test by handing it objects one by one. Update your test to create one, then return it as the photoObservable:

val subject = PublishSubject.create<Photo>()
val photoProviderMock = object: PhotoProvider {
  override fun photoObservable(): Observable<Photo> {
    return subject
  }
}

Next, at the end of the test, add code to pass some photos through the Observable and check whether disableButton is still false or has been flipped to true:

subject.onNext(Photo())
Assert.assertFalse(viewModel.disableButton)
subject.onNext(Photo())
subject.onNext(Photo())
subject.onNext(Photo())
Assert.assertFalse(viewModel.disableButton)
subject.onNext(Photo())
Assert.assertTrue(viewModel.disableButton)

Run the test by clicking the Run button in the left sidebar, and it will pass — disableButton is still false after adding four photos, but true after adding a fifth!

Now, you can precisely control when the subject emits new values and be more confident in your tests. Wahoo!

Testing ColorViewModel

Now that you’re an expert in testing, it’s time to add some real unit tests to the ViewModelTest class.

Open ViewModelTest.kt and add the following empty unit test:

@Test
fun `color is red when hex string is FF0000`() {
}

This unit test is testing that when the user enters the hex color “FF0000”, the view models colorNameLiveData live data object emits the “Red” color name.

If you go back into the ColorViewModel class you can pick out the relevant piece of code in the init block. It looks like this:

// If our color name enum contains the given hex string, send that color name over.
hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .filter { hexString ->
    ColorName.values().map { it.hex }.contains(hexString)
  }
  .map { hexString ->
    ColorName.values().first { it.hex == hexString }
  }
  .map { it.toString() }
  .subscribe(colorNameLiveData::postValue)
  .addTo(disposables)

You’ll also notice that the ColorViewModel class is set up to take in schedulers for both background work and main thread work. Nice!

Since this block utilizes the subscribeOn and observeOn operators, and doesn’t utilize any Observables that deal directly with time, it’s a good candidate for using a TrampolineScheduler.

Go back to ViewModelTest.kt. In the empty unit test method you just added, add the following code:

val trampolineScheduler = TrampolineScheduler.instance()
val viewModel = ColorViewModel(trampolineScheduler,
  trampolineScheduler, colorCoordinator)

You’re getting an instance of TrampolineScheduler and constructing a ColorViewModel, passing that trampoline scheduler in for both the background and main thread schedulers. You’re also passing in a ColorCoordinator mock that’s defined at the top of the file. ColorCoordinator is a simple class that parses out RGB values of a color and wraps a call to the Color.parseColor Android function to make testing the ViewModel easier.

Since you’re passing in a TrampolineScheduler, you know all of the RxJava work will be done synchronously. All that’s left is adding the business logic of the test! Add the following code below the ViewModel declaration:

viewModel.digitClicked("F")
viewModel.digitClicked("F")
viewModel.digitClicked("0")
viewModel.digitClicked("0")
viewModel.digitClicked("0")
viewModel.digitClicked("0")

Assert.assertEquals(ColorName.RED.toString(),
  viewModel.colorNameLiveData.value)

Boom! You’re simulating the user clicking the relevant digits and then asserting that the colorNameLiveData current value is equal to the RED color name.

Run the test and you should see it pass.

That was easy enough with a TrampolineScheduler, but what would it look like using the TestScheduler? Add the following new unit test:

@Test
fun `color is red when hex string is FF0000 using test scheduler`() {
  val testScheduler = TestScheduler()
  val viewModel = ColorViewModel(testScheduler,
    testScheduler, colorCoordinator)

  viewModel.digitClicked("F")
  viewModel.digitClicked("F")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")

  Assert.assertEquals(null, viewModel.colorNameLiveData.value)
  Assert.assertEquals(ColorName.RED.toString(),
    viewModel.colorNameLiveData.value)
}

In this version of the unit test, you’re doing the exact same thing as before except passing in a TestScheduler instead of a TrampolineScheduler.

Run the test. You should see the following:

java.lang.AssertionError:
Expected :RED
Actual   :null

As you saw earlier, TestScheduler requires more ceremony to use than TrampolineScheduler. You need to tell it to advance time or trigger its actions before any work done on that scheduler will actually happen.

Add the following line between the two assert calls:

testScheduler.triggerActions()

Run the test again and it should succeed. On to the next test!

Last but not least, it’d be good to test that the Clear button correctly clears the hex string display and replaces it with a single # character. Add the following new unit test:

@Test
fun `hex subject is reset after clear is clicked`() {
}

What kind of scheduler do you want to use for this test? Take a look at the ColorViewModel class again. Every time digitClicked is called, the view model calls onNext with the relevant character on the hexStringSubject.

In the top of the init block, you can see the code that determines what the app shows in the hex string field:

hexStringSubject
  .subscribeOn(backgroundScheduler)
  .observeOn(mainScheduler)
  .subscribe(hexStringLiveData::postValue)
  .addTo(disposables)

Pretty simple — the current string in hexStringSubject is just fed into the hexStringLiveData variable. That means that using a TrampolineScheduler in the test should be good enough; there’s no need to use TestScheduler.

Back in the test, set up the code. Again, you’ll want to use a TrampolineScheduler to create your ViewModel and feed it the digits of a hex color:

@Test
fun `hex subject is reset after clear is clicked`() {
  val trampolineScheduler = TrampolineScheduler.instance()
  val viewModel = ColorViewModel(trampolineScheduler,
    trampolineScheduler, colorCoordinator)

  viewModel.digitClicked("F")
  viewModel.digitClicked("F")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")
  viewModel.digitClicked("0")
}

Next, add the following lines at the bottom of the test to validate both that the subject has fully updated and that clicking Clear actually does what you want it to:

Assert.assertEquals("#FF0000", viewModel.hexStringSubject.value)
viewModel.clearClicked()
Assert.assertEquals("#", viewModel.hexStringSubject.value)

Run the test. Everything works exactly as expected, and you’ve got a more resilient and tested app! Well done!

Key points

• You can use the test() method on any reactive type to easily test them.

• TestObserver provides a useful set of tools to test the values and state of your Observables.

• With TestObserver, you can assert that your Observable has completed, has emitted a few values, or has even thrown an error.

• In order to test classes that don’t expose their internal Observables, you should use the Dependency Injection design pattern to inject your schedulers.

• If you need a synchronous scheduler that allows you to trigger new actions, you can use TestScheduler.

• If all you need is to make your Observables synchronous, use TrampolineScheduler.

• Subjects can be used to precisely control when elements are emitted and, combined with mocked data, make for great testing tools.

Where to go from here?

Testing is an important piece to writing great apps. Hopefully after reading this chapter, you’ve picked up some tricks to use the next time you need to test some reactive code. Happy testing!