src/test/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImplTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP Policy Engine
   4  * ================================================================================
   5  * Copyright (C) 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.dmaap.dmf.mr.service.impl;
  22
  23 import static org.hamcrest.CoreMatchers.containsString;
  24 import static org.junit.Assert.assertEquals;
  25 import static org.junit.Assert.assertFalse;
  26 import static org.junit.Assert.assertTrue;
  27 import static org.mockito.Matchers.any;
  28 import static org.mockito.Matchers.anyInt;
  29 import static org.mockito.Matchers.anyLong;
  30 import static org.mockito.Matchers.anyString;
  31 import static org.mockito.Matchers.eq;
  32 import static org.mockito.Mockito.doNothing;
  33 import static org.mockito.Mockito.doReturn;
  34 import static org.mockito.Mockito.doThrow;
  35 import static org.mockito.Mockito.mock;
  36 import static org.mockito.Mockito.never;
  37 import static org.mockito.Mockito.verify;
  38 import static org.mockito.Mockito.when;
  39
  40 import com.att.nsa.limits.Blacklist;
  41 import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
  42 import com.att.nsa.security.db.simple.NsaSimpleApiKey;
  43 import java.io.ByteArrayInputStream;
  44 import java.io.IOException;
  45 import java.io.InputStream;
  46 import java.util.ArrayList;
  47 import java.util.Collections;
  48 import java.util.ConcurrentModificationException;
  49 import javax.servlet.http.HttpServletRequest;
  50 import joptsimple.internal.Strings;
  51 import org.apache.http.HttpStatus;
  52 import org.apache.kafka.clients.producer.ProducerRecord;
  53 import org.json.JSONObject;
  54 import org.junit.Before;
  55 import org.junit.Rule;
  56 import org.junit.Test;
  57 import org.junit.rules.ExpectedException;
  58 import org.junit.runner.RunWith;
  59 import org.mockito.ArgumentCaptor;
  60 import org.mockito.InOrder;
  61 import org.mockito.Mock;
  62 import org.mockito.Mockito;
  63 import org.mockito.MockitoAnnotations;
  64 import org.mockito.Spy;
  65 import org.mockito.runners.MockitoJUnitRunner;
  66 import org.onap.dmaap.dmf.mr.CambriaApiException;
  67 import org.onap.dmaap.dmf.mr.backends.Consumer;
  68 import org.onap.dmaap.dmf.mr.backends.ConsumerFactory;
  69 import org.onap.dmaap.dmf.mr.backends.ConsumerFactory.UnavailableException;
  70 import org.onap.dmaap.dmf.mr.backends.MetricsSet;
  71 import org.onap.dmaap.dmf.mr.backends.Publisher;
  72 import org.onap.dmaap.dmf.mr.beans.DMaaPCambriaLimiter;
  73 import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
  74 import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaMetaBroker;
  75 import org.onap.dmaap.dmf.mr.exception.DMaaPAccessDeniedException;
  76 import org.onap.dmaap.dmf.mr.exception.DMaaPErrorMessages;
  77 import org.onap.dmaap.dmf.mr.metabroker.Topic;
  78 import org.onap.dmaap.dmf.mr.resources.CambriaOutboundEventStream;
  79 import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticator;
  80 import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
  81 import org.springframework.mock.web.MockHttpServletRequest;
  82 import org.springframework.mock.web.MockHttpServletResponse;
  83
  84 @RunWith(MockitoJUnitRunner.class)
  85 public class EventsServiceImplTest {
  86
  87     private InputStream iStream = null;
  88     private DMaaPContext dMaapContext = new DMaaPContext();
  89     private DMaaPErrorMessages pErrorMessages = new DMaaPErrorMessages();
  90     @Mock
  91     private ConfigurationReader configurationReader;
  92     @Mock
  93     private Blacklist blacklist;
  94     @Mock
  95     private DMaaPAuthenticator<NsaSimpleApiKey> dmaaPAuthenticator;
  96     @Mock
  97     private NsaSimpleApiKey nsaSimpleApiKey;
  98     @Mock
  99     private DMaaPKafkaMetaBroker dmaapKafkaMetaBroker;
 100     @Mock
 101     private Topic createdTopic;
 102     @Mock
 103     private ConsumerFactory factory;
 104     @Mock
 105     private Consumer consumer;
 106     @Mock
 107     private Publisher publisher;
 108     @Mock
 109     private DMaaPCambriaLimiter limiter;
 110     @Mock
 111     private MetricsSet metrics;
 112     @Spy
 113     private EventsServiceImpl eventsService;
 114
 115
 116     @Rule
 117     public ExpectedException thrown = ExpectedException.none();
 118
 119     private MockHttpServletRequest request;
 120
 121     @Before
 122     public void setUp() throws Exception {
 123         MockitoAnnotations.initMocks(this);
 124         String source = "source of my InputStream";
 125         iStream = new ByteArrayInputStream(source.getBytes("UTF-8"));
 126
 127         request = new MockHttpServletRequest();
 128         MockHttpServletResponse response = new MockHttpServletResponse();
 129         dMaapContext.setRequest(request);
 130         dMaapContext.setResponse(response);
 131         when(blacklist.contains(anyString())).thenReturn(false);
 132         when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
 133         when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
 134         dMaapContext.setConfigReader(configurationReader);
 135         eventsService.setErrorMessages(pErrorMessages);
 136         doReturn("100").when(eventsService).getPropertyFromAJSCmap("timeout");
 137     }
 138
 139     @Test
 140     public void getEvents_shouldFailOnAafAuthorization() throws Exception {
 141         String topicPrefix = "org.onap.aaf.enforced";
 142         String topicName = topicPrefix + ".topicName";
 143         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 144         when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
 145         when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
 146         when(eventsService.isCadiEnabled()).thenReturn(true);
 147
 148         thrown.expect(DMaaPAccessDeniedException.class);
 149         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_UNAUTHORIZED)));
 150
 151         eventsService.getEvents(dMaapContext, topicName, "CG1", "23");
 152     }
 153
 154     @Test
 155     public void getEvents_shouldFail_whenRemoteAddressIsBlacklisted() throws Exception {
 156         String remoteIp = "10.154.17.115";
 157         request.setRemoteAddr(remoteIp);
 158         when(blacklist.contains(remoteIp)).thenReturn(true);
 159         when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
 160
 161         thrown.expect(CambriaApiException.class);
 162         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_FORBIDDEN)));
 163
 164         eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
 165     }
 166
 167     @Test
 168     public void getEvents_shouldFail_whenRequestedTopicNotExists() throws Exception {
 169         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 170         when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(null);
 171
 172         thrown.expect(CambriaApiException.class);
 173         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
 174
 175         eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
 176     }
 177
 178     @Test
 179     public void getEvents_shouldFail_whenConsumerLockCannotBeAcquired() throws Exception {
 180         //given
 181         String topicName = "testTopic345";
 182         String consumerGroup = "CG5";
 183         String clientId = "13";
 184         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 185         when(configurationReader.getfRateLimiter()).thenReturn(limiter);
 186         when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
 187         when(configurationReader.getfConsumerFactory()).thenReturn(factory);
 188         when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
 189         doThrow(new UnavailableException("Could not acquire consumer lock")).when(factory)
 190             .getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
 191
 192         thrown.expect(CambriaApiException.class);
 193         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_SERVICE_UNAVAILABLE)));
 194
 195         //when
 196         eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
 197
 198         //then
 199         verify(factory).getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
 200
 201     }
 202
 203     @Test
 204     public void getEvents_shouldFail_whenBrokerServicesAreUnavailable() throws Exception {
 205         String topicName = "testTopic";
 206         String consumerGroup = "CG1";
 207         String clientId = "23";
 208         when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
 209         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 210         when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
 211         when(configurationReader.getfConsumerFactory()).thenReturn(factory);
 212
 213         givenUserAuthorizedWithAAF(request, topicName, "sub");
 214
 215         thrown.expect(CambriaApiException.class);
 216         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_SERVICE_UNAVAILABLE)));
 217
 218         //when
 219         eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
 220
 221         //then
 222         verify(factory).destroyConsumer(topicName, consumerGroup, clientId);
 223     }
 224
 225     private void givenUserAuthorizedWithAAF(MockHttpServletRequest request, String topicName, String operation) {
 226         String permission = "org.onap.dmaap.mr.topic|:topic." + topicName + "|" + operation;
 227         request.addUserRole(permission);
 228     }
 229
 230     @Test
 231     public void getEvents_shouldHandleConcurrentModificationError() throws Exception {
 232         String testTopic = "testTopic";
 233         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 234         when(dmaapKafkaMetaBroker.getTopic(testTopic)).thenReturn(createdTopic);
 235         when(configurationReader.getfConsumerFactory()).thenReturn(factory);
 236         when(configurationReader.getfRateLimiter()).thenThrow(new ConcurrentModificationException("Error occurred"));
 237         givenUserAuthorizedWithAAF(request, testTopic, "sub");
 238
 239         thrown.expect(CambriaApiException.class);
 240         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_CONFLICT)));
 241
 242         eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
 243     }
 244
 245     @Test
 246     public void getEvents_shouldNotAuthorizeClient_whenSubscribingToMetricsTopic() throws Exception {
 247         //given
 248         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 249         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 250         dMaapContext.setRequest(permittedRequest);
 251         String metricsTopicName = "msgrtr.apinode.metrics.dmaap";
 252         String consumerGroup = "CG5";
 253         String clientId = "7";
 254         givenConfiguredWithMocks(metricsTopicName);
 255         when(factory.getConsumerFor(eq(metricsTopicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
 256             .thenReturn(consumer);
 257         doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 258
 259         //when
 260         eventsService.getEvents(dMaapContext, metricsTopicName, consumerGroup, clientId);
 261
 262         //then
 263         verify(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 264         verify(dmaaPAuthenticator, never()).authenticate(dMaapContext);
 265         verify(permittedRequest, never()).isUserInRole(anyString());
 266     }
 267
 268     @Test
 269     public void getEvents_shouldNotAuthorizeClient_whenTopicNoteEnforcedWithAaf_andTopicHasNoOwnerSet()
 270         throws Exception {
 271         //given
 272         String topicName = "someSimpleTopicName";
 273         String consumerGroup = "CG5";
 274         String clientId = "7";
 275         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 276         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 277         dMaapContext.setRequest(permittedRequest);
 278         givenConfiguredWithMocks(topicName);
 279         when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
 280             .thenReturn(consumer);
 281         doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 282         when(createdTopic.getOwner()).thenReturn(Strings.EMPTY);
 283
 284         //when
 285         eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
 286
 287         //then
 288         verify(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 289         verify(dmaaPAuthenticator, never()).authenticate(dMaapContext);
 290         verify(permittedRequest, never()).isUserInRole(anyString());
 291     }
 292
 293     @Test
 294     public void getEvents_shouldFailDmaapAuthorization_whenTopicOwnerIsSet_andUserHasNoReadPermissionToTopic()
 295         throws Exception {
 296         //given
 297         String topicName = "someSimpleTopicName";
 298         String consumerGroup = "CG5";
 299         String clientId = "7";
 300         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 301         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 302         dMaapContext.setRequest(permittedRequest);
 303         givenConfiguredWithMocks(topicName);
 304         when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
 305             .thenReturn(consumer);
 306         doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 307         when(createdTopic.getOwner()).thenReturn("SimpleTopicOwner");
 308         when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
 309         doThrow(new AccessDeniedException("userName")).when(createdTopic).checkUserRead(nsaSimpleApiKey);
 310
 311         thrown.expect(AccessDeniedException.class);
 312
 313         //when
 314         eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
 315
 316         //then
 317         verify(createdTopic).checkUserRead(nsaSimpleApiKey);
 318         verify(eventsService, never()).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 319         verify(permittedRequest, never()).isUserInRole(anyString());
 320     }
 321
 322
 323     @Test
 324     public void getEvents_shouldSuccessfullyRegisterConsumerToEventsStream_withAafAuthorization() throws Exception {
 325         //given
 326         String topicName = "testTopic";
 327         String consumerGroup = "CG2";
 328         String clientId = "6";
 329         String messageLimit = "10";
 330         String timeout = "25";
 331         String meta = "yes";
 332         String pretty = "on";
 333         String cacheEnabled = "false";
 334
 335         givenConfiguredWithMocks(topicName);
 336         givenConfiguredWithProperties(messageLimit, timeout, meta, pretty, cacheEnabled);
 337         when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
 338             .thenReturn(consumer);
 339         givenUserAuthorizedWithAAF(request, topicName, "sub");
 340
 341         //when
 342         eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
 343
 344         //then
 345         ArgumentCaptor<CambriaOutboundEventStream> osWriter = ArgumentCaptor.forClass(CambriaOutboundEventStream.class);
 346         verifyInvocationOrderForSuccessCase(topicName, consumerGroup, clientId, osWriter);
 347         assertEventStreamProperties(osWriter.getValue(), messageLimit, timeout);
 348     }
 349
 350     private void assertEventStreamProperties(CambriaOutboundEventStream stream, String messageLimit, String timeout) {
 351         assertEquals(Integer.valueOf(messageLimit).intValue(), stream.getfLimit());
 352         assertEquals(Integer.valueOf(timeout).intValue(), stream.getfTimeoutMs());
 353         assertTrue(stream.isfWithMeta());
 354         assertTrue(stream.isfPretty());
 355     }
 356
 357     private void givenConfiguredWithProperties(String messageLimit, String timeout, String meta, String pretty,
 358         String cacheEnabled) {
 359         when(eventsService.getPropertyFromAJSCmap("meta")).thenReturn(meta);
 360         when(eventsService.getPropertyFromAJSCmap("pretty")).thenReturn(pretty);
 361         when(eventsService.getPropertyFromAJSCmap(ConsumerFactory.kSetting_EnableCache)).thenReturn(cacheEnabled);
 362         request.addParameter("timeout", timeout);
 363         request.addParameter("limit", messageLimit);
 364     }
 365
 366     private void givenConfiguredWithMocks(String topicName) throws Exception {
 367         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 368         when(configurationReader.getfRateLimiter()).thenReturn(limiter);
 369         when(configurationReader.getfMetrics()).thenReturn(metrics);
 370         when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
 371         when(configurationReader.getfConsumerFactory()).thenReturn(factory);
 372         when(configurationReader.getfPublisher()).thenReturn(publisher);
 373     }
 374
 375     private void verifyInvocationOrderForSuccessCase(String topicName, String consumerGroup, String clientId,
 376         ArgumentCaptor<CambriaOutboundEventStream> osWriter) throws Exception {
 377
 378         InOrder inOrder = Mockito.inOrder(configurationReader, factory, metrics, limiter, consumer, eventsService);
 379         inOrder.verify(configurationReader).getfMetrics();
 380         inOrder.verify(configurationReader).getfRateLimiter();
 381         inOrder.verify(limiter).onCall(eq(topicName), eq(consumerGroup), eq(clientId), anyString());
 382         inOrder.verify(factory).getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
 383         inOrder.verify(eventsService).respondOkWithStream(eq(dMaapContext), osWriter.capture());
 384         inOrder.verify(consumer).commitOffsets();
 385         inOrder.verify(metrics).consumeTick(anyInt());
 386         inOrder.verify(limiter).onSend(eq(topicName), eq(consumerGroup), eq(clientId), anyLong());
 387         inOrder.verify(consumer).close();
 388         inOrder.verifyNoMoreInteractions();
 389     }
 390
 391     @Test
 392     public void pushEvents_shouldFail_whenRemoteAddressIsBlacklisted() throws Exception {
 393         String remoteIp = "10.132.64.112";
 394         request.setRemoteAddr(remoteIp);
 395         when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
 396         when(blacklist.contains(anyString())).thenReturn(true);
 397
 398         thrown.expect(CambriaApiException.class);
 399         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_FORBIDDEN)));
 400
 401         eventsService.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
 402     }
 403
 404
 405
 406
 407     @Test
 408     public void pushEvents_shouldFailDmaapAuthorization_whenTopicOwnerIsSet_andUserHasNoWritePermissionToTopic()
 409         throws Exception {
 410         //given
 411         String topicName = "someSimpleTopicName";
 412
 413         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 414         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 415         dMaapContext.setRequest(permittedRequest);
 416         givenConfiguredWithMocks(topicName);
 417         when(createdTopic.getOwner()).thenReturn("SimpleTopicOwner");
 418         when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
 419         doThrow(new AccessDeniedException("userName")).when(createdTopic).checkUserWrite(nsaSimpleApiKey);
 420
 421         thrown.expect(AccessDeniedException.class);
 422
 423         //when
 424         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 425
 426         //then
 427         verify(createdTopic).checkUserWrite(nsaSimpleApiKey);
 428         verify(eventsService, never()).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
 429         verify(permittedRequest, never()).isUserInRole(anyString());
 430     }
 431
 432     @Test
 433     public void pushEvents_shouldFailOnAafAuthorization_whenCadiIsEnabled_topicNameEnforced_andUserHasNoPermission()
 434         throws Exception {
 435         //given
 436         String topicPrefix = "org.onap.aaf.enforced";
 437         String topicName = topicPrefix + ".topicName";
 438         String permission = "org.onap.dmaap.mr.topic|:topic." + topicName + "|pub";
 439         HttpServletRequest deniedRequest = mock(HttpServletRequest.class);
 440         when(deniedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 441         when(deniedRequest.isUserInRole(permission)).thenReturn(false);
 442         when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
 443         when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
 444         when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
 445         when(eventsService.isCadiEnabled()).thenReturn(true);
 446         dMaapContext.setRequest(deniedRequest);
 447
 448         thrown.expect(DMaaPAccessDeniedException.class);
 449         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_UNAUTHORIZED)));
 450
 451         //when
 452         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 453
 454         //then
 455         verify(deniedRequest).isUserInRole(permission);
 456     }
 457
 458
 459     @Test
 460     public void pushEvents_shouldPublishMessagesWithoutTransaction() throws Exception {
 461         //given
 462         String topicName = "topicWithoutTransaction";
 463         givenConfiguredWithMocks(topicName);
 464         doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
 465
 466         //when
 467         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 468
 469         //then
 470         verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 471         ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
 472         verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
 473         assertEquals(1, captor.getValue().getLong("count"));
 474     }
 475
 476     @Test
 477     public void pushEvents_shouldHandlePublisherError_whenPushWithoutTransaction() throws Exception {
 478         //given
 479         String topicName = "topicWithoutTransaction";
 480         givenConfiguredWithMocks(topicName);
 481         doThrow(new IOException()).when(publisher)
 482             .sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 483
 484         thrown.expect(CambriaApiException.class);
 485         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
 486
 487         //when
 488         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 489
 490         //then
 491         verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 492         verify(eventsService, never()).respondOk(any(DMaaPContext.class), any(JSONObject.class));
 493     }
 494
 495
 496     @Test
 497     public void pushEvents_shouldPublishMessagesWithTransaction() throws Exception {
 498         //given
 499         String topicPrefix = "org.onap.dmaap.mr";
 500         String topicName = topicPrefix + ".topicWithTransaction";
 501         givenConfiguredWithMocks(topicName);
 502         when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
 503         when(eventsService.isCadiEnabled()).thenReturn(true);
 504         doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
 505
 506         request.addUserRole("org.onap.dmaap.mr.topic|:topic." + topicName + "|pub");
 507
 508         //when
 509         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 510
 511         //then
 512         verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 513         ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
 514         verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
 515         assertEquals(1, captor.getValue().getLong("count"));
 516         assertFalse(captor.getValue().getString("transactionId").isEmpty());
 517     }
 518
 519     @Test
 520     public void pushEvents_shouldHandlePublisherError_whenPushWithTransaction() throws Exception {
 521         //given
 522         String topicPrefix = "org.onap.dmaap.mr";
 523         String topicName = topicPrefix + ".topicWithTransaction";
 524         givenConfiguredWithMocks(topicName);
 525         when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
 526         when(eventsService.isCadiEnabled()).thenReturn(true);
 527         doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
 528         request.addUserRole("org.onap.dmaap.mr.topic|:topic." + topicName + "|pub");
 529         doThrow(new IOException()).when(publisher)
 530             .sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 531
 532         thrown.expect(CambriaApiException.class);
 533         thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
 534
 535         //when
 536         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 537
 538         //then
 539         verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 540         verify(eventsService, never()).respondOk(any(DMaaPContext.class), any(JSONObject.class));
 541     }
 542
 543     @Test
 544     public void pushEvents_shouldNotPerformAnyAuthorization_whenPublishToMetricTopic() throws Exception {
 545         //given
 546         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 547         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 548         dMaapContext.setRequest(permittedRequest);
 549         String metricsTopicName = "msgrtr.apinode.metrics.dmaap";
 550         givenConfiguredWithMocks(metricsTopicName);
 551         doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
 552
 553         //when
 554         eventsService.pushEvents(dMaapContext, metricsTopicName, iStream, "5", "13:00:00");
 555
 556         //then
 557         ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
 558         verify(publisher)
 559             .sendBatchMessageNew(eq(metricsTopicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 560         verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
 561         verify(permittedRequest, never()).isUserInRole(anyString());
 562         verify(createdTopic, never()).checkUserWrite(any(NsaSimpleApiKey.class));
 563         assertEquals(1, captor.getValue().getLong("count"));
 564     }
 565
 566     @Test
 567     public void pushEvents_shouldNotPerformAnyAuthorization_whenTopicHasNoOwner() throws Exception {
 568         //given
 569         HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
 570         when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
 571         dMaapContext.setRequest(permittedRequest);
 572         String topicName = "notEnforcedAafTopic";
 573         givenConfiguredWithMocks(topicName);
 574         doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
 575         when(createdTopic.getOwner()).thenReturn(null);
 576
 577         //when
 578         eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
 579
 580         //then
 581         ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
 582         verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
 583         verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
 584         verify(permittedRequest, never()).isUserInRole(anyString());
 585         verify(createdTopic, never()).checkUserWrite(any(NsaSimpleApiKey.class));
 586         assertEquals(1, captor.getValue().getLong("count"));
 587     }
 588
 589 }