Testing

Reactor에서 Test하기

reactor-test라는 테스트 전용 모듈을 통해 여러 가지 유형의 테스트를 지원한다.

dependencies {
		testImplementation 'io.projectreactor:reactor-test'
}

 

StepVerifier를 사용한 테스팅

Reactor에서 일반 적으로 Flux 또는 Mono를 Reactor Seqence로 정의한 후, 구독 시점에 해당 Operator 체인이 시나리오대로 동작하는지 테스트한다.

• Reactor Seqence에서 다음에 발생한 시그널, 기대하던 데이터들이 emit 되었는지, 특정 시간 동안 emit된 데이터가 있는지 등을 단계적으로 테스트

 

Signal 이벤트 테스트

expectXXXX() 메서드

메서드	설명
expectSubscription()	구독이 이루어짐을 기대한다.
expectNext(T t)	OnNext Signal을 통해 전달되는 값이 파라미터로 전달된 값과 같음을 기대한다.
expectComplete()	onComplete Signal이 전송되기를 기대한다.
expectError()	onError Signal이 전송되기를 기대한다.
expectNextCount(long count)	구독 시점 또는 이전(previous) expectNext()를 통해 기댓값이 평가된 데이터 이후부터 emit된 수를 기대한다.
expectNoEvent(Duration duration)	주어진 시간 동안 Signal 이벤트가 발생하지 않았음을 기대한다.
expectAccessibleContext()	구독 시점 이후에 Context가 전파되었음을 기대한다.
expectNextSequence(Iterable<? extends T> iterable	emit된 데이터들이 파라미터로 전달된 Iterable의 요소와 매치됨을 기대한다.

 

verifyXXXX() 메서드

호출되면 내부적으로 테스트 대상 Operator 체인에 대한 구독이 이루어지고 기댓값을 평가한다.

메서드	설명
verify()	검증을 트리거한다.
verifyComplete()	검증을 트리거하고, onComplete Signal을 기대한다.
verifyError()	검증을 트리거하고, onError Signal을 기대한다.
verifyTimeout(Duration duration)	검증을 트리거하고, 주어진 시간이 초과되어도 Publisher가 종료되지 않음을 기대한다.

 

예제1

public class StepVerifierExam {
    @Test
    public void sayHelloReactorTest() {
        StepVerifier
                .create(Mono.just("Hello Reactor")) // 테스트 대상 Sequence 생성
                .expectNext("Hello Reactor")    // emit 된 데이터 검증
                .expectComplete()   // onComplete Signal 검증
                .verify();          // 검증 실행.
    }
}

 

expectXXXX()를 통해 Sequence에서 예상되는 Signal의 기댓값을 평가한다.

.expectNext("Hello Reactor")
.expectComplete()   // 기대했던 대로 종료되었는지

 

예제2

GeneralTestExample 클래스의 sayHello() 메서드를 테스트

class ExampleTest13_3 {
/**
   * GeneralTestExample 클래스의 sayHello() 메서드를 테스트
   * 
   * <pre>{@code
   * public static Flux<String> sayHello() {
   *     return Flux
   *         .just("Hello", "Reactor");
   * }
   * }</pre>
   *
   * @return "Hello"와 "Reactor"를 반환한다.
*/
  @Test
  public void sayHelloTest() {
    StepVerifier
        .create(GeneralTestExample.sayHello())
        .expectSubscription()
        .as("# expect subscription")
        .expectNext("Hi")
        .as("# expect Hi")
        .expectNext("Reactor")
        .as("# expect Reactor")
        .verifyComplete();
  }
}

 

as()를 사용하면 이전 기댓값 평가 단계에 대한 설명을 추가할 수 있다.

java.lang.AssertionError: expectation "# expect Hi" failed (expected value: Hi; actual value: Hello)

• 테스트를 실패하게 되면 실패한 단계에 해당하는 설명이 로그로 출력된다. (Hi를 기대했지만 Hello라서 실패)

 

예제3

GeneralTestExample 클래스의 divideByTwo() 메서드를 테스트

class ExampleTest13_4 {
/**
   * GeneralTestExample 클래스의 divideByTwo() 메서드를 테스트합니다.
   *
   * <pre>{@code
   * public static Flux<Integer> divideByTwo(Flux<Integer> source) {
   *     return source
   *         .zipWith(Flux.just(2, 2, 2, 2, 0), (x, y) -> x / y);
   * }
   * }</pre>
   *
   * @param source 입력 Flux로, 나눗셈을 수행할 데이터 스트림입니다.
   * @return source Flux에서 emit되는 각각의 데이터를 2로 나눈 결과를 포함하는 Flux를 반환합니다.
   *         마지막에는 0으로 나눗셈을 시도하여 {@link ArithmeticException}이 발생합니다.
   * @throws ArithmeticException 나누기 연산 도중 0으로 나눌 경우 발생합니다.
*/
  @Test
  public void divideByTwoTest() {
    Flux<Integer> source = Flux.just(2, 4, 6, 8, 10);
    StepVerifier
        .create(GeneralTestExample.divideByTwo(source))
        .expectSubscription()
        .expectNext(1)
        .expectNext(2)
        .expectNext(3)
        .expectNext(4)
//                .expectNext(1, 2, 3, 4)
        .expectError()
        .verify();
  }
}

• 마지막에 expectError()를 통해 에러를 기대하며, 10을 0으로 나누게 되면 ArithmeticException()이 발생하기 때문에 passed 된다.

 

expectNext()는 1개 이상의 emit된 데이터를 평가할 수 있기 때문에 아래도 동일한 결과가 나온다.

class ExampleTest13_4 {
  @Test
  public void divideByTwoTest() {
    Flux<Integer> source = Flux.just(2, 4, 6, 8, 10);
    StepVerifier
        .create(GeneralTestExample.divideByTwo(source))
        .expectSubscription()
        .expectNext(1, 2, 3, 4)
        .expectError()
        .verify();
  }
}

 

예제4

class ExampleTest13_5 {
/**
 * GeneralTestExample 클래스의 takeNumber() 메서드를 테스트합니다.
 *
 * <p>이 메서드는 Source Flux에서 파라미터로 전달된 숫자의 개수만큼 데이터를 emit하는 기능을 검증합니다.</p>
 *
 * <pre>{@code
 * public static Flux<Integer> takeNumber(Flux<Integer> source, long n) {
 *     return source
 *         .take(n);
 * }
 * }</pre>
 *
 * @return 지정된 개수만큼 데이터를 emit하는 Flux의 동작을 검증합니다.
 */
  @Test
  public void takeNumberTest() {
    Flux<Integer> source = Flux.range(0, 1000);
    StepVerifier
        .create(GeneralTestExample.takeNumber(source, 500),
            StepVerifierOptions.create().scenarioName("Verify from 0 to 499")) // 테스트 실패시 시나리오명 출력 
        .expectSubscription() // 구독이 발생했음을 기대한다.
        .expectNext(0) // 숫자 0이 emit되었음을 기대한다.
        .expectNextCount(498) // 498개의 숫자가 emit되었음을 기대한다.
        .expectNext(500) // 숫자 500이 emit되었음을 기대한다.
        .expectComplete() // onComplete Signal이 전송됨을 기대한다.
        .verify();
  }
}

 

실행 결과

java.lang.AssertionError: [Verify from 0 to 499] expectation "expectNext(499)" failed (expected value: 499; actual value: 498)

• 테스트 실패하므로 시나리오명이 출력되는 것을 볼 수 있다.

 

테스트 대상 Flux가 총 500개의 숫자를 emit하기 때문에 실패한다.

// .expectNextCount(498) 기존에 실패한 코드
.expectNextCount(499) // 0을 제외했음으로 1~499까지 총 499개

• 숫자 0을 평가하고 나서 expectNextCount()를 통해 그다음부터 emit된 개수는 499개다.

 

마지막으로 평가해야 되는 기댓값은 500이 아니라 0~499 이므로 499여야 한다.

// .expectNext(499) 기존에 실패한 코드
.expectNext(499)

 

시간 기반(Time-based) 테스트

StepVerifier는 Virtual Time을 이용해 미래에 실행되는 Reactor Sequence의 시간을 앞당겨 테스트할 수 있는 기능을 지원한다.

 

시간 기반 테스트 대상 클래스

public class TimeBasedTestExample {
  public static Flux<Tuple2<String, Integer>> getCOVID19Count(Flux<Long> source) {
    return source
        .flatMap(notUse -> Flux.just(
                Tuples.of("서울", 10),
                Tuples.of("경기도", 5),
                Tuples.of("강원도", 3),
                Tuples.of("충청도", 6),
                Tuples.of("경상도", 5),
                Tuples.of("전라도", 8),
                Tuples.of("인천", 2),
                Tuples.of("대전", 1),
                Tuples.of("대구", 2),
                Tuples.of("부산", 3),
                Tuples.of("제주도", 0)
            )
        );
  }

  public static Flux<Tuple2<String, Integer>> getVoteCount(Flux<Long> source) {
    return source
        .zipWith(Flux.just(
                Tuples.of("중구", 15400),
                Tuples.of("서초구", 20020),
                Tuples.of("강서구", 32040),
                Tuples.of("강동구", 14506),
                Tuples.of("서대문구", 35650)
            )
        )
        .map(Tuple2::getT2);
  }
}

• Tuples는 서로 다른 타입의 데이터를 저장할 수 있도록 Reactor에서 제공하는 Collection이다.
  • 총 8개의 데이터를 저장할 수 있는 오버로딩된 of() 메서드를 제공한다.

 

예제1

테스트 대상 메서드의 Sequence가 1시간 뒤에 실제로 동작하는지 확인하는 테스트

class ExampleTest13_7 {
  @Test
  public void getCOVID19CountTest() {
    StepVerifier
        .withVirtualTime(() -> TimeBasedTestExample.getCOVID19Count(
                Flux.interval(Duration.ofHours(1)).take(1)
            )
        )
        .expectSubscription()
        .then(() -> VirtualTimeScheduler
            .get()
            .advanceTimeBy(Duration.ofHours(1)))
        .expectNextCount(11)
        .expectComplete()
        .verify();

  }
}

• 1시간 뒤에 emit된 데이터가 가상 스케줄러와 expectNextCount()를 통해 11개임을 기대한다.

 

withVirtualTime() 메서드는 VirtualTimeScheduler라는 가상 스케줄러의 제어를 받도록 해준다.

.withVirtualTime(() -> TimeBasedTestExample.getCOVID19Count(
				Flux.interval(Duration.ofHours(1)).take(1)
		)
)

 

구독에 대한 기댓값을 평가하고 난 후 then() 메서드를 사용해서 후속 작업을 할 수 있다.

.then(() -> VirtualTimeScheduler
		.get()
		.advanceTimeBy(Duration.ofHours(1)))

• VirtualTimeScheduler의 advanceTimeBy()를 이용해 시간을 1시간 당기는 작업을 수행한다.

 

예제2

지정한 시간 내에 테스트 대상 메서드의 작업이 종료되는지 확인하는 예제

class ExampleTest13_8 {
  @Test
  public void getCOVID19CountTest() {
    StepVerifier
        .create(TimeBasedTestExample.getCOVID19Count(
                Flux.interval(Duration.ofMinutes(1)).take(1)
            )
        )
        .expectSubscription()
        .expectNextCount(11)
        .expectComplete()
        .verify(Duration.ofSeconds(3));
  }
}

 

3초 내에 기댓값의 평가가 끝나지 않으면 시간 초과로 간주한다.

.verify(Duration.ofSeconds(3));

 

1분 뒤에 emit하도록 설정했기 때문에 실패한다.

.create(TimeBasedTestExample.getCOVID19Count(
				Flux.interval(Duration.ofMinutes(1)).take(1)
		)
)

 

시간 초과로 AssertionError가 발생하는걸 볼 수 있다.

java.lang.AssertionError: VerifySubscriber timed out on reactor.core.publisher.FluxFlatMap$FlatMapMain@3c7f66c4

 

예제3

지정한 시간 동안 어떤 Signal 이벤트도 발생하지 않았음을 기대한다.

class ExampleTest13_9 {
    @Test
    public void getVoteCountTest() {
        StepVerifier
                .withVirtualTime(() -> TimeBasedTestExample.getVoteCount(
                                Flux.interval(Duration.ofMinutes(1))
                        )
                )
                .expectSubscription()
                .expectNoEvent(Duration.ofMinutes(1))
                .expectNoEvent(Duration.ofMinutes(1))
                .expectNoEvent(Duration.ofMinutes(1))
                .expectNoEvent(Duration.ofMinutes(1))
                .expectNoEvent(Duration.ofMinutes(1))
                .expectNextCount(5)
                .expectComplete()
                .verify();
    }
}

• onNext Signal 이벤트가 한 번 발생하면 다음 이벤트가 발생하기 전까지 1분 동안 아무 이벤트도 발생하지 않는다.

 

테스트 시간을 줄일 수 있도록 withVirtualTime() 메서드를 통해 VirtualTimeScheduler의 제어를 받는다.

.withVirtualTime(() -> TimeBasedTestExample.getVoteCount(
				Flux.interval(Duration.ofMinutes(1))
		)
)

• 원래라면 실제 테스트가 끝날 때까지 5분의 시간이 소요된다.

 

expectNoEvent()의 파라미터로 시간을 지정한다.

.expectNoEvent(Duration.ofMinutes(1)) // 1분 앞당기고 이후 ["중구", 15400]를 받는다. (통과)
.expectNoEvent(Duration.ofMinutes(1)) // 앞에서 1분 앞당기고 이후 1분을 앞당기고 이후 ["서초구", 20020]를 받는다. 밑에 라인 반복
.expectNoEvent(Duration.ofMinutes(1)) // interval로 1분을 주었기 때문에 통과하며 expectNoEvent() 파리미터로 2분을 주면 실패한다.
.expectNoEvent(Duration.ofMinutes(1))
.expectNoEvent(Duration.ofMinutes(1))

• 지정한 시간 동안 어떤 이벤트도 발생하지 않을 것이라고 기대하는 동시에 지정한 시간만큼 시간을 앞당긴다.
• 1분씩 다섯 번의 expectNoEvent()를 호출함으로 5분의 시간을 앞당긴다.

 

Backpressure 테스트

StepVerifier를 사용하면 BackPressure에 대한 테스트를 수행할 수 있다.

 

BackPressure 테스트 대상 클래스

public class BackpressureTestExample {
  public static Flux<Integer> generateNumber() {
    return Flux
        .create(emitter -> {
          for (int i = 1; i <= 100; i++) {
            emitter.next(i);
          }
          emitter.complete();
        }, FluxSink.OverflowStrategy.ERROR);
  }
}

• BackPressure 전략으로 ERROR 전략을 지정했기 때문에 오버플로가 발생하면 OverflowException이 발생한다.

 

예제1 (failed)

class ExampleTest13_11 {
  @Test
  public void generateNumberTest() {
    StepVerifier
        .create(BackpressureTestExample.generateNumber(), 1L)
        .thenConsumeWhile(num -> num >= 1)
        .verifyComplete();
  }
}

• generateNumber() 메서드는 한 번에 100개의 숫자 데이터를 emit 한다.

 

create() 메서드에서 데이터의 요청 개수를 1로 지정해서 오버플로가 발생한다.

java.lang.AssertionError: expectation "expectComplete" failed (expected: onComplete(); actual: onError(reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)))

