2 * ============LICENSE_START=======================================================
\r
4 * ================================================================================
\r
5 * Copyright (C) 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
11 * http://www.apache.org/licenses/LICENSE-2.0
\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
21 package com.att.nsa.cambria.service.impl;
\r
23 import static org.mockito.Mockito.when;
\r
24 import static org.mockito.Matchers.anyString;
\r
25 import static org.junit.Assert.assertFalse;
\r
26 import static org.junit.Assert.assertTrue;
\r
28 import java.io.ByteArrayInputStream;
\r
29 import java.io.File;
\r
30 import java.io.IOException;
\r
31 import java.io.InputStream;
\r
32 import java.lang.reflect.Constructor;
\r
33 import java.lang.reflect.InvocationTargetException;
\r
34 import java.lang.reflect.Method;
\r
35 import java.util.ConcurrentModificationException;
\r
36 import java.util.Map;
\r
37 import java.util.Properties;
\r
39 import org.junit.After;
\r
40 import org.junit.Before;
\r
41 import org.junit.Test;
\r
42 import org.junit.runner.RunWith;
\r
43 import org.mockito.Mock;
\r
44 import org.mockito.MockitoAnnotations;
\r
45 import org.powermock.api.mockito.PowerMockito;
\r
46 import org.powermock.core.classloader.annotations.PrepareForTest;
\r
47 import org.powermock.modules.junit4.PowerMockRunner;
\r
48 import org.springframework.beans.factory.annotation.Autowired;
\r
49 import org.springframework.mock.web.MockHttpServletRequest;
\r
50 import org.springframework.mock.web.MockHttpServletResponse;
\r
52 import com.att.ajsc.beans.PropertiesMapBean;
\r
53 import com.att.ajsc.filemonitor.AJSCPropertiesMap;
\r
54 import com.att.dmf.mr.CambriaApiException;
\r
55 import com.att.dmf.mr.security.DMaaPAAFAuthenticator;
\r
56 import com.att.dmf.mr.security.DMaaPAuthenticator;
\r
57 import com.att.dmf.mr.security.DMaaPAuthenticatorImpl;
\r
58 import com.att.dmf.mr.utils.ConfigurationReader;
\r
59 import com.att.dmf.mr.backends.ConsumerFactory.UnavailableException;
\r
60 import com.att.dmf.mr.beans.DMaaPCambriaLimiter;
\r
61 import com.att.dmf.mr.backends.ConsumerFactory;
\r
62 import com.att.dmf.mr.beans.DMaaPContext;
\r
63 import com.att.dmf.mr.beans.DMaaPKafkaMetaBroker;
\r
64 import com.att.dmf.mr.constants.CambriaConstants;
\r
65 import com.att.dmf.mr.exception.DMaaPAccessDeniedException;
\r
66 import com.att.dmf.mr.exception.DMaaPErrorMessages;
\r
67 import com.att.dmf.mr.metabroker.Topic;
\r
68 import com.att.dmf.mr.metabroker.Broker.TopicExistsException;
\r
69 import com.att.dmf.mr.service.impl.EventsServiceImpl;
\r
70 import com.att.dmf.mr.utils.PropertyReader;
\r
71 import com.att.nsa.configs.ConfigDbException;
\r
72 import com.att.nsa.drumlin.till.nv.rrNvReadable.invalidSettingValue;
\r
73 import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;
\r
74 import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
\r
75 import com.att.nsa.limits.Blacklist;
\r
76 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
\r
77 import com.att.nsa.security.NsaApiKey;
\r
78 import com.att.nsa.security.db.simple.NsaSimpleApiKey;
\r
80 import kafka.admin.AdminUtils;
\r
82 @RunWith(PowerMockRunner.class)
\r
83 @PrepareForTest({ DMaaPAuthenticatorImpl.class, AJSCPropertiesMap.class })
\r
84 public class EventsServiceImplTest {
\r
86 private InputStream iStream = null;
\r
87 DMaaPContext dMaapContext = new DMaaPContext();
\r
88 EventsServiceImpl service = new EventsServiceImpl();
\r
89 DMaaPErrorMessages pErrorMessages = new DMaaPErrorMessages();
\r
91 ConfigurationReader configurationReader;
\r
93 Blacklist blacklist;
\r
95 DMaaPAuthenticator<NsaSimpleApiKey> dmaaPAuthenticator;
\r
97 DMaaPAAFAuthenticator dmaapAAFauthenticator;
\r
101 NsaSimpleApiKey nsaSimpleApiKey;
\r
103 DMaaPKafkaMetaBroker dmaapKafkaMetaBroker;
\r
105 Topic createdTopic;
\r
107 ConsumerFactory factory;
\r
110 public void setUp() throws Exception {
\r
111 MockitoAnnotations.initMocks(this);
\r
112 String source = "source of my InputStream";
\r
113 iStream = new ByteArrayInputStream(source.getBytes("UTF-8"));
\r
115 MockHttpServletRequest request = new MockHttpServletRequest();
\r
116 MockHttpServletResponse response = new MockHttpServletResponse();
\r
117 dMaapContext.setRequest(request);
\r
118 dMaapContext.setResponse(response);
\r
119 when(blacklist.contains(anyString())).thenReturn(false);
\r
120 when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
\r
121 dMaapContext.setConfigReader(configurationReader);
\r
123 service.setErrorMessages(pErrorMessages);
\r
124 PowerMockito.mockStatic(AJSCPropertiesMap.class);
\r
125 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "timeout")).thenReturn("100");
\r
127 AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "timeout");
\r
131 @Test(expected = CambriaApiException.class)
\r
132 public void testGetEvents() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
133 TopicExistsException, AccessDeniedException, UnavailableException, IOException {
\r
134 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
135 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
136 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
137 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
138 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
139 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
142 @Test(expected = CambriaApiException.class)
\r
143 public void testGetEventsBlackListErr() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
144 TopicExistsException, AccessDeniedException, UnavailableException, IOException {
\r
145 when(blacklist.contains(anyString())).thenReturn(true);
\r
146 when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
\r
147 dMaapContext.setConfigReader(configurationReader);
\r
148 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
151 @Test(expected = CambriaApiException.class)
\r
152 public void testGetEventsNoTopicError() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
153 TopicExistsException, AccessDeniedException, UnavailableException, IOException {
\r
154 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
155 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
156 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
157 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(null);
\r
158 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
159 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
162 @Test(expected = CambriaApiException.class)
\r
163 public void testGetEventsuserNull() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
164 TopicExistsException, AccessDeniedException, UnavailableException, IOException {
\r
165 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
\r
166 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
167 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
168 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
169 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
170 MockHttpServletRequest mockRequest = new MockHttpServletRequest();
\r
171 mockRequest.addHeader("Authorization", "passed");
\r
172 dMaapContext.setRequest(mockRequest);
\r
173 dMaapContext.getRequest().getHeader("Authorization");
\r
174 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
177 @Test(expected = CambriaApiException.class)
\r
178 public void testGetEventsExcp2() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
179 TopicExistsException, AccessDeniedException, UnavailableException, IOException {
\r
180 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
181 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
182 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
183 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
184 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
185 when(configurationReader.getfRateLimiter()).thenThrow(new ConcurrentModificationException("Error occurred"));
\r
186 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
189 @Test(expected = CambriaApiException.class)
\r
190 public void testPushEvents() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
191 TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
\r
192 invalidSettingValue, loadException {
\r
194 // AdminUtils.createTopic(configurationReader.getZk(), "testTopic", 10,
\r
195 // 1, new Properties());
\r
197 configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
\r
199 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
200 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
201 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
202 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
203 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
205 service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
\r
207 service.getEvents(dMaapContext, "testTopic", "CG1", "23");
\r
210 * String trueValue = "True";
\r
211 * assertTrue(trueValue.equalsIgnoreCase("True"));
\r
216 @Test(expected = CambriaApiException.class)
\r
217 public void testPushEventsBlackListedIp() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
218 TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
\r
219 invalidSettingValue, loadException {
\r
221 // AdminUtils.createTopic(configurationReader.getZk(), "testTopic", 10,
\r
222 // 1, new Properties());
\r
223 when(blacklist.contains(anyString())).thenReturn(true);
\r
224 when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
\r
225 configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
\r
226 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
227 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
228 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
229 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
230 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
232 service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
\r
236 @Test(expected = NullPointerException.class)
\r
237 public void testPushEventsNoUser() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
238 TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
\r
239 invalidSettingValue, loadException {
\r
241 configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
\r
243 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
\r
244 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
245 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
246 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
247 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
248 MockHttpServletRequest mockRequest = new MockHttpServletRequest();
\r
249 mockRequest.addHeader("Authorization", "passed");
\r
250 mockRequest.addHeader("Authorization", "passed");
\r
251 dMaapContext.setRequest(mockRequest);
\r
252 dMaapContext.getRequest().getHeader("Authorization");
\r
253 service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
\r
257 @Test(expected = CambriaApiException.class)
\r
258 public void testPushEventsWtTransaction() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
259 TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
\r
260 invalidSettingValue, loadException {
\r
262 configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
\r
264 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
265 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
266 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
267 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
268 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
269 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "transidUEBtopicreqd")).thenReturn("true");
\r
271 service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
\r
275 @Test(expected = CambriaApiException.class)
\r
276 public void testPushEventsWtTransactionError() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
\r
277 TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
\r
278 invalidSettingValue, loadException {
\r
280 configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
\r
282 when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
\r
283 when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
\r
284 when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
\r
285 when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
\r
286 PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
\r
287 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "transidUEBtopicreqd")).thenReturn("true");
\r
288 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "event.batch.length")).thenReturn("0");
\r
289 when(configurationReader.getfPublisher()).thenThrow(new ConcurrentModificationException("Error occurred"));
\r
291 service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
\r
296 public void testIsTransEnabled1() {
\r
298 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
\r
299 "transidUEBtopicreqd")).thenReturn("true");
\r
300 assertTrue(service.isTransEnabled());
\r
304 public void testIsTransEnabled2() {
\r
306 when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
\r
307 "transidUEBtopicreqd")).thenReturn("false");
\r
308 assertFalse(service.isTransEnabled());
\r