Нужна помощь в написании теста junit для функции, которая возвращает поток Flux в качестве вывода

Question

Мне нужна помощь в написании модульного теста для функции, которая возвращает поток Flux в качестве вывода.

Я пытался использовать StepVerifier, но, думаю, я использую его неправильно.

При необходимости проверить следующую функцию.

public Flux<List<String>> streamCompletedScansAfterLastSubmit(@PathVariable("username") String username) {
    Flux<Long> interval = Flux.interval(Duration.ofSeconds(5));
    Flux<List<String>> completeScans = Flux.fromStream(Stream.generate(() -> scanService.getCompletedScansAfterLastSubmitForUser(username)));
    return Flux.zip(interval, completeScans).map(Tuple2::getT2);
  }

Я попробовал

  public void shouldPublishTheAssessmentStatusOnceFinished() {
    when(scanService.getCompletedScansAfterLastSubmitForUser(Mockito.anyString())).thenReturn(Arrays.asList("Scan1:Success"));
    StepVerifier.create(apiScanController.streamCompletedScansAfterLastSubmit("quays_ka"))
      .expectSubscription()
      .expectNext(Arrays.asList("Scan1:Success"))
      .verifyComplete();
  }

Я получаю ошибку:

java.lang.AssertionError: ожидание «ожидаемое завершение не выполнено» (ожидаемое: onComplete (); фактическое: onNext ([Scan1: успех]))

Каким-то образом сервисный метод вызывается дважды (я проверял это с помощью doAnswer)

Answer 1

Stream.generate(Supplier) генерирует бесконечное значение Stream, поэтому completeScans тоже бесконечно.

Поскольку interval также бесконечно, при использовании фиктивного сервиса сжатие двух результатов приводит к Fluxкоторый выдает ["Scan1:Success"] каждые 5 секунд.

Напротив, ваш StepVerifier ожидает очень конечного Flux одного списка перед завершением, поэтому он завершается ошибкой, как только видит второе вхождениесписок.