• OverflowException으로 인해 onError Signal이 발생한다.

 

예제2 (passed)

오버플로로 인해 에러가 발생함을 기대한다.

class ExampleTest13_12 {
  @Test
  public void generateNumberTest() {
    StepVerifier
        .create(BackpressureTestExample.generateNumber(), 1L)
        .thenConsumeWhile(num -> num >= 1)
        .expectError() // 에러를 기대한다.
        .verifyThenAssertThat()
        .hasDroppedElements();

  }
}

 

verifyThenAssertThat() 메서드를 사용하면 검증을 트리거하고 난 후, 추가적인 Assertion을 할 수 있다.

.verifyThenAssertThat()
.hasDroppedElements();

• hasDroppedElements() 메서드를 통해 Drop된 데이터가 있음을 Assertion 한다.

 

Context 테스트

StepVerifier를 사용해서 Reactor Sequence에서 사용되는 Context를 테스트할 수 있다.

 

Context 테스트 대상 클래스

파라미터에 keySource와 Context에 저장된 secretKey 값을 비교해서 일치하면 Context에 저장된 메시지를 반환한다.

public class ContextTestExample {
  public static Mono<String> getSecretMessage(Mono<String> keySource) {
    return keySource
        .zipWith(Mono.deferContextual(ctx ->
            Mono.just((String) ctx.get("secretKey"))))
        .filter(tp ->
            tp.getT1().equals(
                new String(Base64.getDecoder().decode(tp.getT2())))
        )
        .transformDeferredContextual(
            (mono, ctx) -> mono.map(notUse -> ctx.get("secretMessage"))
        );
  }
}

 

