Testing

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

 

의존성 추가(build.gradle)

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

 

StepVerifier

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

 

Signal 이벤트 테스트

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

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

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

 

  다음은 expectXXX()로 시작하는 대표적인 메서드만 정리한 목록입니다.

메서드	설명
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의 요소와 매치됨을 기대합니다.

 

  다음은 테스트 대상 Operator 체인에 대한 검증을 트리거하는 verifyXXX() 메서드입니다.

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

 

  다음은 총 네 가지 단계(Subscription, 두 번의 onNext Signal, onComplete Signal)을 테스트 하는 예제 코드입니다. 첫 번째로 emit된 데이터가 'Hi'라고 기대했는데, 실제 emit된 값은 'Hello'이기 때문에 테스트 결과는 'failed'입니다.

import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;

/**
 * StepVerifier 활용 예제
 */
public class ExampleTest13_3 {
    @Test
    public void sayHelloTest() {
        StepVerifier
                .create(GeneralTestExample.sayHello())
                .expectSubscription()
                .as("# expect subscription")
                .expectNext("Hi")
                .as("# expect Hi")
                .expectNext("Reactor")
                .as("# expect Reactor")
                .verifyComplete();
    }
}

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

• as()를 사용해서 이전 기댓값 평가 단계에 대한 설명(description)을 추가할 수 있습니다. 테스트 실패 시 실패한 단계에 해당하는 설명이 로그로 출력됩니다.

 

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

  StepVerifier는 가상의 시간(Virtual Time)을 이용해 미래에 실행되는 Reactor Sequence의 시간을 앞당겨 테스트할 수 있는 기능을 지원합니다. 다음의 코드는 현재 시점에서 1시간 뒤에 COVID-19 확진자 발생 현황을 체크하고자 하는데 테스트 대상 메서드의 Sequence가 1시간 뒤에 실제로 동작하는지 확인하는 것입니다.

 

  withVirtualTime() 메서드는 VirtualTimeScheduler라는 가상 스케줄러의 제어를 받도록 해줍니다. advanceTimeBy()를 이용해 시간을 1시간 당기는 작업을 수행합니다.

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import reactor.test.scheduler.VirtualTimeScheduler;

import java.time.Duration;

/**
 * StepVerifier 활용 예제
 * - 주어진 시간을 앞당겨서 테스트 한다.
 */
public 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();

    }
}

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

Note. Tuples

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

 

Backpressure 테스트

• hasDropped(), hasDiscarded() 등을 이용해서 backpressure 테스트를 수행할 수 있습니다.

  다음은 generateNumber() 메서드는 한 번에 100개의 숫자 데이터를 emit하는데, StepVerifier의 create() 메서드에서 데이터의 요청 개수를 1로 지정해서 오버플로가 발생하는 예제 코드입니다.

import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;

/**
 * StepVerifier Backpressure 테스트 예제
 */
public class ExampleTest13_12 {
    @Test
    public void generateNumberTest() {
        StepVerifier
                .create(BackpressureTestExample.generateNumber(), 1L)
                .thenConsumeWhile(num -> num >= 1)
                .expectError()          // 에러를 기대합니다.
                .verifyThenAssertThat() // 검증을 트리거하고 난 후 추가적인 Assertion을 합니다.
                .hasDroppedElements();  // Drop된 데이터가 있음을 Assertion 합니다.

    }
}

public class BackpressureTestExample {
  public static Flux<Integer> generateNumber() {
    return Flux
           .create(emitter -> {
               for (int i = 1; i <= 100; i++) {
                   emitter.next(i);
               } // 100개의 숫자를 emit 합니다.
               emitter.complete();
           }, FluxSink.OverflowStrategy.ERROR); // Backpressure 전략
  }
}

 

Context 테스트

• expectAccessibleContext()를 이용해서 접근 가능한 Context가 있는지 테스트 할 수 있습니다.
• hasKey()를 사용하여 Context의 key가 존재하는지 검증할 수 있습니다.

  다음은 Context를 테스트하는 예제 코드입니다. Context에는 두 개의 데이터가 저장되었는데, 하나는 Base64 형식으로 인코딩 된 secret key가 저장되고, 또 하나는 secret key에 해당하는 secret message가 저장되어 있습니다.

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

/**
 * StepVerifier Context 테스트 예제
 */
