13장 Testing

Reactor-test 모듈의 기능을 사용하기 위해서는 build.gradle 파일의 dependencies 블록에 다음 의존성을 추가해야 합니다

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

 

13.1 StepVerifier를 사용한 테스팅

Reactor 에서 가장 일반적인 테스트 방식은, 구독 시점에 해당 Operator 체인이 시나리오대로 동작하는지 테스트하는 것입니다

Reactor Sequence 에서 다음에 발생할 Signal 이 무엇인지, 기대하던 데이터들이 emit 되었는지, 특정 시간 동안 emit 된 데이터가 있는지 등을 단계적으로 테스트할 수 있습니다.

이처럼 Reactor 에서는 Operator 체인의 다양한 동작 방식을 테스트하기 위해 StepVerifier 라는 API 를 제공합니다.

 

1) Signal 이벤트 테스트

Step Verifier 를 이용한 가장 기본적인 테스트 방식은 바로 Reactor Sequence 에서 발생하는 Signal 이벤트를 테스트하는 것입니다

 

example 13-1. 기본 테스트 예제

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

result

• create() 을 통해 테스트 대상 Sequence 를 생성합니다
• expectNext("Hello Reactor"), expectComplete() 를 통해 예상되는 emitted data 와 Signal 의 기댓값을 평가합니다
• verify()를 호출함으로써 전체 Operator 체인의 테스트를 트리거합니다

표13-1 expectXXX() 메서드

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

 

표13-2 verifyXXXX() 메서드

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

 

exmple 13-2 Signal 이벤트 테스트 대상 클래스 (GeneralTestExample.java)

public class GeneralTestExample {
    public static Flux<String> sayHello() {
        return Flux
                .just("Hello", "Reactor");
    }

    public static Flux<Integer> divideByTwo(Flux<Integer> source) {
        return source
                .zipWith(Flux.just(2, 2, 2, 2, 0), (x, y) -> x/y);
    }

    public static Flux<Integer> takeNumber(Flux<Integer> source, long n) {
        return source
                .take(n);
    }
}

example 13_3

@Test
public void sayHelloTest() {
    StepVerifier
            .create(GeneralTestExample.sayHello())
            .expectSubscription()
            .as("# expect subscription")
            .expectNext("Hi")
            .as("# expect Hi")
            .expectNext("Reactor")
            .as("# expect Reactor")
            .verifyComplete();
}

as 는 설명입니다

.as("# expect subscription")

Flux.just("Hello", "Reactor"); Hello 를 emit 하고 
.expectNext("Hi") 를 기대하여, 실패한 것을 알 수 있습니다.

expectation "# expect Hi" failed (expected value: Hi; actual value: Hello)

 

example 13-4 Step Verifier 기본 테스트 예제 3

@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();
}

// result 
4 actionable tasks: 2 executed, 2 up-to-date
2:45:23 PM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_4.divideByTwoTest"'.

source : Flux.just(2, 4, 6, 8, 10);

.zipWith(Flux.just(2, 2, 2, 2, 0), (x, y) -> x/y);

2/2 -> 1

4/2 -> 2

6/2 -> 3

8/2 -> 4

10/0 -> Error 

가 기대대로 출력되었습니다

 

.expectNext(1)
.expectNext(2)
.expectNext(3)
.expectNext(4)

과

.expectNext(1, 2, 3, 4)

는 같은 작업을 수행합니다

 

example 13-5 StepVerifier 기본 테스트 예제 4 

@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)
            .expectNextCount(498)
            .expectNext(500)
            .expectComplete()
            .verify();
}
// result
[Verify from 0 to 499] expectation "expectNext(500)" failed (expected value: 500; actual value: 499)
java.lang.AssertionError: [Verify from 0 to 499] expectation "expectNext(500)" failed (expected value: 500; actual value: 499)

takeNumber 메서드로 테스트 

takeNumber : Source Flux 에서 파라미터로 전달된 숫자의 개수만큼만 데이터를 emit

