A kind of Spek logo (thanks to @vascogmm)

This article aims to provide a simple example of how your RxJava code can be spec’d using the Spek framework while taking advantage of PublishSubjects to produce pleasant unit tests for both writer and reader.

Consider the following scenario: A DistancesProvider gives distances from one location to another in both kilometres and miles. A UnitProvider defines a distance unit which can be either kilometres or miles.

fun distances() : Observable<List<Int>> // [kilometres, miles] fun unit() : Observable<Int> // 0 -> kilometres, 1 -> miles

Using these providers we can build a simple GPS system that emits distances in real time. There’s only one condition: the system should only show distances in either kilometres or miles, never both at the same time.

We can easily implement this requirement using the combineLastest operator, which receives events from both providers and applies a function to select which distance to show.

fun distances(): Observable<Int> {

return Observable.combineLatest(

distancesProvider.distances(),

unitProvider.unit(),

{ distances, unit -> distances[unit] })

}

For this simple method, a possible specification using the Spek DSL would be: given a GPS system on a distance update of 30 km it should emit the value 30. Which in code should be something like:

given("a GPS system") {



on("a distance update of 30 km") {



it("should emit the value 30") {



}

}

}

In order to actually have a test running, Spek allow us to define pre and post conditions using the beforeEachTest and afterEachTest methods, which are quite similar to the Before and After JUnit annotations. So, let’s instantiate a mock for the DistancesProvider and, of course, the class being tested which we’ll call DistancesUseCase . As a side note, the Mockito-Kotlin library comes in very handy and saves a bunch of time when it comes to writing mocks.

val distancesProvider: DistancesProvider = mock()

val unitProvider: UnitProvider = mock()

val tested = DistancesUseCase(distancesProvider, unitProvider) whenever(distancesProvider.distances()).thenReturn(...)

whenever(unitProvider.unit()).thenReturn(...)

Usually, we’d return certain values according to the test. For instance,

Observable.just(listOf(30, 19)) for the distances() method;

for the method; Observable.just(0) for the unit() method;

This is fine, but for each test, we’d need to set new return types. A faster approach that also turns specifications much simpler and readable comes with PublishSubjects. Since they can act as both Observable and Observer, they work well as values returned by the DistancesProvider and the UnitProvider , while taking advantage of the observer interface methods to decide when each event is emitted.

var distanceSub: PublishSubject<List<Int>> = PublishSubject.create()

var unitSub: PublishSubject<Int> = PublishSubject.create() whenever(distancesProvider.distances()).thenReturn(distanceSub)

whenever(unitProvider.unit()).thenReturn(unitSub)

To finish the setup let’s grab a TestSubscriber<Int> that we’ll use to subscribe to the Observable<Int> returned by the distance method in order to reason about all possible outcomes.

val distancesProvider: DistancesProvider = mock()

val unitProvider: UnitProvider = mock()

val tested = DistancesUseCase(distanceProvider, unitProvider)



var distanceSub: PublishSubject<List<Int>> = PublishSubject.create()

var unitSub: PublishSubject<Int> = PublishSubject.create()



var testSubscriber = TestSubscriber<Int>() given("a GPS system") { beforeEachTest {

distanceSub = PublishSubject.create()

unitSub = PublishSubject.create()

testSubscriber = TestSubscriber() whenever(distancesProvider.distances()).thenReturn(distanceSub)

whenever(unitProvider.unit()).thenReturn(unitSubject)



tested.distances().subscribe(testSubscriber)

}



afterEachTest {

reset(distancesProvider)

reset(unitProvider)

}

Now we have all ingredients needed to test our GPS system. Following the specification above, we want to verify that when a distance update of 30 km comes in, the system outputs the value 30. Using the distanceSub and unitSub this task becomes really easy:

given("a GPS system") { beforeEachTest { ... }

afterEachTest { ... } on("a distance update of 30 km") {

distanceSub.onNext(listOf(30, 19)) // [30km, 19miles]

unitSub.onNext(0) // km



it("should emit the value 30") {

testSubscriber.assertValue(30)

}

}

As the Spek documentation states, the on action can be repeated any number of times for each group (given, context, describe). The it test can also be called any number of times for each on action or for each group.

Therefore, we can comfortably add a few tests to the action “a distance update of 30 km” already defined:

on("a distance update of 30 km") {

distanceSub.onNext(listOf(30, 19)) // [30km, 19miles]

unitSub.onNext(0) // km



it("should emit the value 30") {

testSubscriber.assertValue(30)

} it("should not complete") {

testSubscriber.assertNotCompleted()

} it("should not throw errors") {

testSubscriber.assertNoErrors()

}

}

This last test shows a useful assertion to verify that no errors are returned. The previous test help us understanding that we want a reactive stream that should not terminate after emitting an event. So, imagine that unitSub doesn’t produce any more items and completes. Does the main observable complete? Or do we still get distances from the distanceSub ?

on("a distance update of 30 km, no more units and more distances") {

distanceSub.onNext(listOf(30, 19)) // [30km, 19miles]

unitSub.onNext(0) // km

unitSub.onCompleted()

distanceSub.onNext(listOf(20, 12)) // [20km, 12miles] it("should emit the value 30 followed by value 20") {

testSubscriber.assertValues(30, 20)

} it("should not complete") {

testSubscriber.assertNotCompleted()

}

}

Both tests pass, which means we still get distances after unitSub completes. What if the unit is changed immediately before the stream completes?

on("a distance update of 30 km, miles, no units, more distances") {

distanceSub.onNext(listOf(30, 19)) // [30km, 19miles]

unitSub.onNext(0) // km

unitSub.onNext(1) // miles

unitSub.onCompleted()

distanceSub.onNext(listOf(20, 12)) // [20km, 12miles] it("should emit the value 30 followed by value 12") {

testSubscriber.assertValues(30, 12)

}

}

FAILED! Hum… Why?

java.lang.AssertionError:

Number of items does not match. Provided: 2 Actual: 3.

Provided values: [30, 12]

Actual values: [30, 19, 12]

Ah, of course! combineLatest blends the last value of each stream with the most recent one. In this case, unitSub.onNext(1) is combined with the previous value of distanceSub.onNext(listOf(30, 19)) . And only after that we see the last distance update in miles being emitted, settling the result as (30, 19, 12) .

The previous sequence of events inside the on action (“a distance update of 30 km, miles, no units, more distances”) becomes quite confusing and poorly readable. But we can manage to write the same specification using Spek’s nested groups, as follows:

context("distance update of 30 km") {

beforeEachTest {

distanceSub.onNext(listOf(30, 19))

unitSub.onNext(0)

}



it("should emit the value 30") {

testSubscriber.assertValue(30)

}



context("unit update to miles") {

beforeEachTest {

unitSub.onNext(1)

}



it("should emit 19 as second value") {

testSubscriber.assertValues(30, 19)

}



context("no more units") {

beforeEachTest {

unitSub.onCompleted()

}



it("should not complete") {

testSubscriber.assertNotCompleted()

}

}

}

}

Groups define the context in which each test should run and, when nested, they provide a good flow that follows the stream’s behaviour.