RxJava 单元测试答案

【问题标题】：RxJava UnittestsRxJava 单元测试
【发布时间】：2017-08-04 09:14:01
【问题描述】：

我在测试我的 RxJava 代码时遇到了麻烦，因为它似乎总是在测试中运行，而无需等待调用 subscribe() 时应该发生的执行。

这是要测试的存储库：

public class Repository<T extends Entity> {

    private final DataSource<T> remoteDataSource;
    private final DataSource<T> localDataSource;

    public Repository(DataSource<T> remoteDataSource, DataSource<T> localDataSource) {
        this.remoteDataSource = remoteDataSource;
        this.localDataSource = localDataSource;
    }

    public Observable<Boolean> save(T entity) {

        return localDataSource
                .save(entity)
                .flatMap(success -> {
                    if (success) {
                        return remoteDataSource.save(entity);
                    }
                    return Observable.just(Boolean.FALSE);
                })
                .doOnNext(success -> {
                    if (success) {
                        if (cache == null) {
                            cache = new LinkedHashMap<>();
                        }
                        cache.put(entity.getId(), entity);
                    }
                });
    }
}

这个DataSource接口：

public interface DataSource<T extends Entity> {
    <T> Observable<T> get();

    Observable<Boolean> save(T entity);

    Observable<Boolean> clear();

    Observable<Boolean> remove(T entity);
}

这是包含一个 MockDataSource 的单元测试，它应该模拟具有不同执行时间的数据访问：

public class RepositoryTest {

    @Test
    public void testRepository() {

        MockSource remoteSource = new MockSource("RemoteSource", 1000L);
        MockSource localSource = new MockSource("LocalSource", 200L);
        Repository<Poi> poiRepository = new Repository<>(remoteSource, localSource);

        Poi poi1 = newMockPoi();

        Observable<Boolean> obs = poiRepository.save(poi1);
        TestSubscriber<Boolean> testSubscriber = new TestSubscriber<>();
        obs.subscribe(testSubscriber);

        testSubscriber.assertNoErrors();
        testSubscriber.assertReceivedOnNext(Arrays.asList(true));


    }

    private Poi newMockPoi() {
        Poi poi = new Poi();
        poi.name = RandomStringUtils.randomAlphabetic(12);
        poi.description = RandomStringUtils.randomAlphabetic(255);
        poi.latitude = new Random().nextDouble();
        poi.longitude = new Random().nextDouble();
        return poi;
    }


    private class Poi extends Entity {
        String name;
        String description;
        Double latitude;
        Double longitude;
    }

    private class MockSource implements DataSource<Poi> {

        private String name;
        private final long delayInMilliseconds;

        private Map<Long, Poi> pois = new LinkedHashMap<>();

        private MockSource(String name, long delayInMilliseconds) {
            this.delayInMilliseconds = delayInMilliseconds;
            this.name = name;
        }

        @Override
        public Observable<List<Poi>> get() {
            return Observable
                    .zip(
                            Observable
                                    .just(pois)
                                    .map(Map::entrySet)
                                    .flatMapIterable(entries -> entries)
                                    .map(Map.Entry::getValue)
                                    .toList(),
                            Observable
                                    .interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs)
                    .doOnNext(pois -> System.out.println("Soure " + name + " emitted entity"));
        }

        @Override
        public Observable<Boolean> save(Poi entity) {
            return Observable
                    .zip(
                            Observable.just(true).asObservable(),
                            Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs)
                    .doOnNext(value -> pois.put(entity.getId(), entity))
                    .doOnNext(pois -> System.out.println("Soure " + name + " saved entity"));
        }

        @Override
        public Observable<Boolean> clear() {
            return Observable
                    .zip(
                            Observable.just(true).asObservable(),
                            Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs)
                    .doOnNext(value -> pois.clear())
                    .doOnNext(pois -> System.out.println("Soure " + name + " cleared all entities"));
        }

        @Override
        public Observable<Boolean> remove(Poi entity) {
            return Observable
                    .zip(
                            Observable.just(true).asObservable(),
                            Observable.interval(delayInMilliseconds, TimeUnit.MILLISECONDS), (obs, timer) -> obs)
                    .doOnNext(value -> pois.remove(entity))
                    .doOnNext(pois -> System.out.println("Soure " + name + " removed entity"));
        }
    }
}

这是输出：

java.lang.AssertionError: Number of items does not match. Provided: 1  
Actual: 0.
Provided values: [true]
Actual values: []
 (0 completions)

at rx.observers.TestSubscriber.assertionError(TestSubscriber.java:667)
at rx.observers.TestSubscriber.assertReceivedOnNext(TestSubscriber.java:320)
at nl.itc.geofara.app.data.source.RepositoryTest.testRepository(RepositoryTest.java:37)

此外，在存储库的保存方法中设置断点并在调试模式下运行会显示其中的代码，例如.flatMap(success -> ...) 永远不会被调用。

【问题讨论】：

参见stackoverflow.com/a/39828581/4191629，“疑难解答”，“比赛条件”
我以前见过 Schedulers.immediate() 的概念，但是给定的源从未在任何地方调用 subscribeOn()。我认为所有任务现在都在与单元测试本身相同的线程上执行。我的假设错了吗？
好吧 sh*t... 你是对的！
@maciekjanusz 添加解决方案 #1 作为答案，我将其标记为正确。解释为什么这会很棒！
已添加答案。乐于助人！

标签： unit-testing junit rx-java

【解决方案1】：

您在可观察执行和 JUnit 线程之间遇到竞态条件，因为您在模拟源中使用的Observable.interval 引入了implicit subscription on computation scheduler：

调度器：默认情况下，interval 在计算调度程序上运行。

另外，请参阅this answer，“疑难解答”，“比赛条件”中的更多说明。

【讨论】：