public static Flux<Integer> takeNumber(Flux<Integer> source, long n) {
    return source
            .take(n);
}

 

StepVerifierOptions 는 이름 그대로 StepVerifier 에 옵션, 즉 추가적인 기능을 덧붙이는 작업을 하는 클래스

예제에서는 테스트에 실패할 경우 파라미터로 입력한 시나리오 명을 출력합니다

StepVerifier
        .create(GeneralTestExample.takeNumber(source, 500),
                StepVerifierOptions.create().scenarioName("Verify from 0 to 499"))

 

example 13_5 결과 설명

• expectSubscription() 으로 구독이 발생했음을 기대합니다.
• expectNext(0) 으로 숫자 0 이 emit 되었음을 기대합니다.
• expectNextCount() 로 498개의 숫자가 emit 되었음을 기대합니다.
• expectNext() 로 숫자500 이 emit 되었음을 기대합니다.

0 , ...498개, 다음 499 가 emit 되는데, 500을 기대하였으므로,  fail 됩니다

[Verify from 0 to 499] expectation "expectNext(500)" failed (expected value: 500; actual value: 499)

• expectComplete() 으로 onComplete Signal 이 전송됨을 기대합니다.

 

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

example 13_6 시간 기반 테스트 대상 클래스 (TimeBasedTestExample.java)

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);
  }
}

 

 

@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();

}

covid-19 확진자 현황을 1시간 마다 emit 하는 코드

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

VirtualTimeScheduler.advanceTimeBy(Duration.ofHours(1)) 를 이용해 1시간 당기는 작업을 수행합니다

1시간 뒤에 받을 데이터를 expectNextCount(11) 로 기대값 검증을 할 수 있습니다

 

example 13-8 StepVerifier 시간 기반 테스트 예제 2 

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

// result 

VerifySubscriber timed out on reactor.core.publisher.FluxFlatMap$FlatMapMain@370413f9

ExampleTest13_8 > getCOVID19CountTest() FAILED
    java.lang.AssertionError at ExampleTest13_8.java:25
1 test completed, 1 failed
> Task :test FAILED

 

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

interval 1분을 지정했으나,

.verify(Duration.ofSeconds(3));

verfify 에서는 3초 밖에 기다리지 않아 AssertError 로 테스트가 실패합니다

 

example 13-9 StepVerifier 시간 기반 테스트 예제 3 

@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();
}
// result : 테스트 성공

 

.withVirtualTime() 사용하며, 1분 마다 투표 현황을 emit 합니다

.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();

 

1분 마다 event 가 없음을 기대하며 테스트 합니다

.expectNoEvent(Duration.ofMinutes(1))

: 1분 동안 어떤 이벤트도 발생하지 않을 것이라고 기대하는 동시에 지정한 시간만큼 시간을 앞당깁니다.

 

1분 간격인데, 1분 마다 이벤트가 발생하지 않습니다 1분이 초과할 때 event 가 발생하나 봅니다

 

.expectNoEvent(Duration.ofMinutes(2))

로 바꾸어 실행하면 아래 같은 오류가 발생합니다

expectation failed (expected no event: onNext([중구,15400]))

 

Backpressure 테스트

코드 13-10 Backpressure 테스트 대상 클래스 ( BackpressureTestExample.java)

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);
  }
}

create() Operator 내부에서 for 문을 이용해 프로그래밍 방식으로 100개의 숫자를 emit 하고 있으며, Backpressure 전략으로 ERROR 전략을 지정했기 때문에 오버플로가 발생하면 OverflowException 이 발생할 것입니다

 

코드 13-11 StepVerifier Backpressure 테스트 예제 1 (ExampleTest13_11.java)

