summaryrefslogtreecommitdiffstats
path: root/rest-services/dmaap-client/src
diff options
context:
space:
mode:
authorIzabela Zawadzka <izabela.zawadzka@nokia.com>2019-05-14 09:05:49 +0200
committerIzabela Zawadzka <izabela.zawadzka@nokia.com>2019-05-14 14:50:57 +0200
commite8beff45a8cde47485d44b692ef579ce2b755f69 (patch)
tree174718c3df4f4d0f23a8afde2a2b75df1d5509b8 /rest-services/dmaap-client/src
parentb1e0ceb789a9c0781bb3b97d4b1ff0e75396fd29 (diff)
Add tests for MR Subscriber
Change-Id: I705bcd5c0a75d9ac637afdb8cd6ac17b0af55c78 Issue-ID: DCAEGEN2-1422 Signed-off-by: Izabela Zawadzka <izabela.zawadzka@nokia.com>
Diffstat (limited to 'rest-services/dmaap-client/src')
-rw-r--r--rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java232
1 files changed, 216 insertions, 16 deletions
diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java
index ab51bfef..a2c000f5 100644
--- a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java
+++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberIT.java
@@ -20,8 +20,10 @@
package org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api;
+import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendError;
import static org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer.sendResource;
+import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import java.time.Duration;
import org.junit.jupiter.api.BeforeAll;
@@ -30,9 +32,13 @@ import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouter
import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSource;
import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterSubscribeResponse;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeResponse;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.netty.http.server.HttpServerRoutes;
import reactor.test.StepVerifier;
/**
@@ -40,36 +46,230 @@ import reactor.test.StepVerifier;
* @since May 2019
*/
class MessageRouterSubscriberIT {
- private MessageRouterSubscriber sut = DmaapClientFactory.createMessageRouterSubscriber(MessageRouterSubscriberConfig.createDefault());
- private static DummyHttpServer server;
+ private static final Duration TIMEOUT = Duration.ofSeconds(10);
+ private static final String ERROR_MESSAGE = "Something went wrong";
+ private static final String CONSUMER_GROUP = "group1";
+ private static final String SUCCESS_CONSUMER_ID = "consumer200";
+ private static final String FAILING_WITH_401_CONSUMER_ID = "consumer401";
+ private static final String FAILING_WITH_403_CONSUMER_ID = "consumer403";
+ private static final String FAILING_WITH_409_CONSUMER_ID = "consumer409";
+ private static final String FAILING_WITH_429_CONSUMER_ID = "consumer429";
+ private static final String FAILING_WITH_500_CONSUMER_ID = "consumer500";
+
+ private static final String CONSUMER_PATH = String.format("/events/TOPIC/%s", CONSUMER_GROUP);
+
+ private static final String SUCCESS_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, SUCCESS_CONSUMER_ID);
+ private static final String FAILING_WITH_401_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, FAILING_WITH_401_CONSUMER_ID);
+ private static final String FAILING_WITH_403_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, FAILING_WITH_403_CONSUMER_ID);
+ private static final String FAILING_WITH_409_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, FAILING_WITH_409_CONSUMER_ID);
+ private static final String FAILING_WITH_429_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, FAILING_WITH_429_CONSUMER_ID);
+ private static final String FAILING_WITH_500_RESP_PATH = String
+ .format("%s/%s", CONSUMER_PATH, FAILING_WITH_500_CONSUMER_ID);
+
+ private static MessageRouterSubscribeRequest mrSuccessRequest;
+ private static MessageRouterSubscribeRequest mrFailingRequest;
+ private MessageRouterSubscriber sut = DmaapClientFactory
+ .createMessageRouterSubscriber(MessageRouterSubscriberConfig.createDefault());
private static MessageRouterSource sourceDefinition;
+
@BeforeAll
static void setUp() {
- server = DummyHttpServer.start(routes ->
- routes.get("/events/TOPIC/group1/consumer8", (req, resp) -> sendResource(resp, "/sample-mr-subscribe-response.json"))
- );
- sourceDefinition = ImmutableMessageRouterSource.builder()
- .name("the topic")
- .topicUrl(String.format("http://%s:%d/events/TOPIC", server.host(), server.port()))
- .build();
+ DummyHttpServer server = DummyHttpServer.start(MessageRouterSubscriberIT::setRoutes);
+
+ sourceDefinition = createMessageRouterSource(server);
+
+ mrSuccessRequest = createSuccessRequest();
+
+ mrFailingRequest = createFailingRequest(FAILING_WITH_500_CONSUMER_ID);
}
@Test
- void testStub() {
- final MessageRouterSubscribeRequest mrRequest = ImmutableMessageRouterSubscribeRequest.builder()
- .sourceDefinition(sourceDefinition)
- .consumerGroup("group1")
- .consumerId("consumer8")
+ void subscriber_shouldGetCorrectResponse(){
+ Mono<MessageRouterSubscribeResponse> response = sut
+ .get(mrSuccessRequest);
+
+ JsonArray expectedItems = new JsonArray();
+ expectedItems.add("I");
+ expectedItems.add("like");
+ expectedItems.add("pizza");
+
+ MessageRouterSubscribeResponse expectedResponse = ImmutableMessageRouterSubscribeResponse
+ .builder()
+ .items(expectedItems)
.build();
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldGetUnauthorizedErrorResponse(){
+ MessageRouterSubscribeRequest request = createFailingRequest(FAILING_WITH_401_CONSUMER_ID);
+ Mono<MessageRouterSubscribeResponse> response = sut.get(request);
+
+ MessageRouterSubscribeResponse expectedResponse = createErrorResponse(String
+ .format("401 Unauthorized\n%s", ERROR_MESSAGE));
+
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldGetForbiddenErrorResponse(){
+ MessageRouterSubscribeRequest request = createFailingRequest(FAILING_WITH_403_CONSUMER_ID);
+ Mono<MessageRouterSubscribeResponse> response = sut.get(request);
+
+ MessageRouterSubscribeResponse expectedResponse = createErrorResponse(String
+ .format("403 Forbidden\n%s", ERROR_MESSAGE));
+
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldGetConflictErrorResponse(){
+ MessageRouterSubscribeRequest request = createFailingRequest(FAILING_WITH_409_CONSUMER_ID);
+ Mono<MessageRouterSubscribeResponse> response = sut.get(request);
+
+ MessageRouterSubscribeResponse expectedResponse = createErrorResponse(String
+ .format("409 Conflict\n%s", ERROR_MESSAGE));
+
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldGetTooManyRequestsErrorResponse(){
+ MessageRouterSubscribeRequest request = createFailingRequest(FAILING_WITH_429_CONSUMER_ID);
+ Mono<MessageRouterSubscribeResponse> response = sut.get(request);
+
+ MessageRouterSubscribeResponse expectedResponse = createErrorResponse(String
+ .format("429 Too Many Requests\n%s", ERROR_MESSAGE));
+
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldGetInternalServerErrorResponse(){
+ Mono<MessageRouterSubscribeResponse> response = sut
+ .get(mrFailingRequest);
+
+ MessageRouterSubscribeResponse expectedResponse = createErrorResponse(String
+ .format("500 Internal Server Error\n%s", ERROR_MESSAGE));
+
+ StepVerifier.create(response)
+ .expectNext(expectedResponse)
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldParseCorrectResponse() {
final Flux<String> result = sut
- .getElements(mrRequest)
+ .getElements(mrSuccessRequest)
.map(JsonElement::getAsString);
StepVerifier.create(result)
.expectNext("I", "like", "pizza")
.expectComplete()
- .verify(Duration.ofSeconds(10));
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldParseErrorResponse(){
+ Flux<String> result = sut
+ .getElements(mrFailingRequest)
+ .map(JsonElement::getAsString);
+
+ StepVerifier.create(result)
+ .expectError(IllegalStateException.class)
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldSubscribeCorrectly(){
+ Flux<String> subscriptionForElements = sut
+ .subscribeForElements(mrSuccessRequest, Duration.ofSeconds(1))
+ .map(JsonElement:: getAsString);
+
+ StepVerifier.create(subscriptionForElements.take(2))
+ .expectNext("I", "like")
+ .expectComplete()
+ .verify(TIMEOUT);
+ }
+
+ @Test
+ void subscriber_shouldParseErrorWhenSubscribed(){
+ Flux<String> subscriptionForElements = sut
+ .subscribeForElements(mrFailingRequest, Duration.ofSeconds(1))
+ .map(JsonElement:: getAsString);
+
+ StepVerifier.create(subscriptionForElements.take(2))
+ .expectError(IllegalStateException.class)
+ .verify(TIMEOUT);
+ }
+
+ private static HttpServerRoutes setRoutes(HttpServerRoutes routes){
+ return routes
+ .get(SUCCESS_RESP_PATH, (req, resp) ->
+ sendResource(resp, "/sample-mr-subscribe-response.json"))
+ .get(FAILING_WITH_401_RESP_PATH, (req, resp) ->
+ sendError(resp, 401, ERROR_MESSAGE))
+ .get(FAILING_WITH_403_RESP_PATH, (req, resp) ->
+ sendError(resp, 403, ERROR_MESSAGE))
+ .get(FAILING_WITH_409_RESP_PATH, (req, resp) ->
+ sendError(resp, 409, ERROR_MESSAGE))
+ .get(FAILING_WITH_429_RESP_PATH, (req, resp) ->
+ sendError(resp, 429, ERROR_MESSAGE))
+ .get(FAILING_WITH_500_RESP_PATH, (req, resp) ->
+ sendError(resp, 500, ERROR_MESSAGE));
+ }
+
+ private static MessageRouterSource createMessageRouterSource(DummyHttpServer server){
+ return ImmutableMessageRouterSource.builder()
+ .name("the topic")
+ .topicUrl(String.format("http://%s:%d/events/TOPIC", server.host(), server.port()))
+ .build();
+ }
+
+ private static MessageRouterSubscribeRequest createSuccessRequest(){
+ return ImmutableMessageRouterSubscribeRequest.builder()
+ .sourceDefinition(sourceDefinition)
+ .consumerGroup(CONSUMER_GROUP)
+ .consumerId(SUCCESS_CONSUMER_ID)
+ .build();
+ }
+
+ private static MessageRouterSubscribeRequest createFailingRequest(String consumerId){
+ return ImmutableMessageRouterSubscribeRequest
+ .builder()
+ .sourceDefinition(sourceDefinition)
+ .consumerGroup(CONSUMER_GROUP)
+ .consumerId(consumerId)
+ .build();
+ }
+
+ private static MessageRouterSubscribeResponse createErrorResponse(String failReason){
+ return ImmutableMessageRouterSubscribeResponse
+ .builder()
+ .failReason(failReason)
+ .build();
}
}