예제 (passed)

class ExampleTest13_14 {
  @Test
  public void getSecretMessageTest() {
    Mono<String> source = Mono.just("hello");

    StepVerifier
        .create(
            ContextTestExample
                .getSecretMessage(source)
                .contextWrite(context ->
                    context.put("secretMessage", "Hello, Reactor"))
                .contextWrite(context -> context.put("secretKey", "aGVsbG8="))
        )
        .expectSubscription() // 구독이 발생함을 기대한다.
        .expectAccessibleContext() // 구독 이후, Context가 전파됨을 기대한다.
        .hasKey("secretKey") // Context에 "secretKey" 키에 해당하는 값이 있음을 기대한다.
        .hasKey("secretMessage") // Context에 "secretMessage" 키에 해당하는 값이 있음을 기대한다.
        .then() // Sequence의 다음 Signal 이벤트의 기댓값을 평가할 수 있도록 한다.
        .expectNext("Hello, Reactor") // "Hello, Reactor" 문자열이 emit 되었음을 기대한다.
        .expectComplete() // onComplete Signal이 전송됨을 기대한다.
        .verify();
  }
}

 

Record 기반 테스트

expectNext()로 emit된 데이터의 단순 기댓값만 평가하는 것이 아니라 조금 더 구체적인 조건으로 Assertion 해야 하는 경우