@Test
public void generateNumberTest() {
    StepVerifier
        .create(BackpressureTestExample.generateNumber(), 1L)
        .thenConsumeWhile(num -> num >= 1)
        .verifyComplete();
}
// result 
23:28:29.259 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
23:28:29.292 [Test worker] reactor.core.publisher.Operators DEBUG- onNextDropped: 3
23:28:29.292 [Test worker] reactor.core.publisher.Operators DEBUG- onNextDropped: 4
23:28:29.292 [Test worker] reactor.core.publisher.Operators DEBUG- onNextDropped: 5
...(중략)...
23:28:29.314 [Test worker] reactor.core.publisher.Operators DEBUG- onNextDropped: 99
23:28:29.315 [Test worker] reactor.core.publisher.Operators DEBUG- onNextDropped: 100

expectation "expectComplete" failed (expected: onComplete(); actual: onError(reactor.core.Exceptions$OverflowException: The receiver is overrun by more signals than expected (bounded queue...)))
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...)))
...(중략)...

ExampleTest13_11 > generateNumberTest() FAILED
    java.lang.AssertionError at ExampleTest13_11.java:15
1 test completed, 1 failed

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

StepVerifier 의 .create(BackpressureTestExample.generateNumber(), 1L) 에서 1개의 데이터만 요청하여,

오버플로가 발생했기 때문에 테스트 결과는 failed 가 됩니다

.thenConsumeWhile(num -> num >=1)  을 통해 emit 되는 데이터를 소비하고 있지만, 예상한 것보다 더 많은 데이터를 수신함으로써 결국에서 오버플로가 발생한 것입니다

 

코드 13-12 StepVerifier Backpressure 테스트 예제 2 (Example13_112.java)

@Test
public void generateNumberTest() {
    StepVerifier
        .create(BackpressureTestExample.generateNumber(), 1L)
        .thenConsumeWhile(num -> num >= 1)
        .expectError()
        .verifyThenAssertThat()
        .hasDroppedElements();

}
// result
23:38:33.510 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
23:38:33.527 [Test worker] reactor.core.publisher.Hooks DEBUG- Hooking new default : onErrorDropped
23:38:33.528 [Test worker] reactor.core.publisher.Hooks DEBUG- Hooking new default : onNextDropped
23:38:33.528 [Test worker] reactor.core.publisher.Hooks DEBUG- Hooking onOperatorError: reactor.test.DefaultStepVerifierBuilder$HookRecorder$$Lambda$367/0x000000080027e040@6ab62c94
23:38:33.544 [Test worker] reactor.core.publisher.Hooks DEBUG- Reset to factory defaults : onNextDropped
23:38:33.545 [Test worker] reactor.core.publisher.Hooks DEBUG- Reset to factory defaults : onErrorDropped
23:38:33.545 [Test worker] reactor.core.publisher.Hooks DEBUG- Reset to factory defaults : onOperatorError
BUILD SUCCESSFUL in 2s
4 actionable tasks: 2 executed, 2 up-to-date
11:38:33 PM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_12.generateNumberTest"'.

13-12는 오버플로로 인해 에러가 발생함을 기대하는 예제입니다

먼저 expectError() 를 통해 에러를 기대합니다.

그리고 오버플로로 인해 내부적으로 Drop 된 데이터가 있음을 Assertion 합니다

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

코드 13-12 에서는 hasDroppedElements() 메서드를 이용해서 Drop 된 데이터가 있음을 Assertion 합니다.

 

3) Conext 테스트

코드 13-13 Context 테스트 대상 클래스 ( ContextTestExample.java )

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(Base64Utils.decodeFromString(tp.getT2())))
        )
        .transformDeferredContextual(
            (mono, ctx) -> mono.map(notUse -> ctx.get("secretMessage"))
        );
  }
}

Context 에는 두 개의 데이터가 저장되었는데, 하나는 Base64 형식으로 인코딩된 secret key가 저장되고, 또 하나는 secret key에 해당하는 secret message 가 저장되어 있습니다