public 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="))
                        // aGVsbG8=: "hello"를 Base64로 인코딩한 문자열입니다.
                )
                .expectSubscription()         // 구독이 발생함을 기대합니다.
                .expectAccessibleContext()    // Context가 전파됨을 기대합니다.
                .hasKey("secretKey")          // "secretKey" 키에 해당하는 값이 있음을 기대합니다.
                .hasKey("secretMessage")      // "secretMessage" 키에 해당하는 값이 있음을 기대합니다.
                .then()                       // Sequence 다음 Signal 이벤트의 기댓값을 평가할 수 있습니다.
                .expectNext("Hello, Reactor") // "Hello, Reactor" 문자열이 emit 되었음을 기대합니다.
                .expectComplete()             // onComplete Signal이 전송됨을 기대합니다.
                .verify();
    }
}

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

 

Record 기반 테스트

• recordWith()를 사용하여 emit된 데이터를 기록할 수 있습니다.
• consumeRecordedWith()를 사용하여 기록된 데이터들을 소비하며 검증할 수 있습니다.
• expectRecordedMatches()를 사용하여 기록된 데이터의 컬렉션을 검증할 수 있습니다.

  다음은 Record 기반 테스트 예제 코드입니다. 예제 코드에서 getCapitalizedCountry() 메서드는 알파벳으로 된 국가명을 전달받아 첫 글자를 대문자로 변환하도록 정의된 Flux를 리턴합니다.

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import java.util.ArrayList;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;

/**
 * StepVerifier Record 테스트 예제
 */
public class ExampleTest13_16 {
    @Test
    public void getCountryTest() {
        StepVerifier
                .create(RecordTestExample.getCapitalizedCountry(
                        Flux.just("korea", "england", "canada", "india")))
                .expectSubscription()
                // emit된 데이터에 대한 기록을 시작합니다.
                .recordWith(ArrayList::new)
                // 파라미터로 전달한 Predicate과 일치하는 데이터는 다음 단계에서 소비할 수 있도록 합니다.
                .thenConsumeWhile(country -> !country.isEmpty())
                // 컬렉션에 기록된 데이터를 소비합니다.
                .consumeRecordedWith(countries -> {
                    assertThat(
                            countries
                                    .stream()
                                    .allMatch(country ->
                                            Character.isUpperCase(country.charAt(0))),
                            is(true)
                    );
                })
                .expectComplete()
                .verify();
    }
}

 

  다음은 테스트 시나리오는 같지만 expectRecordedMatches() 메서드를 사용한 예제 코드입니다.

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import java.util.ArrayList;

/**
 * StepVerifier Record 테스트 예제
 */
public 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();
    }
}

 

TestPublisher

• Testing 목적에 사용하기 위한 Publisher 입니다.
• 개발자가 직접 프로그래밍을 통해 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을 발생시킵니다.

 

  다음 코드는 오동작하는 TestPublisher로서 동작하도록 ALLOW_NULL 위반 조건을 지정하여 데이터의 값이 null이라도 정상 동작하는 TestPublisher를 생성합니다. 정상 동작하는 Publisher의 경우(주석 처리된 코드), onNext Signal을 전송하기 전에 Validation 과정을 거쳐 전송할 데이터가 null이면 NullPointerException을 던집니다.

import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;

import java.util.Arrays;
import java.util.List;

/**
 * 오동작 하는 TestPublisher 예제
 */
public class ExampleTest13_19 {
    @Test
    public void divideByTwoTest() {
//        TestPublisher<Integer> source = TestPublisher.create(); // 정상 동작 Publisher
        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);
    }
}

 

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

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

 

PublisherProbe를 사용한 테스팅

• Operator 체인의 실행 경로를 검증할 수 있습니다.
• 주로 조건에 따른 분기로 인해서 Sequence가 분기 되는 경우, 실행 경로를 추적해서 정상적으로 실행이 되었는지 확인할 수 있습니다.
• 해당 실행 경로대로 정상적으로 실행되었는지의 여부는 assertWasSubscribed(), assertWasRequested(), assertWasCancelled()를 통해 검증할 수 있습니다.

  다음은 PublisherProbe를 사용한 테스트 예제 코드입니다.

import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import reactor.test.publisher.PublisherProbe;

/**
 * PublisherProbe 예제
 */
public class ExampleTest13_21 {
    @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 class PublisherProbeTestExample {
  public static Mono<String> processTask(Mono<String> main, Mono<String> standby) {
    return main
           .flatMap(message -> Mono.just(message))
           .switchIfEmpty(standby); // Upstream Publisher가 데이터 emit 없이 종료되는 경우, 대체 Publisher가 데이터를 emit 합니다.
  }
  
  public static Mono<String> supplyMainPower() {
    return Mono.empty();
  }
  
  public static Mono supplyStandbyPower() {
    return Mono.just("# supply Standby Power");
  }
}

 

 

 

[참고 정보]

• 스프링으로 시작하는 리액티브 프로그래밍
• 전체 소스코드