From 7f83d709f540876fe290614ef47d9b91044936bd Mon Sep 17 00:00:00 2001 From: pwielebs Date: Fri, 28 Feb 2020 11:50:07 +0100 Subject: Improve coverage in SDK (2) Restored test classes: - MessageRouterSubcriberTest - MessageRouterPublisherTest Issue-ID: DCAEGEN2-2087 Signed-off-by: Piotr Wielebski Change-Id: If1923b58e7b0b706f93fdaa0cfddad5dca28976e --- .../client/api/MessageRouterPublisherTest.java | 222 +++++++++++++++++ .../client/api/MessageRouterSubscriberTest.java | 276 +++++++++++++++++++++ 2 files changed, 498 insertions(+) create mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java create mode 100644 rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java (limited to 'rest-services/dmaap-client/src') diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java new file mode 100644 index 00000000..1268a16a --- /dev/null +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterPublisherTest.java @@ -0,0 +1,222 @@ +/* + * ============LICENSE_START==================================== + * DCAEGEN2-SERVICES-SDK + * ========================================================= + * Copyright (C) 2019 Nokia. All rights reserved. + * ========================================================= + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END===================================== + */ + +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.sendString; + +import com.google.gson.JsonElement; +import com.google.gson.JsonPrimitive; +import io.vavr.collection.List; + +import java.time.Duration; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSink; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.MessageRouterSink; +import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.test.DummyHttpServer; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.ContentType; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishRequest; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.ImmutableMessageRouterPublishResponse; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishRequest; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse; +import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +/** + * @author Piotr Jaszczyk + * @since May 2019 + */ +class MessageRouterPublisherTest { + + private static final String ERROR_MESSAGE = "Something went wrong"; + private static final String SUCCESS_RESP_TOPIC_PATH = "/events/TOPIC"; + private static final String FAILING_WITH_400_RESP_PATH = "/events/TOPIC400"; + private static final String FAILING_WITH_401_RESP_PATH = "/events/TOPIC401"; + private static final String FAILING_WITH_403_RESP_PATH = "/events/TOPIC403"; + private static final String FAILING_WITH_404_RESP_PATH = "/events/TOPIC404"; + private static final String FAILING_WITH_500_TOPIC_PATH = "/events/TOPIC500"; + private static final Duration TIMEOUT = Duration.ofSeconds(10); + private static final Flux messageBatch = Flux.just("ala", "ma", "kota") + .map(JsonPrimitive::new); + private static final List messageBatchItems = List.of("ala", "ma", "kota"); + + private static DummyHttpServer server; + private MessageRouterPublisher sut = DmaapClientFactory + .createMessageRouterPublisher(MessageRouterPublisherConfig.createDefault()); + + + @BeforeAll + static void setUp() { + server = DummyHttpServer.start(routes -> + routes.post(SUCCESS_RESP_TOPIC_PATH, (req, resp) -> sendString(resp, Mono.just("OK"))) + .post(FAILING_WITH_400_RESP_PATH, (req, resp) -> + sendError(resp, 400, ERROR_MESSAGE)) + .post(FAILING_WITH_401_RESP_PATH, (req, resp) -> + sendError(resp, 401, ERROR_MESSAGE)) + .post(FAILING_WITH_403_RESP_PATH, (req, resp) -> + sendError(resp, 403, ERROR_MESSAGE)) + .post(FAILING_WITH_404_RESP_PATH, (req, resp) -> + sendError(resp, 404, ERROR_MESSAGE)) + .post(FAILING_WITH_500_TOPIC_PATH, (req, resp) -> + sendError(resp, 500, ERROR_MESSAGE)) + ); + } + + @Test + void test_put_givenMessageBatch_shouldMakeSuccessfulPostRequestReturningBatch() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(SUCCESS_RESP_TOPIC_PATH, + ContentType.TEXT_PLAIN); + final List expectedItems = messageBatchItems.map(JsonPrimitive::new); + + + //when + final Flux result = sut.put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(ImmutableMessageRouterPublishResponse.builder().items(expectedItems).build()) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void publisher_shouldHandleBadRequestError() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_400_RESP_PATH, + ContentType.TEXT_PLAIN); + final MessageRouterPublishResponse expectedResponse = createErrorResponse( + "400 Bad Request\n%s", ERROR_MESSAGE); + + //when + final Flux result = sut.put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void publisher_shouldHandleUnauthorizedError() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_401_RESP_PATH, + ContentType.TEXT_PLAIN); + final MessageRouterPublishResponse expectedResponse = createErrorResponse( + "401 Unauthorized\n%s", ERROR_MESSAGE); + + //when + final Flux result = sut.put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void publisher_shouldHandleForbiddenError() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_403_RESP_PATH, + ContentType.TEXT_PLAIN); + final MessageRouterPublishResponse expectedResponse = createErrorResponse( + "403 Forbidden\n%s", ERROR_MESSAGE); + + //when + final Flux result = sut + .put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void publisher_shouldHandleNotFoundError() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_404_RESP_PATH, + ContentType.TEXT_PLAIN); + final MessageRouterPublishResponse expectedResponse = createErrorResponse( + "404 Not Found\n%s", ERROR_MESSAGE); + + //when + final Flux result = sut + .put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void publisher_shouldHandleInternalServerError() { + //given + final MessageRouterPublishRequest mrRequest = createMRRequest(FAILING_WITH_500_TOPIC_PATH, + ContentType.TEXT_PLAIN); + final MessageRouterPublishResponse expectedResponse = createErrorResponse( + "500 Internal Server Error\n%s", ERROR_MESSAGE); + + //when + final Flux result = sut + .put(mrRequest, messageBatch); + + //then + StepVerifier.create(result) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + + private MessageRouterPublishRequest createMRRequest(String topicPath, ContentType contentType) { + final MessageRouterSink sinkDefinition = ImmutableMessageRouterSink.builder() + .name("the topic") + .topicUrl(String.format("http://%s:%d%s", + server.host(), + server.port(), + topicPath) + ) + .build(); + + return ImmutableMessageRouterPublishRequest.builder() + .sinkDefinition(sinkDefinition) + .contentType(contentType) + .build(); + } + + private MessageRouterPublishResponse createErrorResponse(String failReasonFormat, Object... formatArgs) { + return ImmutableMessageRouterPublishResponse + .builder() + .failReason(String.format(failReasonFormat, formatArgs)) + .build(); + } +} + diff --git a/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java new file mode 100644 index 00000000..18584789 --- /dev/null +++ b/rest-services/dmaap-client/src/test/java/org/onap/dcaegen2/services/sdk/rest/services/dmaap/client/api/MessageRouterSubscriberTest.java @@ -0,0 +1,276 @@ +/* + * ============LICENSE_START==================================== + * DCAEGEN2-SERVICES-SDK + * ========================================================= + * Copyright (C) 2019 Nokia. All rights reserved. + * ========================================================= + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END===================================== + */ + +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.JsonElement; +import com.google.gson.JsonPrimitive; +import io.vavr.collection.List; + +import java.time.Duration; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.onap.dcaegen2.services.sdk.model.streams.dmaap.ImmutableMessageRouterSource; +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; + +/** + * @author Piotr Jaszczyk + * @since May 2019 + */ +class MessageRouterSubscriberTest { + 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() { + DummyHttpServer server = DummyHttpServer.start(MessageRouterSubscriberTest::setRoutes); + + sourceDefinition = createMessageRouterSource(server); + + mrSuccessRequest = createSuccessRequest(); + + mrFailingRequest = createFailingRequest(FAILING_WITH_500_CONSUMER_ID); + } + + @Test + void subscriber_shouldGetCorrectResponse() { + Mono response = sut + .get(mrSuccessRequest); + + List expectedItems = List.of("I", "like", "pizza"); + + MessageRouterSubscribeResponse expectedResponse = ImmutableMessageRouterSubscribeResponse + .builder() + .items(expectedItems.map(JsonPrimitive::new)) + .build(); + + StepVerifier.create(response) + .expectNext(expectedResponse) + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void subscriber_shouldGetUnauthorizedErrorResponse() { + MessageRouterSubscribeRequest request = createFailingRequest(FAILING_WITH_401_CONSUMER_ID); + Mono 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 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 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 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 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 result = sut + .getElements(mrSuccessRequest) + .map(JsonElement::getAsString); + + StepVerifier.create(result) + .expectNext("I", "like", "pizza") + .expectComplete() + .verify(TIMEOUT); + } + + @Test + void subscriber_shouldParseErrorResponse() { + Flux result = sut + .getElements(mrFailingRequest) + .map(JsonElement::getAsString); + + StepVerifier.create(result) + .expectError(IllegalStateException.class) + .verify(TIMEOUT); + } + + @Test + void subscriber_shouldSubscribeCorrectly() { + Flux 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 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(); + } +} + -- cgit 1.2.3-korg