getSecretMessage() 메서드는 파라미터로 입력받은 Mono<String> keySource 와 Context 에 저장된 secret key 의 값을 비교해서 일치하면 Context 에 저장된 Mono<String> secretMessage 를 리턴해 줍니다

getSecretMessage() 에서 Context 에 저장된 데이터를 정상적으로 사용하는지 테슽해 보겠습니다

 

코드 13-14 StepVerifier Context 테스트 예제 ( ExampleTest13_14.java )

@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();
}
// result
23:52:17.891 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
BUILD SUCCESSFUL in 2s
4 actionable tasks: 3 executed, 1 up-to-date
11:52:18 PM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_14.getSecretMessageTest"'.

.contextWrite(context -> context.put("secretKey", "aGVsbG8="))
에서 hello 를 Base64 로 인코딩한 문자열을 context 에 기록합니다

 

Mono<String> source = Mono.just("hello");

에서 hello 를 emit 하는 Mono source

.getSecretMessage(source) 을 통해 호출합니다

 

• expectSubscription() 으로 구독이 발생함을 기대합니다
• expectAccessibleContext() 로 구독 이후, Context 가 전파됨을 기대 합니다
• hasKey() 로 전파된 Context 에 "secretKey" 키에 해당하는 값이 있음을 기대합니다
• hasKey() 로 전파된 Context 에 "secretMessage" 키에 해당하는 값이 있음을 기대합니다
• then() 메서드로 Sequence 의 다음 Signal 이벤트의 기댓값을 평가할 수 있도록 합니다
• expectNext() 로 "Hello, Reactor" 문자열이 emit 되었음을 기대합니다
• expectComplete() 으로 onComplete Signal 이 전송됨을 기대합니다.

테스트 결과는 성공입니다

 

4) Recored 기반 테스트

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

recordWith() 를 사용합니다

파라미터로 전달한 Java 의 컬렉션에 emit 된 데이터를 추가(기록)하는 세션을 시작합니다

컬렉션에 기록된 데이터에 다양한 조건을 지정함으로써 emit 된 데이터를 Assertion 할 수 있습니다.

 

13-15 Record 기반 테스트 대상 클래스 (RecordTestExample.java)

public class RecordTestExample {

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

getCapitalizedCountry() : 첫 글자를 대문자로 변환

 

코드 13-16 StepVerifier Record 기반 테스트 예제 1 ( ExampleTest13_16.java )

@Test
public void getCountryTest() {
  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();
}

// result : 성공

10:07:58.122 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
BUILD SUCCESSFUL in 9s
4 actionable tasks: 3 executed, 1 up-to-date
10:07:58 AM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_16.getCountryTest"'.

 

• expectSubscription() 으로 구독이 발생함을 기대합니다.
• recordWith(ArrayList::new) 로 emit 된 데이터에 대한 기록을 시작합니다
• thenConsumeWhile() 로 파라미터로 전달한 Predicate 과 일치하는 데이터는 다음 단계에서 소비할 수 있도록 합니다
• consumeRecordedWith()  컬렉션에 기록된 데이터를 소비합니다. 여기서는 모든 데이터의 첫 글자가 대문자인지 여부를 와 assertThat 으로 확인함으로써 getCapitalizedCountry() 메서드를 Assertion 합니다.
• expectComplete() 으로 onComplte Signal 이 전송됨을 기대합니다.

 

코드 13-17 StepVerifier Record 기반 테스트 예제 2 ( ExampleTest13_17.java )

@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();
}

// result 

10:24:03.979 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
BUILD SUCCESSFUL in 2s
4 actionable tasks: 2 executed, 2 up-to-date
10:24:04 AM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_17.getCountryTest"'.

13-16 과 거의 유사하지만 assertThat 대신 expectRecordedMatches() 를 사용하여 코드가 조금 더 간결해 졌습니다

 

기억하세요

StepVerifier 를 사용한 테스트 유형

