您如何使用 io 调度程序使用 mockito、rxjava2 在 spring mvc 中测试服务？

Question

我正在尝试测试一种使用 rxjava2、flatmap 和 io 调度程序的服务方法。尽管出现运行，该测试似乎并未使用 mockito 调用任何模拟方法。 observable 永远不会 returns 一个对象。

如何使用多线程 rxjava2 代码测试 spring 服务？

这是我正在测试的服务中的方法

@Transactional(isolation = Isolation.READ_UNCOMMITTED)
public io.reactivex.Observable<EntryStatistic> compute(final User user, final int startYear, final int endYear) {
    final GregorianCalendar calendarg = new GregorianCalendar();

    if (endYear < startYear)
        throw new IllegalArgumentException("endYear");
    if (endYear < 2003)
             throw new IllegalArgumentException("endYear");
    if (startYear < 2003)
        throw new IllegalArgumentException("startYear");

    return io.reactivex.Observable.range(startYear, endYear - startYear + 1)
            .flatMap(new Function<Integer, ObservableSource<EntryStatistic>>() {
                @Override
                public ObservableSource<EntryStatistic> apply(final Integer yr) throws Exception {
                    return io.reactivex.Observable.fromCallable(new Callable<EntryStatistic>() {
                        @Override
                        public EntryStatistic call() throws Exception {

                            log.debug("testing with year: " + yr + " user: " + user.getUsername() );
                            EntryStatistic es = entryStatisticRepository.findByUserAndYear(user, yr);
                            final long count = entryRepository.calendarCount(yr, user.getUsername());

                            if (es == null) {
                                log.trace("Creating new entry statistic");
                                es = new EntryStatistic();
                                es.setUser(user);
                            }

                            es.setCount(count);
                            es.setYear(yr);
                            es.setModified(calendarg.getTime());

                            log.trace("save and flush time");
                            return entryStatisticRepository.saveAndFlush(es);

                        }
                    });
                }
            }).subscribeOn(Schedulers.io());
}

这里是测试代码：

@Test
public void computeSingle() {
    when(entryStatisticRepository.findByUserAndYear(user, TEST_YEAR + 1)).thenReturn(entryStatistic);
    when(user.getUsername()).thenReturn(TEST_USER);
    when(entryRepository.calendarCount(TEST_YEAR, TEST_USER)).thenReturn(1L);
    when(entryStatistic.getUser()).thenReturn(user);
    when(entryStatistic.getCount()).thenReturn(1L);
    when(entryStatistic.getYear()).thenReturn(TEST_YEAR);
    when(entryStatisticRepository.saveAndFlush(entryStatistic)).thenReturn(entryStatistic);

    TestObserver<EntryStatistic> testObserver = entryStatisticService.compute(user, TEST_YEAR, TEST_YEAR )
            .test();

    testObserver.awaitTerminalEvent();
    testObserver
        .assertNoErrors()
        .assertValue(new Predicate<EntryStatistic>() {
            @Override
            public boolean test(final EntryStatistic entryStatistic) throws Exception {
                return entryStatistic.getCount() == 1L ;
            }
        });

    verify(entryStatisticRepository, atLeastOnce()).findByUserAndYear(user, TEST_YEAR);
    verify(entryRepository, atLeastOnce()).calendarCount(TEST_YEAR, TEST_USER);
}

最后，这是我为代码强制使用单线程调度程序作为 junit 规则所做的尝试。

public class TrampolineSchedulerRule implements TestRule {

@Override
public Statement apply(final Statement base, Description d) {
  return new Statement() {
    @Override
    public void evaluate() throws Throwable {
        RxJavaPlugins.setIoSchedulerHandler(new Function<Scheduler, Scheduler>() {
            @Override
            public Scheduler apply(final Scheduler scheduler) throws Exception {
                return Schedulers.trampoline();
            }
        });
        RxJavaPlugins.setComputationSchedulerHandler(new Function<Scheduler, Scheduler>() {
                         @Override
                         public Scheduler apply(final Scheduler scheduler) throws Exception {
                             return Schedulers.trampoline();
                         }
                     });
        RxJavaPlugins.setNewThreadSchedulerHandler(new Function<Scheduler, Scheduler>() {
                                      @Override
                                      public Scheduler apply(final Scheduler scheduler) throws Exception {
                                          return Schedulers.trampoline();
                                      }
                                  });


      try {
        base.evaluate();
      } finally {
        RxJavaPlugins.reset();
      }
    }
  };
}
}

问题是没有返回任何对象。我看到错误

java.lang.AssertionError: No values (latch = 0, values = 0, errors = 0, completions = 1)

Answer 1

模拟的设置方式存在问题。这个模拟电话

 when(entryStatisticRepository.saveAndFlush(entryStatistic)).thenReturn(entryStatistic);

期望 entryStatistic 对象被传递给 saveAndFlush，但实际上新的 EntryStatistic 对象被传递给 saveAndFlush 并且没有返回所需的模拟值。这意味着从 Callable 返回空值。由于无法在 rx-java2 中发出 null 值，这导致 flatMap 没有发出任何值，您会立即收到没有值的完成。

可以通过删除加一来修复测试（因为范围只会发出一个等于 TEST_YEAR 的值）

when(entryStatisticRepository.findByUserAndYear(user, TEST_YEAR+1)).thenReturn(entryStatistic);

一个有趣的问题是为什么 flatMap returns 没有项目并且

成功完成

Observable.range(0, 10).flatMap(i -> Observable.fromCallable(() -> null))

并完成

错误

Observable.fromCallable(() -> null)

并且还完成了

的错误

Observable.range(0, 10).flatMap(i -> Observable.error(new RuntimeException()))

我想这就是 rx 开发人员决定在 flatMap 中处理空值的方式，在这里我们正在处理无论如何都不应该使用的东西（编辑：确认是一个错误，请参阅评论）

您如何使用 io 调度程序使用 mockito、rxjava2 在 spring mvc 中测试服务？

How do you test a service in spring mvc with mockito, rxjava2 using an io scheduler?

junit4

mockito

spring-boot

rx-java2