StepVerifier를 사용한 테스팅
Reactor에서 가장 일반적인 테스트 방식은
Flux 또는 Mono를 Reactor Sequence로 정의한 후,
구독 시점에 해당 Operator 체인이 시나리오대로 동작하는지를 테스트하는 것입니다.
- Operator 체인의 다양한 동작 방식을 테스트하기 위해
StepVerifierAPI 제공testImplementation 'io.projecctreactor:reactor-test'
Signal 이벤트 테스트
@Test
public void sayHelloReactorTest() {
StepVerifier
.create(Mono.just("Hello Reactor")) // 테스트 대상 Sequence 생성
.expectNext("Hello Reactor") // emit 된 데이터 검증
.expectComplete() // onComplete Signal 검증
.verify(); // 검증 실행.
}expect 메서드 목록
| 메서드 | 설명 |
|---|---|
| expectSubscription() | 구독이 이루어짐을 기대한다. |
| expectNext(T t) | onNext Signal을 통해 전달되는 값이 파라미터로 전달된 값과 같음을 기대한다. |
| expectComplete() | onComplete Signal이 전송되기를 기대한다. |
| expectError() | onError Signal이 전송되기를 기대한다. |
| expectNextCount(long count) | 구독 시점 또는 이전 expectNext()를 통해 기대값이 평가된 데이터 이후부터 emit 된 데이터 수를 기대한다. |
| expectNoEvent(Duration duration) | 주어진 시간 동안 Signal 이벤트가 발생하지 않았음을 기대한다. |
| expectAccessibleContext() | 구독 시점 이후에 Context가 전파되었음을 기대한다. |
| expectNextSequence(Iterable<? extends T> iterable) | emit된 데이터들이 파라미터로 전달된 Iterable의 요소와 매치됨을 기대한다. |
verify 메서드 목록
| 메서드 | 설명 |
|---|---|
| verify() | 검증을 트리거한다. |
| verifyComplete() | 검증을 트리거하고, onComplete Signal을 기대한다. |
| verifyError() | 검증을 트리거하고, onError Signal을 기대한다. |
| verify(Duration duration) | 검증을 트리거하고, 주어진 시간이 초과되어도 Publisher가 종료되지 않음을 기대한다. |
| verifyThenAssertThat() | 검증을 트리거하고, 추가적인 Assertion을 할 수 있다. |
Test Description 넣기
as()StepVerifier .create(GeneralTestExample.sayHello()) .expectSubscription() .as("# expect subscription") .expectNext("Hi") .as("# expect Hi") .expectNext("Reactor") .as("# expect Reactor") .verifyComplete();StepVerifierOptions.create().scenarioName()Flux<Integer> source = Flux.range(0, 1000); StepVerifier .create(GeneralTestExample.takeNumber(source, 500), StepVerifierOptions.create().scenarioName("Verify from 0 to 499")) .expectSubscription() .expectNext(0) .expectNextCount(498) .expectNext(500) .expectComplete() .verify();
Time-based 테스트
- StepVerifier는 가상의 시간을 이용해 미래에 실행되는 Reactor Sequence의 시간을 앞당겨 테스트할 수 있는 기능을 지원한다.
withVirtualTime()VirtualTimeScheduler
StepVerifier
.withVirtualTime(() -> TimeBasedTestExample.getCOVID19Count(
Flux.interval(Duration.ofHours(1)).take(1)
)
)
.expectSubscription()
.then(() -> VirtualTimeScheduler
.get()
.advanceTimeBy(Duration.ofHours(1)))
.expectNextCount(11)
.expectComplete()
.verify();withVirtualTime()- VirtualTimeScheduler라는 가상 스케줄러의 제어를 받도록 해준다.
VirtualTImeScheduler.get().advanceTimeBy()then()메서드를 사용해서, 1시간을 당기는 작업을 수행
StepVerifier
.create(TimeBasedTestExample.getCOVID19Count(
Flux.interval(Duration.ofMinutes(1)).take(1)
)
)
.expectSubscription()
.expectNextCount(11)
.expectComplete()
.verify(Duration.ofSeconds(3));
/*
java.lang.AssertionError: VerifySubscriber timed out on reactor.core.publisher.FluxFlatMap$FlatMapMain@564928fc
*/verify(Duration duration)- 테스트 대상 메서드에 대한 기대값을 평가하는 데 걸리는 시간을 제한한다.
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();expectNoEvent(Duration duration)- 지정한 시간동안 어떤 이벤트도 발생하지 않을 것을 기대
- 지정한 시간칸큼 시간을 앞당긴다.
Backpressure 테스트
- Backpressure에 대한 테스트 역시 수행 가능
thenConsumeWhile()
StepVerifier
.create(BackpressureTestExample.generateNumber(), 1L)
.thenConsumeWhile(num -> num >= 1)
.expectError()
.verifyThenAssertThat()
.hasDroppedElements();thenConsumeWhile()메서드를 사용하여 emit되는 데이터를 소비하도록 설정verifyThenAssertThat()메서드를 사용하면 검증을 트리거한고 난 후, 추가적인 Assertion을 할 수 있다.hasDroppedElements()메서드를 이용해서 Drop된 데이터가 있음을 Assertion한다.
Context 테스트
- Context에 대한 테스트 역시 수행 가능
expectAccessibleContext()hasKey()
- Context 테스트 이후,
then()을 호출해서 Signal 이벤트에 대한 평가를 진행할 수 있도록 한다.
@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()
.hasKey("secretKey")
.hasKey("secretMessage")
.then()
.expectNext("Hello, Reactor")
.expectComplete()
.verify();
}
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(Base64Utils.decodeFromString(tp.getT2())))
)
.transformDeferredContextual(
(mono, ctx) -> mono.map(notUse -> ctx.get("secretMessage"))
);
}Record 테스트
recordWith()- emit된 데이터에 대한 기록을 시작한다.
thenConsumeWhile()- 조건을 만족하는 데이터를 다음 단계에서 소비할 수 있도록 한다.
consumeRecordedWith()- 컬렉션에 기록된 데이터를 소비한다.
StepVerifier
.create(RecordTestExample.getCapitalizedCountry(
Flux.just("korea", "england", "canada", "india")))
.expectSubscription()
.recordWith(ArrayList::new)
.thenConsumeWhile(country -> !country.isEmpty())
.consumeRecordedWith(countries -> {
assertThat(
countries
.stream()
.allMatch(country ->
Character.isUpperCase(country.charAt(0))),
is(true)
);
})
.expectComplete()
.verify();expectRecordedMatches()- 컬렉션의에 기록된 모든 데이터에 대한 기대를 확인한다.
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();TestPublisher를 사용한 테스팅
TestPublisher를 사용하면 개발자가 직접 프로그래밍 방식으로 Signal을 발생시키면서 원하는 상황을 미세하게 재연하며 테스트를 진행할 수 있다.- TestPublisher가 발생시키는 Signal 종류
next(T),next(T, T...)- 1개 이상의 onNext Signal을 발생시킨다.
emit(T...)- 1개 이상의 onNext Signal을 발생시킨 후, onComplete Signal을 발생시킨다.
complete()- onComplete Signal을 발생시킨다.
error(Throwable e)- onError Signal을 발생시킨다.
TestPublisher<Integer> source = TestPublisher.create();
StepVerifier
.create(GeneralTestExample.divideByTwo(source.flux()))
.expectSubscription()
.then(() -> source.emit(2, 4, 6, 8, 10))
.expectNext(1, 2, 3, 4)
.expectError()
.verify();Misbehaving TestPublisher
TestPublisher.ViolationALLOW_NULL- 전송할 데이터가 null이어도 NPE 발생시키지 않고 다음 호출을 진행할 수 있도록 한다.
CLEANUP_ON_TERMINATE- onComplete, onError, emit 같은 Terminal Signal을 연달아 여러 번 보낼 수 있도록 한다.
DEFER_CANCELLATION- cancel Signal을 무시하고 계속해서 Signal을 emit할 수 있도록 한다.
REQUEST_OVERFLOW- 요청 개수보다 더 많은 Signal이 발생하더라도 IllegalStateException을 발생시키지 않고 다음 호출을 진행할 수 있도록 한다.
TestPublisher<Integer> source =
TestPublisher.createNoncompliant(TestPublisher.Violation.ALLOW_NULL);
StepVerifier
.create(GeneralTestExample.divideByTwo(source.flux()))
.expectSubscription()
.then(() -> {
Arrays.asList(2, 4, 6, 8, null).stream()
.forEach(data -> source.next(data));
source.complete();
})
.expectNext(1, 2, 3, 4, 5)
.expectComplete()
.verify();
}PublisherProbe를 사용한 테스팅
PublisherProbe를 사용하면 Sequence의 실행이 분기되는 상황에서 Publisher가 어느 경로로 싱행되는지 테스트할 수 있다.PublisherProbe.of- 메서드로 테스트할 대상 Publisher를 Wrapping한다.
- PublisherProbe 가 제공하는 Assertion
assertWasSubscribed()assertWasRequested()assertWasNotCanceled()
@Test
public void publisherProbeTest() {
PublisherProbe<String> probe =
PublisherProbe.of(PublisherProbeTestExample.supplyStandbyPower());
StepVerifier
.create(PublisherProbeTestExample
.processTask(
PublisherProbeTestExample.supplyMainPower(),
probe.mono())
)
.expectNextCount(1)
.verifyComplete();
probe.assertWasSubscribed();
probe.assertWasRequested();
probe.assertWasNotCancelled();
}
public static Mono<String> processTask(Mono<String> main, Mono<String> standby) {
return main
.flatMap(Mono::just)
.switchIfEmpty(standby);
}
public static Mono<String> supplyMainPower() {
return Mono.empty();
}
public static Mono supplyStandbyPower() {
return Mono.just("# supply Standby Power");
}
Hello velog!
공감하며 읽었습니다. 좋은 글 감사드립니다.