• Reactor Sequence 에서 발생하는 Signal 이벤트를 테스트 할 수 있다.
• withVirtualTime() 과 virtualTimeScheduler() 등을 이용해서 시간 기반 테스트를 진행할 수 있다.
• thenConsumeWhile() 을 이용해서 Backpressure 테스트를 진행할 수 있다.
• expectAccessibleContext() 를 이용해서 Sequence 에 전파되는 Context 를 테스트 할 수 있다.
• recordWith(), consumeRecordedWith(), expectRecordedMatches() 등을 이용해서 Record 기반 테스트를 진행할 수 있다.

 

 

13.2 TestPublisher를 사용한 테스팅

reactor-test 모듈에서 지원하는 테스트 전용 Publisher 인 TestPublisher를 이용해 테스트를 진행할 수 있습니다.

 

정상 동작하는 (Well-behaved) TestPublisher

정상 동작하는 TestPublisher 라는 말의 의미는 emit 하는 데이터가 Null 인지, 요청하는 개수보다 더 많은 데이터를 emit 하는지 등의 리액티브 스트림즈 사양 위반 여부를 사전에 체크한다는 의미입니다

 

코드 13-18 정상 동작하는 TestPublisher 예제 ( ExampleTest13_18.java )

@Test
public void divideByTwoTest() {
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();
}
// result
10:47:32.149 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
BUILD SUCCESSFUL in 2s
4 actionable tasks: 2 executed, 2 up-to-date
10:47:32 AM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_18.divideByTwoTest"'.

• Testpublisher.create() 으로 TestPublisher 를 생성합니다 
• source.flux() : 테스트 대상 클래스에 파라미터로 Flux를 전달하기 위해 flux() 메서드를 이용하여 Flux 로 변환합니다  
• .then(() -> source.emit(2,4,6,8,10)) 을 사용해서 테스트에 필요한 데이터를 emit 합니다

 

TestPublisher 를 사용했다는 점 말고는 테스트 시나리오 자체가 코드 13-4와 유사합니다

 

TestPublisher 가 발생시키는 Signal 유형

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

 

오동작하는 (Misbehavingt) TestPublisher

오동작하는 TestPublisher 의 의미는 리액티브 스트림즈 사양 위반 여부를 사전에 체크하지 않는다는 의미입니다. 따라서 리액티브 스트림즈 사양에 위반되더라도 TestPublisher 는 데이터를 emit 할 수 있습니다

 

코드 13-19 오동작하는 TestPublisher 예제 ( ExampleTest13_19.java )

@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);
}

// result 1 : TestPublisher.createNoncompliant(TestPublisher.Violation.ALLOW_NULL);
java.lang.NullPointerException: e
	at java.base/java.util.Objects.requireNonNull(Objects.java:246)
	at reactor.util.concurrent.SpscArrayQueue.offer(SpscArrayQueue.java:51)
	at reactor.core.publisher.FluxZip$ZipInner.onNext(FluxZip.java:909)
	at reactor.test.publisher.DefaultTestPublisher$TestPublisherSubscription.onNext(DefaultTestPublisher.java:233)
	at reactor.test.publisher.DefaultTestPublisher.next(DefaultTestPublisher.java:401)
	at reactor.test.publisher.DefaultTestPublisher.next(DefaultTestPublisher.java:41)
	at chapter13.ExampleTest13_19.lambda$divideByTwoTest$0(ExampleTest13_19.java:26)
	at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
	at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:658)

ExampleTest13_19 > divideByTwoTest() FAILED
    java.lang.NullPointerException at ExampleTest13_19.java:26
    

// result 2 : TestPublisher<Integer> source = TestPublisher.create(); 
emitted values must be non-null
java.lang.NullPointerException: emitted values must be non-null

ExampleTest13_19 > divideByTwoTest() FAILED
    java.lang.NullPointerException at ExampleTest13_19.java:26
1 test completed, 1 failed

result 1 : TestPublisher.createNoncompliant(TestPublisher.Violation.ALLOW_NULL);

