单元测试有延迟的Rxjava可观察量

Question

我希望能够对Observable具有延迟发射的单元进行单元测试,但实际上并没有等待延迟时间.有没有办法做到这一点？

我目前正在使用CountDownHatch来延迟断言,并且工作正常,但会增加测试运行时间.

例:

val myObservable: PublishSubject<Boolean> = PublishSubject.create<Boolean>()
fun myObservable(): Observable<Boolean> = myObservable.delay(3, TimeUnit.SECONDS)

@Test
fun testMyObservable() {
    val testObservable = myObservable().test()
    myObservable.onNext(true)

    // val lock = CountDownLatch(1)
    // lock.await(3100, TimeUnit.MILLISECONDS)
    testObservable.assertValue(true)
}

Answer 1

由于@ aaron-他的答案,我能够提出一个完整的解决方案.

它使用组合TestScheduler来提前时间,并RxJavaPlugins使用testScheduler覆盖默认调度程序.这允许在myObservable()不修改或需要传入的情况下测试功能Scheduler.

@Before
fun setUp() = RxJavaPlugins.reset()

@After
fun tearDown() = RxJavaPlugins.reset()

@Test
fun testMyObservable() {
    val testScheduler = TestScheduler()
    RxJavaPlugins.setComputationSchedulerHandler { testScheduler }

    val testObservable = myObservable().test()
    myObservable.onNext(true)

    testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
    testObservable.assertEmpty()
    testScheduler.advanceTimeBy(2, TimeUnit.SECONDS)
    testObservable.assertValue(true)
}

Answer 2

TestScheduler这是完美的.它有一个方便的方法advanceTimeBy(long, TimeUnit)可以让你控制时间.并且Observable.delay有一个重载方法需要一个Scheduler.

因此,只要使用默认Scheduler.computation()的myObservable功能,并使用TestScheduler单元测试.