Repetitive actions with RxJava and unit tests

rxjava-polling.md

The Use Case

class PayInvoiceUseCase constructor(
    private val invoiceRepository: InvoiceRepository,
    private val schedulerProvider: SchedulerProvider
) {
    fun getInvoiceWPolling(invoiceId: Long): Observable<Result<Invoice, ErrorReason>> {
        val stopped = AtomicBoolean()
        println("start polling")
        return Observable.interval(2L, TimeUnit.SECONDS, schedulerProvider.io())
            .takeWhile{ !stopped.get() }
            .flatMapSingle {
                println("fetch transfer")
                invoiceRepository.getInvoice(invoiceId).doOnSuccess {
                    when (it) {
                        is Result.Success -> {
                            println("stop polling")
                            stopped.set(true)
                        }
                        is Result.Failure -> Unit
                    }
                }
            }
    }

    fun payInvoice(): Single<Int> {
        println("payInvoice")
        return Single.just(111111)
    }

    fun doTheThing(invoiceId: Long) : Observable<Int> {
        return getInvoiceWPolling(invoiceId).flatMapSingle {
            return@flatMapSingle when (it) {
                is Result.Success -> payInvoice()
                is Result.Failure -> Single.just(-1)
            }
        }
    }
}

The Unit Tests

We want to test and that the polling goes on for a while until a condition is met

@RunWith(MockitoJUnitRunner::class)
class PayInvoiceUseCaseTest {
    @Test
    fun `polling goes on until condition is met, then step 2 happen`() {
        val testScheduler = TestScheduler()
        val invoice = Invoice::class.createInstance()
        val invoiceRepository = mock<InvoiceRepository> {

        }
        val schedulerProvider = object : SchedulerProvider {
            override fun ui(): Scheduler = testScheduler

            override fun computation(): Scheduler = testScheduler

            override fun trampoline(): Scheduler = testScheduler

            override fun newThread(): Scheduler = testScheduler

            override fun io(): Scheduler = testScheduler

        }

        val underTest = spy(PayInvoiceUseCase(invoiceRepository, schedulerProvider))

        val failResult = Single.just<Result<Invoice, ErrorReason>>(Result.Failure(ErrorReason.CERTIFICATE_PIN_FAILURE)) // Condition for polling to stop not met
        Mockito.doReturn(failResult).`when`(invoiceRepository).getInvoice(any())
        val testObserver = underTest.doTheThing(3243L).subscribeOn(testScheduler).observeOn(testScheduler).test()
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        val successResult = Single.just<Result<Invoice, ErrorReason>>(Result.Success(invoice))
        Mockito.doReturn(successResult).`when`(invoiceRepository).getInvoice(any()) // Condition for polling to stop met
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        // Polling should have already stopped
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)

        verify(invoiceRepository, times(2)).getInvoice(any())
        verify(underTest, times(1)).doStep2() // step only happens when the condition is met
        testObserver.assertValues(-1, 111111)
    }
}