Sonar Critical Fix
[dcaegen2/analytics/tca.git] / dcae-analytics-cdap-tca / src / test / java / org / openecomp / dcae / apod / analytics / cdap / tca / worker / TCADMaaPMRSubscriberJobTest.java
1 /*\r
2  * ===============================LICENSE_START======================================\r
3  *  dcae-analytics\r
4  * ================================================================================\r
5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
6  * ================================================================================\r
7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
8  *  you may not use this file except in compliance with the License.\r
9  *   You may obtain a copy of the License at\r
10  *\r
11  *          http://www.apache.org/licenses/LICENSE-2.0\r
12  *\r
13  *  Unless required by applicable law or agreed to in writing, software\r
14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
16  *  See the License for the specific language governing permissions and\r
17  *  limitations under the License.\r
18  *  ============================LICENSE_END===========================================\r
19  */\r
20 \r
21 package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
22 \r
23 import co.cask.cdap.api.metrics.Metrics;\r
24 import co.cask.cdap.api.worker.WorkerContext;\r
25 import com.google.common.collect.ImmutableList;\r
26 import org.junit.Before;\r
27 import org.junit.Test;\r
28 import org.mockito.Mockito;\r
29 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;\r
30 import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;\r
31 import org.openecomp.dcae.apod.analytics.cdap.tca.BaseAnalyticsCDAPTCAUnitTest;\r
32 import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;\r
33 import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
34 import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;\r
35 import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;\r
36 import org.quartz.JobDataMap;\r
37 import org.quartz.JobExecutionContext;\r
38 \r
39 import java.io.IOException;\r
40 import java.util.Arrays;\r
41 import java.util.Collections;\r
42 \r
43 import static org.mockito.ArgumentMatchers.anyInt;\r
44 import static org.mockito.ArgumentMatchers.anyString;\r
45 import static org.mockito.ArgumentMatchers.eq;\r
46 import static org.mockito.Mockito.doNothing;\r
47 import static org.mockito.Mockito.doThrow;\r
48 import static org.mockito.Mockito.mock;\r
49 import static org.mockito.Mockito.verify;\r
50 import static org.mockito.Mockito.when;\r
51 \r
52 /**\r
53  * @author Rajiv Singla . Creation Date: 12/20/2016.\r
54  */\r
55 public class TCADMaaPMRSubscriberJobTest extends BaseAnalyticsCDAPTCAUnitTest {\r
56 \r
57     private JobExecutionContext jobExecutionContext;\r
58     private TCADMaaPMRSubscriberJob subscriberJob;\r
59     private JobDataMap jobDataMap;\r
60     private WorkerContext workerContext;\r
61     private DMaaPMRSubscriber subscriber;\r
62     private Metrics metrics;\r
63 \r
64 \r
65     @Before\r
66     public void before() throws Exception {\r
67 \r
68         jobExecutionContext = mock(JobExecutionContext.class);\r
69         workerContext = mock(WorkerContext.class);\r
70 \r
71         metrics = mock(Metrics.class);\r
72         doNothing().when(metrics).count(anyString(), anyInt());\r
73         subscriber = mock(DMaaPMRSubscriber.class);\r
74 \r
75         jobDataMap = mock(JobDataMap.class);\r
76         when(jobDataMap.getString(eq(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME))).thenReturn\r
77                 (CDAPComponentsConstants.TCA_DEFAULT_SUBSCRIBER_OUTPUT_NAME_STREAM);\r
78         when(jobDataMap.get(eq(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME))).thenReturn(workerContext);\r
79         when(jobDataMap.get(eq(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME))).thenReturn(subscriber);\r
80         when(jobDataMap.get(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME)).thenReturn(metrics);\r
81         when(jobExecutionContext.getMergedJobDataMap()).thenReturn(jobDataMap);\r
82 \r
83         doNothing().when(workerContext).write(anyString(), anyString());\r
84 \r
85         subscriberJob = new TCADMaaPMRSubscriberJob();\r
86     }\r
87 \r
88     @Test\r
89     public void testExecuteWhenMessagesAreFound() throws Exception {\r
90         final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);\r
91         when(subscriberResponse.getResponseCode()).thenReturn(200);\r
92         when(subscriberResponse.getResponseMessage()).thenReturn("testMessage");\r
93         when(subscriberResponse.getFetchedMessages()).thenReturn(ImmutableList.of("testMessage1", "testMessage1"));\r
94         when(subscriber.fetchMessages()).thenReturn(subscriberResponse);\r
95         subscriberJob.execute(jobExecutionContext);\r
96         verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants\r
97                 .DMAAP_MR_SUBSCRIBER_TOTAL_MESSAGES_PROCESSED_METRIC), eq(2));\r
98     }\r
99 \r
100     @Test\r
101     public void testExecuteWhenNoMessagesFound() throws Exception {\r
102         final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);\r
103         when(subscriberResponse.getResponseCode()).thenReturn(200);\r
104         when(subscriberResponse.getResponseMessage()).thenReturn("no messages");\r
105         when(subscriberResponse.getFetchedMessages()).thenReturn(Collections.<String>emptyList());\r
106         when(subscriber.fetchMessages()).thenReturn(subscriberResponse);\r
107         subscriberJob.execute(jobExecutionContext);\r
108         verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants\r
109                 .DMAAP_MR_SUBSCRIBER_RESPONSES_WITH_NO_MESSAGES_METRIC), eq(1));\r
110     }\r
111 \r
112 \r
113     @Test\r
114     public void testExecuteWhenSubscriberReturnNonSuccessfulReturnCode() throws Exception {\r
115         final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);\r
116         when(subscriberResponse.getResponseCode()).thenReturn(500);\r
117         when(subscriber.fetchMessages()).thenReturn(subscriberResponse);\r
118         subscriberJob.execute(jobExecutionContext);\r
119         verify(metrics, Mockito.times(1)).count(eq(CDAPMetricsConstants\r
120                 .DMAAP_MR_SUBSCRIBER_UNSUCCESSFUL_RESPONSES_METRIC), eq(1));\r
121     }\r
122 \r
123     @Test(expected = DCAEAnalyticsRuntimeException.class)\r
124     public void testExecuteWhenWritingToCDAPStreamThrowsException() throws Exception {\r
125         final DMaaPMRSubscriberResponse subscriberResponse = mock(DMaaPMRSubscriberResponse.class);\r
126         when(subscriberResponse.getResponseCode()).thenReturn(200);\r
127         when(subscriberResponse.getFetchedMessages()).thenReturn(Arrays.asList("TestMessage"));\r
128         when(subscriber.fetchMessages()).thenReturn(subscriberResponse);\r
129         doThrow(new IOException()).when(workerContext).write(anyString(), anyString());\r
130         subscriberJob.execute(jobExecutionContext);\r
131     }\r
132 \r
133 \r
134 \r
135 }\r