RxJava单元测试可观察的时间间隔，通过时间推移改变响应

Question

我的存储库中有一个带有Observable interval的网络轮询函数

class repository @Inject constructor(
private val api: api,
private val schedulerProvider: SchedulerProvider
) {
private fun networkPoll(): Observable<response> {
return Observable.interval(1L, TimeUnit.Seconds, schedulerProvider.io())
.startWith(0L)
.flatMap {
api.getStatus().subscribeOn(schedulerProvider.io()).toObservable()
}
}
private fun pollStates(): Observable<State> {
return networkPoll()
.map {
// map the response to State1, State2, State3
}
}
fun pollForState3(): Observable<State> {
return pollStates()
.subscribeOn(schedulerProvider.io())
.takeUntil {
it is State3
}
}
}

如何对pollForState3进行单元测试并通过时间推进来改变响应?

Answer 1

我是这样解决的

private lateinit var repository: Repository
private val schedulerProvider = TestSchedulerProvider()
@Mock
private lateinit var api: Api
@Before
fun setup() {
repository = repository(api, schedulerProvider)
}
@Test
fun test() {
`when`(api.getStatus()).thenReturn(
// return such a way that pollStates function map this to State1
)
val testObserver = repository.pollForState3().test()
schedulerProvider.ioScheduler.advanceTimeBy(1L, TimeUnit.MILLISECONDS)
testObserver.assertValue {
it is State1
}
testObserver.assertNotTerminated()
`when`(api.getStatus()).thenReturn(
// now return such a way that pollStates function map this to State3
)
schedulerProvider.ioScheduler.advanceTimeBy(2L, TimeUnit.SECONDS)
testObserver.assertValue {
it is State3
}
testObserver.assertTerminated()
testObserver.dispose()
}
}