• recordWith()를 사용할 수 있다.
• recordWith()는 파라미터로 전달한 Java Collection에 emit된 데이터를 추가(기록)하는 세션을 시작한다.
• Collection에 기록된 데이터에 다양한 조건을 지정함으로써 emit된 데이터를 Assertion할 수 있다.

 

Record 기반 테스트 대상 클래스

public class RecordTestExample {
  public static Flux<String> getCapitalizedCountry(Flux<String> source) {
    return source
        .map(country -> country.substring(0, 1).toUpperCase() +
            country.substring(1));
  }
}

• getCapitalizedCountry() 메서드가 알파벳으로 된 국가명을 전달받아서 첫 글자를 대문자로 변환하도록 정의된 Flux를 리턴한다.

 

예제1

recordWith()을 사용해서 emit되는 데이터에 대한 세밀한 테스트가 가능하다.

class ExampleTest13_16 {
  @Test
  public void getCountryTest() {
    StepVerifier
        .create(RecordTestExample.getCapitalizedCountry(
            Flux.just("korea", "england", "canada", "india")))
        .expectSubscription() // 구독이 발생함을 기대한다.
        .recordWith(ArrayList::new) // emit된 데이터에 대한 기록을 시작한다.
        .thenConsumeWhile(country -> !country.isEmpty()) // 조건에 일치하는 데이터는 다음 단계에서 소비할 수 있도록 한다.
        .consumeRecordedWith(countries -> { // 컬렉션에 기록된 데이터를 소비한다.
          assertThat(
              countries
                  .stream()
                  .allMatch(country -> // 첫 글자가 대문자인지 여부를 확인한다.
                      Character.isUpperCase(country.charAt(0))),
              is(true)
          );
        })
        .expectComplete() // onComplete Signal이 전송됨을 기대한다.
        .verify();
  }
}

 

