RxJava 2 mock Observable doesn't work

Question

I use RxJava 2 Observable to wrap barcode scanner in one of my app. This snippet show how I subscribe to Observable and listen to it's events

override fun start() {
    disposables.addAll(subscribeToBarcodeReader(), loadUser())
}

private fun subscribeToBarcodeReader(): Disposable {
    return barcodeObservable
        .subscribeOn(bgScheduler)
        .observeOn(mainScheduler)
        .subscribeBy(
                onNext = { barcode -> saveBarcode(barcode) },
                onComplete = { Log.d(TAG, "Barcode observable complete") },
                onError = { e -> Log.d(TAG, "Barcode observable error", e) }
        )
}

barcodeObservable is object of Observable<Barcode> type. This all works fine on real device with real barcode scanner. Now I want to unit test the saveBarcode() method and try to mock the Observable<Barcode> this way

@RunWith(MockitoJUnitRunner::class)
class OrbFoldersPresenterTests {

    @Mock
    private lateinit var jobService: FolderJobService
    private val bgScheduler = TestScheduler()
    private val mainScheduler = TestScheduler()
    private val barcodeObservable: Observable<Barcode> = Observable.just(Barcode(
            "2500020109205", LocalDateTime.now().toString()))
    @Mock
    private lateinit var userStore: UserStore
    @Mock
    private lateinit var prefs: Preferences
    @Mock
    private lateinit var barcodeService: BarcodeReadService

    @Test
    fun testBarcodeSave() {
        val jobId = 201804040001
        val disposables = CompositeDisposable()

        whenever(userStore.authorizedUser()).thenReturn(Maybe.just(User()))

        val presenter = OrbFoldersPresenter(jobId, jobService, disposables, bgScheduler,
                mainScheduler, barcodeObservable, userStore, prefs, barcodeService)
        presenter.start()
    }
}

I'm expecting that when I call presenter.start() presenter must subscribe to barcodeObservable and receive onNext() event with object I've provided. But this isn't happening. What am I doing wrong?

Answer 1

The TestScheduler is useful if you need to test timing with your streams. Since your only concern with your implementation should be to run the stream synchronously, you should better use a Schedulers.trampoline() instead. This will then call subscription and observation without a thread change.