오동작하는 TestPublisher 로서 동작하도록 ALOWW_NULL 위반 조건을 지정하여 데이터 값이 null 이라도 정상 동작하는 TestPublisher 를 생성합니다

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

그리고 null 값을 포함하는 데이터 소스를 사용하도록 했습니다

private static List<Integer> getDataSource() {
return Arrays.asList(2, 4, 6, 8, null);
}

결과

java.lang.NullPointerException: e

...

at reactor.test.publisher.DefaultTestPublisher$TestPublisherSubscription.onNext(DefaultTestPublisher.java:233)

...

at chapter13.ExampleTest13_19.lambda$divideByTwoTest$0(ExampleTest13_19.java:26)

 

result 2 : TestPublisher<Integer> source = TestPublisher.create();

정상동작하는 TestPublisher 로 변경하면 , onNext Signal 을 전송하기 전에 Validation 과정을 거쳐 전송할 데이터가  에러가 납니다 

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

 

 

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

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

기억하세요

• TestPublisher 를 사용하면 개발자가 직접 프로그래밍 방식으로 Signal 을 발생시키면서 원하는 상황을 미세하게 재연하며 테스트를 진행할 수 있다.
• 오동작하는 TestPublisher 를 생성하여 리액티브 스트림즈의 사양(Specification)을 위반하는 상황이 발생하는지를 테스트 할 수 있다.

 

13.3 PublisherProbe를 사용한 테스팅

reactor-test 모듈은 PublisherProbe 를 이용해 Sequece 의 실행 경로를 테스트할 수 있습니다

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

 

13-20 PublisherProbe 기반 테스트 대상 클래스 (PublisherProbeTestExample.java)

public class PublisherProbeTestExample {

	public static Mono<String> processTask(Mono<String> main, Mono<String> standby) {
		return main
				.flatMap(massage -> Mono.just(massage))
				.switchIfEmpty(standby);
	}

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

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

processTask() 메서드는 평소에 주전력을 사용해서 작업을 진행하다가 주전력이 끊겼을 경우에만 예비 전력을 사용해서 작업을 진행하는 시뮬레이션합니다.

.switchIfEmpty(standby); Operator 통해, Upstream Publisher 가 데이터 emit 없이 종료되는 경우, 대체 Publisher 가 데이터를 emit 합니다.

 

13-21 PublisherProbe 기반 테스트 예제 (ExampleTest13_21.java)

@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();
}
// result
13:00:49.869 [Test worker] reactor.util.Loggers DEBUG- Using Slf4j logging framework
BUILD SUCCESSFUL in 2s
4 actionable tasks: 2 executed, 2 up-to-date
1:00:50 PM: 실행이 완료되었습니다 ':test --tests "chapter13.ExampleTest13_21.publisherProbeTest"'.

실행 경로를 테스트할 Publisher 를 PublisherProbe.of() 메서드로 래핑합니다

PublisherProbe.of(PublisherProbeTestExample.supplyStandbyPower());

probe.mono() 에서 리턴된 Mono 객체를 processTask() 메서드의 두번째 파라미터로 전달합니다

 

주전력 supplyMainPower() 는 Mono.empty() 이므로 .switchIfEmpty(standby) 에 의해 대체 전력이 emit 됩니다

StepVerifier 를 이용해서 expectNextCount(1) 과 verfiyComplete() 를 테스트 합니다.

 

Publisher 의 실행 경로를 테스트를 위해, 아래 메서드를 통해 probe 가 구독/요청/취소되지 않았음을 테스트 합니다.

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

 

기억하세요

• PublisherProbe 를 사용하면 Sequence 의 실행이 분기되는 상황에서 Publisher 가 어느 경로로 실행되는지 테스트 할 수 있다.
• PublisherProbe 의 assertWasSubscribed(), assertWasRequested(), assertWasNotCancelled() 등을 통해 Sequence 의 기대하는 실행 경로를 Assertion 할 수 있다.