예제2

예제1과 테스트 시나리오는 같지만 더 간결한 코드

class ExampleTest13_17 {
  @Test
  public void getCountryTest() {
    StepVerifier
        .create(RecordTestExample.getCapitalizedCountry(
            Flux.just("korea", "england", "canada", "india")))
        .expectSubscription()
        .recordWith(ArrayList::new)
        .thenConsumeWhile(country -> !country.isEmpty())
        .expectRecordedMatches(countries ->
            countries
                .stream()
                .allMatch(country ->
                    Character.isUpperCase(country.charAt(0))))
        .expectComplete()
        .verify();
  }
}

• expectRecordedMatches() 메서드 내에서 Predicate을 사용하여 Collection에 기록된 모든 데이터의 첫 글자가 대문자임을 기대한다.
• expectRecordedMatches()는 기록된 데이터를 소비하면서 조금 더 다양한 조건으로 테스트할 수 있다.
• consumeRecordedWith() 메서드 내부에서 assertThat()를 사용하거나 expectRecordedMatches()를 사용하면 된다.

 

TestPublisher를 사용한 테스팅

TestPublisher를 사용하면, 개발자가 직접 프로그래밍 방식으로 Signal을 발생시키면서 원하는 상황을 미세하게 재연하며 테스트할 수 있다.

