Initial TCA commit into DCAEGEN2
[dcaegen2/analytics/tca.git] / dcae-analytics-cdap-plugins / src / test / java / org / openecomp / dcae / apod / analytics / cdap / plugins / streaming / dmaap / DMaaPMRReceiverTest.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.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
22
23 import com.google.common.collect.ImmutableList;
24 import org.apache.spark.storage.StorageLevel;
25 import org.junit.Test;
26 import org.mockito.Mockito;
27 import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;
28 import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
29 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
30 import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
31 import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
32
33 import static org.mockito.Mockito.times;
34 import static org.mockito.Mockito.verify;
35 import static org.mockito.Mockito.when;
36
37 /**
38  * @author Rajiv Singla . Creation Date: 1/24/2017.
39  */
40 public class DMaaPMRReceiverTest extends BaseAnalyticsCDAPPluginsUnitTest {
41
42
43     @Test
44     public void testStoreStructuredRecords() throws Exception {
45
46         final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
47         final TestDMaaPMRReceiver dMaaPMRReceiver =
48                 new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig);
49
50         final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class);
51         final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class);
52         when(dMaaPMRSubscriber.fetchMessages()).thenReturn(subscriberResponse);
53         when(subscriberResponse.getFetchedMessages()).thenReturn(ImmutableList.of("Test Message"));
54         when(subscriberResponse.getResponseCode()).thenReturn(200);
55         when(subscriberResponse.getResponseMessage()).thenReturn("OK");
56         dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber);
57         verify(dMaaPMRSubscriber, times(1)).fetchMessages();
58         verify(subscriberResponse, times(1)).getFetchedMessages();
59     }
60
61     @Test
62     public void testStoreStructuredRecordsWhenSubscriberThrowsException() throws Exception {
63
64         final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
65         final TestDMaaPMRReceiver dMaaPMRReceiver =
66                 new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig);
67
68         final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class);
69         final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class);
70         when(dMaaPMRSubscriber.fetchMessages()).thenThrow(DCAEAnalyticsRuntimeException.class);
71         dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber);
72         verify(dMaaPMRSubscriber, times(1)).fetchMessages();
73         verify(subscriberResponse, times(0)).getFetchedMessages();
74     }
75 }