dcae-analytics-dmaap/src/test/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImplTest.java

   1 /*
   2  * ===============================LICENSE_START======================================
   3  *  dcae-analytics
   4  * ================================================================================
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *   You may obtain a copy of the License at
  10  *
  11  *          http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *  Unless required by applicable law or agreed to in writing, software
  14  *  distributed under the License is distributed on an "AS IS" BASIS,
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *  See the License for the specific language governing permissions and
  17  *  limitations under the License.
  18  *  ============================LICENSE_END===========================================
  19  */
  20
  21 package org.onap.dcae.apod.analytics.dmaap.service.publisher;
  22
  23 import org.apache.commons.lang3.tuple.ImmutablePair;
  24 import org.apache.http.client.ResponseHandler;
  25 import org.apache.http.client.methods.HttpUriRequest;
  26 import org.apache.http.impl.client.CloseableHttpClient;
  27 import org.junit.After;
  28 import org.junit.Before;
  29 import org.junit.Rule;
  30 import org.junit.Test;
  31 import org.junit.rules.ExpectedException;
  32 import org.junit.runner.RunWith;
  33 import org.mockito.Mock;
  34 import org.mockito.Mockito;
  35 import org.mockito.junit.MockitoJUnitRunner;
  36 import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
  37 import org.onap.dcae.apod.analytics.dmaap.BaseAnalyticsDMaaPUnitTest;
  38 import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
  39 import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
  40
  41 import java.io.IOException;
  42 import java.util.ArrayList;
  43
  44 import static org.hamcrest.CoreMatchers.isA;
  45 import static org.hamcrest.core.Is.is;
  46 import static org.junit.Assert.assertThat;
  47 import static org.mockito.BDDMockito.given;
  48 import static org.mockito.Mockito.times;
  49 import static org.mockito.Mockito.verify;
  50
  51 /**
  52  * @author Rajiv Singla . Creation Date: 10/21/2016.
  53  */
  54 @RunWith(MockitoJUnitRunner.class)
  55 public class DMaaPMRPublisherImplTest extends BaseAnalyticsDMaaPUnitTest {
  56
  57     @Mock
  58     private DMaaPMRPublisherQueueFactory dmaapMRPublisherQueueFactory;
  59     @Mock
  60     private CloseableHttpClient closeableHttpClient;
  61     @Mock
  62     private DMaaPMRPublisherQueue dmaapMRPublisherQueue;
  63
  64     @Before
  65     public void setUp() throws Exception {
  66         given(dmaapMRPublisherQueueFactory.create(Mockito.anyInt(), Mockito.anyInt()))
  67                 .willReturn(dmaapMRPublisherQueue);
  68     }
  69
  70     @After
  71     public void tearDown() throws Exception {
  72     }
  73
  74     @Test
  75     public void testPublishSmallMessageList() throws Exception {
  76         given(dmaapMRPublisherQueue.getBatchQueueRemainingSize()).willReturn(10);
  77         given(dmaapMRPublisherQueue.addBatchMessages(Mockito.<String>anyList())).willReturn(2);
  78
  79         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
  80                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
  81
  82         DMaaPMRPublisherResponse dmaapMRPublisherResponse = dmaapMRPublisherImpl.publish(getTwoSampleMessages());
  83
  84         assertThat(dmaapMRPublisherResponse.getResponseCode(), is(202));
  85         assertThat(dmaapMRPublisherResponse.getPendingMessagesCount(), is(2));
  86         assertThat(dmaapMRPublisherResponse.getResponseMessage(),
  87                 is("Accepted - Messages queued for batch publishing to MR Topic"));
  88     }
  89
  90     @Test
  91     public void testPublishBigMessageList() throws Exception {
  92
  93         given(dmaapMRPublisherQueue.getBatchQueueRemainingSize()).willReturn(0);
  94         given(dmaapMRPublisherQueue.getMessageForPublishing()).willReturn(getTwoSampleMessages());
  95         Mockito.when(
  96                 closeableHttpClient.execute(Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
  97                 .thenReturn(new ImmutablePair<>(200, "Message successfully posted"));
  98
  99         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 100                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 101
 102         DMaaPMRPublisherResponse dmaapMRPublisherResponse = dmaapMRPublisherImpl.publish(getTwoSampleMessages());
 103
 104         assertThat(dmaapMRPublisherResponse.getResponseCode(), is(200));
 105         assertThat(dmaapMRPublisherResponse.getPendingMessagesCount(), is(200));
 106         assertThat(dmaapMRPublisherResponse.getResponseMessage(), is("Message successfully posted"));
 107     }
 108
 109     @Test
 110     public void testForcePublishSuccessful() throws Exception {
 111         DMaaPMRPublisherConfig dmaapMRPublisherConfig = new
 112                 DMaaPMRPublisherConfig.Builder(HOST_NAME, TOPIC_NAME)
 113                 .setPortNumber(PORT_NUMBER)
 114                 .setProtocol(HTTP_PROTOCOL)
 115                 .setContentType(CONTENT_TYPE)
 116                 .setMaxRecoveryQueueSize(PUBLISHER_MAX_RECOVERY_QUEUE_SIZE)
 117                 .setMaxBatchSize(PUBLISHER_MAX_BATCH_QUEUE_SIZE).build();
 118
 119         Mockito.when(closeableHttpClient.execute(
 120                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
 121                 .thenReturn(new ImmutablePair<>(200, "Message successfully posted"));
 122
 123         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 124                 dmaapMRPublisherConfig, dmaapMRPublisherQueueFactory, closeableHttpClient);
 125         DMaaPMRPublisherResponse response = dmaapMRPublisherImpl.forcePublish(getTwoSampleMessages());
 126         assertThat(response.getResponseCode(), is(200));
 127     }
 128
 129     @Test
 130     public void testForcePublishFailure() throws Exception {
 131         Mockito.when(closeableHttpClient.execute(
 132                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
 133                 .thenReturn(new ImmutablePair<>(503, "Message successfully posted"));
 134
 135         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 136                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 137         DMaaPMRPublisherResponse response = dmaapMRPublisherImpl.forcePublish(getTwoSampleMessages());
 138         assertThat(response.getResponseCode(), is(503));
 139     }
 140
 141     @Rule
 142     public ExpectedException httpIOException = ExpectedException.none();
 143
 144     @Test
 145     public void testForcePublishHttpFailure() throws Exception {
 146
 147         httpIOException.expect(DCAEAnalyticsRuntimeException.class);
 148         httpIOException.expectCause(isA(IOException.class));
 149
 150         given(closeableHttpClient.execute(
 151                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class))).willThrow(IOException.class);
 152
 153         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 154                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 155         dmaapMRPublisherImpl.forcePublish(getTwoSampleMessages());
 156     }
 157
 158     @Test
 159     public void testFlushSuccessful() throws Exception {
 160         Mockito.when(closeableHttpClient.execute(
 161                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
 162                 .thenReturn(new ImmutablePair<>(200, "Message successfully posted"));
 163
 164         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(getTwoSampleMessages());
 165
 166         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 167                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 168         DMaaPMRPublisherResponse response = dmaapMRPublisherImpl.flush();
 169         assertThat(response.getResponseCode(), is(200));
 170     }
 171
 172     @Test
 173     public void testFlushEmptyList() throws Exception {
 174         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(new ArrayList<String>());
 175
 176         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 177                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 178         DMaaPMRPublisherResponse response = dmaapMRPublisherImpl.flush();
 179         assertThat(response.getResponseCode(), is(204));
 180     }
 181
 182     @Test
 183     public void testClose() throws Exception {
 184         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(new ArrayList<String>());
 185         Mockito.when(closeableHttpClient.execute(
 186                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
 187                 .thenReturn(new ImmutablePair<>(200, "Message successfully posted"));
 188         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(getTwoSampleMessages());
 189
 190         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 191                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 192         dmaapMRPublisherImpl.close();
 193         verify(closeableHttpClient).execute(Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class));
 194     }
 195
 196     @Test
 197     public void testCloseUnsuccessful() throws Exception {
 198         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(new ArrayList<String>());
 199         Mockito.when(closeableHttpClient.execute(
 200                 Mockito.any(HttpUriRequest.class), Mockito.any(ResponseHandler.class)))
 201                 .thenReturn(new ImmutablePair<>(400, "Message successfully posted"));
 202         Mockito.when(dmaapMRPublisherQueue.getMessageForPublishing()).thenReturn(getTwoSampleMessages());
 203
 204         DMaaPMRPublisherImpl dmaapMRPublisherImpl = new DMaaPMRPublisherImpl(
 205                 getPublisherConfig(), dmaapMRPublisherQueueFactory, closeableHttpClient);
 206         dmaapMRPublisherImpl.close();
 207         verify(closeableHttpClient, times(6)).execute(Mockito.any(HttpUriRequest.class),
 208                 Mockito.any(ResponseHandler.class));
 209     }
 210 }