• TestPublisher: reactor-test 모듈에서 지원하는 테스트 전용 Publisher

 

정상 동작(Well-behaved) TestPublisher

• emit하는 데이터가 null인지, 요청하는 개수보다 더 많은 데이터를 emit하는지 등의 리액티브 스트림즈 사양 위반 여부를 사전에 체크한다.

정상 동작하는 TestPublisher 예제

class ExampleTest13_18 {
  @Test
  public void divideByTwoTest() {
    TestPublisher<Integer> source = TestPublisher.create(); // TestPublisher 생성

    StepVerifier
        .create(GeneralTestExample.divideByTwo(source.flux())) // flux로 변환
        .expectSubscription()
        .then(() -> source.emit(2, 4, 6, 8, 10)) // emit() 메서드를 사용해서 테스트에 필요한 데이터를 emit
        .expectNext(1, 2, 3, 4)
        .expectError()
        .verify();
  }
}

• TestPublisher를 사용하면 복잡한 로직이 포함된 대상 메서드를 테스트하거나 조건에 따라서 Signal을 변경해야 되는 등의 특정 상황을 테스트하기가 용이하다.

 

TestPublisher가 발생시키는 Signal 유형

• next(T) 또는 next(T, T…): 1개 이상의 onNext Signal을 발생
• emit(T…): 1개 이상의 onNext Signal을 발생시킨 후, onComplete Signal을 발생
• complete(): onComplete Signal을 발생
• error(Throwable): onError Signal을 발생

 

오동작(Misbehaving) TestPublisher

• 리액티브 스트림즈 사양 위반 여부를 사전에 체크하지 않는다.
• 위반되더라도 TestPublisher는 데이터를 emit할 수 있다.

 

오동작(Misbehaving) TestPublisher를 생성하기 위한 위반(validation) 조건

• ALLOW_NULL: 전송할 데이터가 null이어도 NPE를 발생하지 않고 다음 호출을 진행할 수 있도록 한다.
• CLEANUP_ON_TERMINATE: onComplete, onError, emit 같은 Terminal Signal을 연달아 보낼 수 있도록 한다.
• DEFER_CANCELLATION: cancel Signal을 무시하고 계속 Signal을 emit할 수 있도록 한다.
• REQUEST_OVERFLOW: 요청 개수보다 더 많은 Signal이 발생하더라도 IllegalStateException을 발생시키지 않고 다음 호출을 진행할 수 있도록 한다.

 

오동작하는 TestPublisher 예제

오작동하는 TestPublisher를 생성하여 리액티브 스트림즈의 사양을 위반하는 상황이 발생하는지 테스트

class ExampleTest13_19 {
  @Test
  public void divideByTwoTest() {
//    TestPublisher<Integer> source = TestPublisher.create();
		TestPublisher<Integer> source =
				TestPublisher.createNoncompliant(TestPublisher.Violation.ALLOW_NULL);

    StepVerifier
        .create(GeneralTestExample.divideByTwo(source.flux()))
        .expectSubscription()
        .then(() -> {
          getDataSource().stream()
              .forEach(data -> source.next(data));
          source.complete();
        })
        .expectNext(1, 2, 3, 4, 5)
        .expectComplete()
        .verify();
  }

  private static List<Integer> getDataSource() {
    return Arrays.asList(2, 4, 6, 8, null); // null 값을 포함하는 데이터 소스
  }
}

 

오동작하는 TestPublisher 생성

TestPublisher<Integer> source =
		TestPublisher.createNoncompliant(TestPublisher.Violation.ALLOW_NULL);

• ALLOW_NULL 위반 조건을 지정하여 데이터의 값이 null이라도 정상 동작한다.

 

하지만 onNext Signal을 전송하는 과정에서 NPE가 발생한다.

 

정상 동작하는 TestPublisher를 사용하는 경우

TestPublisher<Integer> source = TestPublisher.create();

 

onNext Signal을 전송하기 전에 Validation 과정을 거쳐 전송할 데이터가 null이면 NPE를 던진다.

java.lang.NullPointerException: emitted values must be non-null

 

PublisherProbe를 사용한 테스팅

reactor-test 모듈은 PublisherProble를 이용해 Sequence의 실행 경로를 테스트할 수 있다.

• 주로 조건에 따라 Sequence가 분기되는 경우, Sequence의 실행 경로를 추적해서 정상적으로 실행되었는지 테스트할 수 있다.

 

PublisherProbe 기반 테스트 대상 클래스

public class PublisherProbeTestExample {
  public static Mono<String> processTask(Mono<String> main, Mono<String> standby) {
    return main
        .flatMap(massage -> Mono.just(massage))
        .switchIfEmpty(standby); // main 작업을 진행할 수 없는 경우 standby로 작업
  }

  public static Mono<String> supplyMainPower() {
    return Mono.empty();
  }

  public static Mono supplyStandbyPower() {
    return Mono.just("# supply Standby Power");
  }
}

• switchIfEmpty() Operator는 Upstream Publisher가 게이터 emit 없이 종료되는 경우, 대체 Publisher가 데이터를 emit한다.

 

예제(passed)

Sequence가 분기되는 상황에서 실제로 어느 Publisher가 동작하는지 해당 Publisher의 실행 경로를 테스트

• 즉, switchIfEmpty() 메서드가 데이터를 emit하고 정상적으로 종료되는지 테스트

class ExampleTest13_21 {
  @Test
  public void publisherProbeTest() {
	  // 실행 경로를 Test할 대상 Publisher를 PublisherProbe.of() 메서드로 래핑한다.
    PublisherProbe<String> probe =
        PublisherProbe.of(PublisherProbeTestExample.supplyStandbyPower());

    StepVerifier
        .create(PublisherProbeTestExample
            .processTask(
                PublisherProbeTestExample.supplyMainPower(),
                probe.mono())
        )
        .expectNextCount(1)
        .verifyComplete();

    probe.assertWasSubscribed();
    probe.assertWasRequested();
    probe.assertWasNotCancelled();
  }
}

 

기대하는 Publisher가 구독을 했는지, 요청을 했는지, 중간에 취소가 되지 않았는지 Assertion 한다.

probe.assertWasSubscribed();
probe.assertWasRequested();
probe.assertWasNotCancelled();

• Publisher의 실행 경로를 테스트

 

supplyMainPower() 메서드가 Mono.empty()를 리턴하기 때문에 supplyStandbyPower()가 동작한다.

.create(PublisherProbeTestExample
		.processTask(
				PublisherProbeTestExample.supplyMainPower(),
				probe.mono()) // supplyStandbyPower()를 PublisherProbe.of()로 래핑했다.
)