EventsService authorization refactor

-ErrorResponseProvider extracted
-subscriber method handle with refactor
-publisher method handle with partial refactor

Issue-ID: DMAAP-1230
Signed-off-by: Tomek Kaminski <tomasz.kaminski@nokia.com>
Change-Id: I92d53edd0c791f8ce90275776b3031bb0047810e
Signed-off-by: Tomek Kaminski <tomasz.kaminski@nokia.com>

6 files changed, 719 insertions, 397 deletions

diff --git a/src/test/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImplTest.java b/src/test/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImplTest.java
new file mode 100644
index 0000000..0b8829c
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImplTest.java
@@ -0,0 +1,117 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dmaap.dmf.mr.security;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.given;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Spy;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.mock.web.MockHttpServletRequest;
+
+@RunWith(MockitoJUnitRunner.class)
+public class DMaaPAAFAuthenticatorImplTest {
+
+    private MockHttpServletRequest request;
+    @Spy
+    private DMaaPAAFAuthenticatorImpl aafAuthorizer;
+
+    @Before
+    public void setUp() throws Exception {
+        request = new MockHttpServletRequest();
+    }
+
+
+    @Test
+    public void aafAuthentication_shouldSuccess_whenRequestIsConfiguredWithProperUserRole() {
+        //given
+        String userRole = "org.onap.dmaap.mr.topic|:topic.org.onap.dmaap.mr.aSimpleTopic|sub";
+        request.addUserRole(userRole);
+
+        //when
+        boolean isAuthorized = aafAuthorizer.aafAuthentication(request, userRole);
+
+        //then
+        assertTrue(isAuthorized);
+    }
+
+    @Test
+    public void aafAuthentication_shouldFail_whenRequestIsConfiguredWithProperUserRole() {
+        //given
+        String userRole = "org.onap.dmaap.mr.topic|:topic.org.onap.dmaap.mr.aSimpleTopic|pub";
+
+        //when
+        boolean isAuthorized = aafAuthorizer.aafAuthentication(request, userRole);
+
+        //then
+        assertFalse(isAuthorized);
+    }
+
+    @Test
+    public void getPermissionAsString_shouldReturnValidTopicPermission_whenTopicWithNamespace() throws Exception {
+        //given
+        String topicPermission = "org.onap.dmaap.mr.topic|:topic.org.onap.dmaap.mr.aSimpleTopic|pub";
+        String topicName = "org.onap.dmaap.mr.aSimpleTopic";
+        String operation = "pub";
+
+        //when
+        String resultPem = aafAuthorizer.aafPermissionString(topicName, operation);
+
+        //then
+        assertEquals(topicPermission, resultPem);
+    }
+
+    @Test
+    public void getPermissionAsString_shouldReturnValidTopicPermission_whenTopicWithoutNamespace() throws Exception {
+        //given
+        String topicPermission = "org.onap.dmaap.mr.topic|:topic.topicName|pub";
+        String topicName = "topicName";
+        String operation = "pub";
+
+        //when
+        String resultPem = aafAuthorizer.aafPermissionString(topicName, operation);
+
+        //then
+        assertEquals(topicPermission, resultPem);
+    }
+
+    @Test
+    public void getPermissionAsString_shouldReturnValidTopicPermission_whenNamespaceReadFromProperty() throws Exception {
+        //given
+        String topicPermission = "com.custom.ns.topic|:topic.topicName|pub";
+        String topicName = "topicName";
+        String operation = "pub";
+        String customNamespace = "com.custom.ns";
+        given(aafAuthorizer.readNamespaceFromProperties()).willReturn(customNamespace);
+
+        //when
+        String resultPem = aafAuthorizer.aafPermissionString(topicName, operation);
+
+        //then
+        assertEquals(topicPermission, resultPem);
+    }
+
+
+}
diff --git a/src/test/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImplTest.java b/src/test/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImplTest.java
new file mode 100644
index 0000000..4abbe89
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImplTest.java
@@ -0,0 +1,598 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dmaap.dmf.mr.service.impl;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.att.nsa.limits.Blacklist;
+import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
+import com.att.nsa.security.db.simple.NsaSimpleApiKey;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.ConcurrentModificationException;
+import javax.servlet.http.HttpServletRequest;
+import joptsimple.internal.Strings;
+import org.apache.http.HttpStatus;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.onap.dmaap.dmf.mr.CambriaApiException;
+import org.onap.dmaap.dmf.mr.backends.Consumer;
+import org.onap.dmaap.dmf.mr.backends.ConsumerFactory;
+import org.onap.dmaap.dmf.mr.backends.ConsumerFactory.UnavailableException;
+import org.onap.dmaap.dmf.mr.backends.MetricsSet;
+import org.onap.dmaap.dmf.mr.backends.Publisher;
+import org.onap.dmaap.dmf.mr.beans.DMaaPCambriaLimiter;
+import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
+import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaMetaBroker;
+import org.onap.dmaap.dmf.mr.exception.DMaaPAccessDeniedException;
+import org.onap.dmaap.dmf.mr.exception.DMaaPErrorMessages;
+import org.onap.dmaap.dmf.mr.metabroker.Topic;
+import org.onap.dmaap.dmf.mr.resources.CambriaOutboundEventStream;
+import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticator;
+import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.mock.web.MockHttpServletResponse;
+
+@RunWith(MockitoJUnitRunner.class)
+public class EventsServiceImplTest {
+
+    private InputStream iStream = null;
+    private DMaaPContext dMaapContext = new DMaaPContext();
+    private DMaaPErrorMessages pErrorMessages = new DMaaPErrorMessages();
+    @Mock
+    private ConfigurationReader configurationReader;
+    @Mock
+    private Blacklist blacklist;
+    @Mock
+    private DMaaPAuthenticator<NsaSimpleApiKey> dmaaPAuthenticator;
+    @Mock
+    private NsaSimpleApiKey nsaSimpleApiKey;
+    @Mock
+    private DMaaPKafkaMetaBroker dmaapKafkaMetaBroker;
+    @Mock
+    private Topic createdTopic;
+    @Mock
+    private ConsumerFactory factory;
+    @Mock
+    private Consumer consumer;
+    @Mock
+    private Publisher publisher;
+    @Mock
+    private DMaaPCambriaLimiter limiter;
+    @Mock
+    private MetricsSet metrics;
+    @Spy
+    private EventsServiceImpl eventsService;
+
+
+    @Rule
+    public ExpectedException thrown = ExpectedException.none();
+
+    private MockHttpServletRequest request;
+
+    @Before
+    public void setUp() throws Exception {
+        MockitoAnnotations.initMocks(this);
+        String source = "source of my InputStream";
+        iStream = new ByteArrayInputStream(source.getBytes("UTF-8"));
+
+        request = new MockHttpServletRequest();
+        MockHttpServletResponse response = new MockHttpServletResponse();
+        dMaapContext.setRequest(request);
+        dMaapContext.setResponse(response);
+        when(blacklist.contains(anyString())).thenReturn(false);
+        when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
+        when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
+        dMaapContext.setConfigReader(configurationReader);
+        eventsService.setErrorMessages(pErrorMessages);
+        doReturn("100").when(eventsService).getPropertyFromAJSCmap("timeout");
+    }
+
+    @Test
+    public void getEvents_shouldFailOnAafAuthorization() throws Exception {
+        String topicPrefix = "org.onap.aaf.enforced";
+        String topicName = topicPrefix + ".topicName";
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
+        when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
+        when(eventsService.isCadiEnabled()).thenReturn(true);
+
+        thrown.expect(DMaaPAccessDeniedException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_UNAUTHORIZED)));
+
+        eventsService.getEvents(dMaapContext, topicName, "CG1", "23");
+    }
+
+    @Test
+    public void getEvents_shouldFail_whenRemoteAddressIsBlacklisted() throws Exception {
+        String remoteIp = "10.154.17.115";
+        request.setRemoteAddr(remoteIp);
+        when(blacklist.contains(remoteIp)).thenReturn(true);
+        when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_FORBIDDEN)));
+
+        eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
+    }
+
+    @Test
+    public void getEvents_shouldFail_whenRequestedTopicNotExists() throws Exception {
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(null);
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
+
+        eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
+    }
+
+    @Test
+    public void getEvents_shouldFail_whenConsumerLockCannotBeAcquired() throws Exception {
+        //given
+        String topicName = "testTopic345";
+        String consumerGroup = "CG5";
+        String clientId = "13";
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(configurationReader.getfRateLimiter()).thenReturn(limiter);
+        when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
+        when(configurationReader.getfConsumerFactory()).thenReturn(factory);
+        when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
+        doThrow(new UnavailableException("Could not acquire consumer lock")).when(factory)
+            .getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_SERVICE_UNAVAILABLE)));
+
+        //when
+        eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
+
+        //then
+        verify(factory).getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
+
+    }
+
+    @Test
+    public void getEvents_shouldFail_whenBrokerServicesAreUnavailable() throws Exception {
+        String topicName = "testTopic";
+        String consumerGroup = "CG1";
+        String clientId = "23";
+        when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
+        when(configurationReader.getfConsumerFactory()).thenReturn(factory);
+
+        givenUserAuthorizedWithAAF(request, topicName, "sub");
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_SERVICE_UNAVAILABLE)));
+
+        //when
+        eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
+
+        //then
+        verify(factory).destroyConsumer(topicName, consumerGroup, clientId);
+    }
+
+    private void givenUserAuthorizedWithAAF(MockHttpServletRequest request, String topicName, String operation) {
+        String permission = "org.onap.dmaap.mr.topic|:topic." + topicName + "|" + operation;
+        request.addUserRole(permission);
+    }
+
+    @Test
+    public void getEvents_shouldHandleConcurrentModificationError() throws Exception {
+        String testTopic = "testTopic";
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic(testTopic)).thenReturn(createdTopic);
+        when(configurationReader.getfConsumerFactory()).thenReturn(factory);
+        when(configurationReader.getfRateLimiter()).thenThrow(new ConcurrentModificationException("Error occurred"));
+        givenUserAuthorizedWithAAF(request, testTopic, "sub");
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_CONFLICT)));
+
+        eventsService.getEvents(dMaapContext, "testTopic", "CG1", "23");
+    }
+
+    @Test
+    public void getEvents_shouldNotAuthorizeClient_whenSubscribingToMetricsTopic() throws Exception {
+        //given
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        String metricsTopicName = "msgrtr.apinode.metrics.dmaap";
+        String consumerGroup = "CG5";
+        String clientId = "7";
+        givenConfiguredWithMocks(metricsTopicName);
+        when(factory.getConsumerFor(eq(metricsTopicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
+            .thenReturn(consumer);
+        doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+
+        //when
+        eventsService.getEvents(dMaapContext, metricsTopicName, consumerGroup, clientId);
+
+        //then
+        verify(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        verify(dmaaPAuthenticator, never()).authenticate(dMaapContext);
+        verify(permittedRequest, never()).isUserInRole(anyString());
+    }
+
+    @Test
+    public void getEvents_shouldNotAuthorizeClient_whenTopicNoteEnforcedWithAaf_andTopicHasNoOwnerSet()
+        throws Exception {
+        //given
+        String topicName = "someSimpleTopicName";
+        String consumerGroup = "CG5";
+        String clientId = "7";
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        givenConfiguredWithMocks(topicName);
+        when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
+            .thenReturn(consumer);
+        doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        when(createdTopic.getOwner()).thenReturn(Strings.EMPTY);
+
+        //when
+        eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
+
+        //then
+        verify(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        verify(dmaaPAuthenticator, never()).authenticate(dMaapContext);
+        verify(permittedRequest, never()).isUserInRole(anyString());
+    }
+
+    @Test
+    public void getEvents_shouldFailDmaapAuthorization_whenTopicOwnerIsSet_andUserHasNoReadPermissionToTopic()
+        throws Exception {
+        //given
+        String topicName = "someSimpleTopicName";
+        String consumerGroup = "CG5";
+        String clientId = "7";
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        givenConfiguredWithMocks(topicName);
+        when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
+            .thenReturn(consumer);
+        doNothing().when(eventsService).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        when(createdTopic.getOwner()).thenReturn("SimpleTopicOwner");
+        when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
+        doThrow(new AccessDeniedException("userName")).when(createdTopic).checkUserRead(nsaSimpleApiKey);
+
+        thrown.expect(AccessDeniedException.class);
+
+        //when
+        eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
+
+        //then
+        verify(createdTopic).checkUserRead(nsaSimpleApiKey);
+        verify(eventsService, never()).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        verify(permittedRequest, never()).isUserInRole(anyString());
+    }
+
+
+    @Test
+    public void getEvents_shouldSuccessfullyRegisterConsumerToEventsStream_withAafAuthorization() throws Exception {
+        //given
+        String topicName = "testTopic";
+        String consumerGroup = "CG2";
+        String clientId = "6";
+        String messageLimit = "10";
+        String timeout = "25";
+        String meta = "yes";
+        String pretty = "on";
+        String cacheEnabled = "false";
+
+        givenConfiguredWithMocks(topicName);
+        givenConfiguredWithProperties(messageLimit, timeout, meta, pretty, cacheEnabled);
+        when(factory.getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString()))
+            .thenReturn(consumer);
+        givenUserAuthorizedWithAAF(request, topicName, "sub");
+
+        //when
+        eventsService.getEvents(dMaapContext, topicName, consumerGroup, clientId);
+
+        //then
+        ArgumentCaptor<CambriaOutboundEventStream> osWriter = ArgumentCaptor.forClass(CambriaOutboundEventStream.class);
+        verifyInvocationOrderForSuccessCase(topicName, consumerGroup, clientId, osWriter);
+        assertEventStreamProperties(osWriter.getValue(), messageLimit, timeout);
+    }
+
+    private void assertEventStreamProperties(CambriaOutboundEventStream stream, String messageLimit, String timeout) {
+        assertEquals(Integer.valueOf(messageLimit).intValue(), stream.getfLimit());
+        assertEquals(Integer.valueOf(timeout).intValue(), stream.getfTimeoutMs());
+        assertTrue(stream.isfWithMeta());
+        assertTrue(stream.isfPretty());
+    }
+
+    private void givenConfiguredWithProperties(String messageLimit, String timeout, String meta, String pretty,
+        String cacheEnabled) {
+        when(eventsService.getPropertyFromAJSCmap("meta")).thenReturn(meta);
+        when(eventsService.getPropertyFromAJSCmap("pretty")).thenReturn(pretty);
+        when(eventsService.getPropertyFromAJSCmap(ConsumerFactory.kSetting_EnableCache)).thenReturn(cacheEnabled);
+        request.addParameter("timeout", timeout);
+        request.addParameter("limit", messageLimit);
+    }
+
+    private void givenConfiguredWithMocks(String topicName) throws Exception {
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(configurationReader.getfRateLimiter()).thenReturn(limiter);
+        when(configurationReader.getfMetrics()).thenReturn(metrics);
+        when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
+        when(configurationReader.getfConsumerFactory()).thenReturn(factory);
+        when(configurationReader.getfPublisher()).thenReturn(publisher);
+    }
+
+    private void verifyInvocationOrderForSuccessCase(String topicName, String consumerGroup, String clientId,
+        ArgumentCaptor<CambriaOutboundEventStream> osWriter) throws Exception {
+
+        InOrder inOrder = Mockito.inOrder(configurationReader, factory, metrics, limiter, consumer, eventsService);
+        inOrder.verify(configurationReader).getfMetrics();
+        inOrder.verify(configurationReader).getfRateLimiter();
+        inOrder.verify(limiter).onCall(eq(topicName), eq(consumerGroup), eq(clientId), anyString());
+        inOrder.verify(factory).getConsumerFor(eq(topicName), eq(consumerGroup), eq(clientId), anyInt(), anyString());
+        inOrder.verify(eventsService).respondOkWithStream(eq(dMaapContext), osWriter.capture());
+        inOrder.verify(consumer).commitOffsets();
+        inOrder.verify(metrics).consumeTick(anyInt());
+        inOrder.verify(limiter).onSend(eq(topicName), eq(consumerGroup), eq(clientId), anyLong());
+        inOrder.verify(consumer).close();
+        inOrder.verifyNoMoreInteractions();
+    }
+
+    @Test
+    public void pushEvents_shouldFail_whenRemoteAddressIsBlacklisted() throws Exception {
+        String remoteIp = "10.132.64.112";
+        request.setRemoteAddr(remoteIp);
+        when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
+        when(blacklist.contains(anyString())).thenReturn(true);
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_FORBIDDEN)));
+
+        eventsService.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
+    }
+
+
+    @Test
+    public void pushEvents_shouldFail_whenRequestedTopicDoesNotExist() throws Exception {
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(null);
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
+
+        eventsService.pushEvents(dMaapContext, "testTopic", iStream, "5", "13:00:00");
+    }
+
+    @Test
+    public void pushEvents_shouldFailDmaapAuthorization_whenTopicOwnerIsSet_andUserHasNoWritePermissionToTopic()
+        throws Exception {
+        //given
+        String topicName = "someSimpleTopicName";
+
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        givenConfiguredWithMocks(topicName);
+        when(createdTopic.getOwner()).thenReturn("SimpleTopicOwner");
+        when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
+        doThrow(new AccessDeniedException("userName")).when(createdTopic).checkUserWrite(nsaSimpleApiKey);
+
+        thrown.expect(AccessDeniedException.class);
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(createdTopic).checkUserWrite(nsaSimpleApiKey);
+        verify(eventsService, never()).respondOkWithStream(eq(dMaapContext), any(CambriaOutboundEventStream.class));
+        verify(permittedRequest, never()).isUserInRole(anyString());
+    }
+
+    @Test
+    public void pushEvents_shouldFailOnAafAuthorization_whenCadiIsEnabled_topicNameEnforced_andUserHasNoPermission()
+        throws Exception {
+        //given
+        String topicPrefix = "org.onap.aaf.enforced";
+        String topicName = topicPrefix + ".topicName";
+        String permission = "org.onap.dmaap.mr.topic|:topic." + topicName + "|pub";
+        HttpServletRequest deniedRequest = mock(HttpServletRequest.class);
+        when(deniedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        when(deniedRequest.isUserInRole(permission)).thenReturn(false);
+        when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
+        when(dmaapKafkaMetaBroker.getTopic(topicName)).thenReturn(createdTopic);
+        when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
+        when(eventsService.isCadiEnabled()).thenReturn(true);
+        dMaapContext.setRequest(deniedRequest);
+
+        thrown.expect(DMaaPAccessDeniedException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_UNAUTHORIZED)));
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(deniedRequest).isUserInRole(permission);
+    }
+
+
+    @Test
+    public void pushEvents_shouldPublishMessagesWithoutTransaction() throws Exception {
+        //given
+        String topicName = "topicWithoutTransaction";
+        givenConfiguredWithMocks(topicName);
+        doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
+        verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
+        assertEquals(1, captor.getValue().getLong("count"));
+    }
+
+    @Test
+    public void pushEvents_shouldHandlePublisherError_whenPushWithoutTransaction() throws Exception {
+        //given
+        String topicName = "topicWithoutTransaction";
+        givenConfiguredWithMocks(topicName);
+        doThrow(new IOException()).when(publisher)
+            .sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        verify(eventsService, never()).respondOk(any(DMaaPContext.class), any(JSONObject.class));
+    }
+
+
+    @Test
+    public void pushEvents_shouldPublishMessagesWithTransaction() throws Exception {
+        //given
+        String topicPrefix = "org.onap.dmaap.mr";
+        String topicName = topicPrefix + ".topicWithTransaction";
+        givenConfiguredWithMocks(topicName);
+        when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
+        when(eventsService.isCadiEnabled()).thenReturn(true);
+        doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
+
+        request.addUserRole("org.onap.dmaap.mr.topic|:topic." + topicName + "|pub");
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
+        verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
+        assertEquals(1, captor.getValue().getLong("count"));
+        assertFalse(captor.getValue().getString("transactionId").isEmpty());
+    }
+
+    @Test
+    public void pushEvents_shouldHandlePublisherError_whenPushWithTransaction() throws Exception {
+        //given
+        String topicPrefix = "org.onap.dmaap.mr";
+        String topicName = topicPrefix + ".topicWithTransaction";
+        givenConfiguredWithMocks(topicName);
+        when(eventsService.getPropertyFromAJSCmap("enforced.topic.name.AAF")).thenReturn(topicPrefix);
+        when(eventsService.isCadiEnabled()).thenReturn(true);
+        doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
+        request.addUserRole("org.onap.dmaap.mr.topic|:topic." + topicName + "|pub");
+        doThrow(new IOException()).when(publisher)
+            .sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+
+        thrown.expect(CambriaApiException.class);
+        thrown.expectMessage(containsString(String.valueOf(HttpStatus.SC_NOT_FOUND)));
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        verify(eventsService, never()).respondOk(any(DMaaPContext.class), any(JSONObject.class));
+    }
+
+    @Test
+    public void pushEvents_shouldNotPerformAnyAuthorization_whenPublishToMetricTopic() throws Exception {
+        //given
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        String metricsTopicName = "msgrtr.apinode.metrics.dmaap";
+        givenConfiguredWithMocks(metricsTopicName);
+        doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
+
+        //when
+        eventsService.pushEvents(dMaapContext, metricsTopicName, iStream, "5", "13:00:00");
+
+        //then
+        ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
+        verify(publisher)
+            .sendBatchMessageNew(eq(metricsTopicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
+        verify(permittedRequest, never()).isUserInRole(anyString());
+        verify(createdTopic, never()).checkUserWrite(any(NsaSimpleApiKey.class));
+        assertEquals(1, captor.getValue().getLong("count"));
+    }
+
+    @Test
+    public void pushEvents_shouldNotPerformAnyAuthorization_whenTopicHasNoOwner() throws Exception {
+        //given
+        HttpServletRequest permittedRequest = mock(HttpServletRequest.class);
+        when(permittedRequest.getHeaders(anyString())).thenReturn(Collections.<String>emptyEnumeration());
+        dMaapContext.setRequest(permittedRequest);
+        String topicName = "notEnforcedAafTopic";
+        givenConfiguredWithMocks(topicName);
+        doNothing().when(eventsService).respondOk(eq(dMaapContext), any(JSONObject.class));
+        when(createdTopic.getOwner()).thenReturn(null);
+
+        //when
+        eventsService.pushEvents(dMaapContext, topicName, iStream, "5", "13:00:00");
+
+        //then
+        ArgumentCaptor<JSONObject> captor = ArgumentCaptor.forClass(JSONObject.class);
+        verify(publisher).sendBatchMessageNew(eq(topicName), Mockito.<ArrayList<ProducerRecord<String, String>>>any());
+        verify(eventsService).respondOk(eq(dMaapContext), captor.capture());
+        verify(permittedRequest, never()).isUserInRole(anyString());
+        verify(createdTopic, never()).checkUserWrite(any(NsaSimpleApiKey.class));
+        assertEquals(1, captor.getValue().getLong("count"));
+    }
+
+}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/security/DMaaPAAFAuthenticatorImplTest.java b/src/test/java/org/onap/dmaap/mr/cambria/security/DMaaPAAFAuthenticatorImplTest.java
deleted file mode 100644
index 7019a2b..0000000
--- a/src/test/java/org/onap/dmaap/mr/cambria/security/DMaaPAAFAuthenticatorImplTest.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine
- * ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- * 
- *      http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
- package org.onap.dmaap.mr.cambria.security;
-
-import static org.junit.Assert.*;
-
-import javax.servlet.http.HttpServletRequest;
-
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.springframework.mock.web.MockHttpServletRequest;
-
-import org.onap.dmaap.dmf.mr.CambriaApiException;
-import org.onap.dmaap.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
-
-
-
-public class DMaaPAAFAuthenticatorImplTest {
-	
-	private MockHttpServletRequest request = null;
-	@Before
-	public void setUp() throws Exception {
-		//creating servlet object
-		request = new MockHttpServletRequest();
-		request.setServerName("www.example.com");
-		request.setRequestURI("/foo");
-		request.setQueryString("param1=value1&param");
-		String url = request.getRequestURL() + "?" + request.getQueryString(); 
-
-		
-	}
-
-	@After
-	public void tearDown() throws Exception {
-	}
-
-	@Test
-	public void testAafAuthentication() {
-		
-		DMaaPAAFAuthenticatorImpl authenticator = new DMaaPAAFAuthenticatorImpl();
-		authenticator.aafAuthentication(request, "admin");
-		assertTrue(true);
-		
-	}
-	
-	
-	
-	/*@Test
-	public void testAafPermissionString() {
-		
-		DMaaPAAFAuthenticatorImpl authenticator = new DMaaPAAFAuthenticatorImpl();
-		try {
-			authenticator.aafPermissionString("testTopic", "admin");
-		} catch (CambriaApiException e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-		}
-		
-		assertTrue(true);
-		
-	}*/
-	
-
-}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/security/JUnitTestSuite.java b/src/test/java/org/onap/dmaap/mr/cambria/security/JUnitTestSuite.java
index 60ae849..ea3f051 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/security/JUnitTestSuite.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/security/JUnitTestSuite.java
@@ -26,9 +26,10 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Suite;
 import org.junit.runners.Suite.SuiteClasses;
 import org.apache.log4j.Logger;
+import org.onap.dmaap.dmf.mr.security.DMaaPAAFAuthenticatorImplTest;
 
 @RunWith(Suite.class)
-@SuiteClasses({ DMaaPAAFAuthenticatorImplTest.class,DMaaPAuthenticatorImplTest.class, 
+@SuiteClasses({ DMaaPAAFAuthenticatorImplTest.class,DMaaPAuthenticatorImplTest.class,
 })
 public class JUnitTestSuite {
 	private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/service/impl/EventsServiceImplTest.java b/src/test/java/org/onap/dmaap/mr/cambria/service/impl/EventsServiceImplTest.java
deleted file mode 100644
index 1e677d8..0000000
--- a/src/test/java/org/onap/dmaap/mr/cambria/service/impl/EventsServiceImplTest.java
+++ /dev/null
@@ -1,312 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine
- * ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- * 
- *      http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
- package org.onap.dmaap.mr.cambria.service.impl;
-
-import static org.mockito.Mockito.when;
-import static org.mockito.Matchers.anyString;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.ConcurrentModificationException;
-import java.util.Map;
-import java.util.Properties;
-
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.mock.web.MockHttpServletRequest;
-import org.springframework.mock.web.MockHttpServletResponse;
-
-import com.att.ajsc.beans.PropertiesMapBean;
-import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-import org.onap.dmaap.dmf.mr.CambriaApiException;
-import org.onap.dmaap.dmf.mr.security.DMaaPAAFAuthenticator;
-import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticator;
-import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticatorImpl;
-import org.onap.dmaap.dmf.mr.utils.ConfigurationReader;
-import org.onap.dmaap.dmf.mr.backends.ConsumerFactory.UnavailableException;
-import org.onap.dmaap.dmf.mr.beans.DMaaPCambriaLimiter;
-import org.onap.dmaap.dmf.mr.backends.ConsumerFactory;
-import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
-import org.onap.dmaap.dmf.mr.beans.DMaaPKafkaMetaBroker;
-import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
-import org.onap.dmaap.dmf.mr.exception.DMaaPAccessDeniedException;
-import org.onap.dmaap.dmf.mr.exception.DMaaPErrorMessages;
-import org.onap.dmaap.dmf.mr.metabroker.Topic;
-import org.onap.dmaap.dmf.mr.metabroker.Broker.TopicExistsException;
-import org.onap.dmaap.dmf.mr.service.impl.EventsServiceImpl;
-import org.onap.dmaap.dmf.mr.utils.PropertyReader;
-import com.att.nsa.configs.ConfigDbException;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.invalidSettingValue;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.loadException;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
-import com.att.nsa.limits.Blacklist;
-import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
-import com.att.nsa.security.NsaApiKey;
-import com.att.nsa.security.db.simple.NsaSimpleApiKey;
-
-import kafka.admin.AdminUtils;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest({ DMaaPAuthenticatorImpl.class, AJSCPropertiesMap.class })
-public class EventsServiceImplTest {
-
-	private InputStream iStream = null;
-	DMaaPContext dMaapContext = new DMaaPContext();
-	EventsServiceImpl service = new EventsServiceImpl();
-	DMaaPErrorMessages pErrorMessages = new DMaaPErrorMessages();
-	@Mock
-	ConfigurationReader configurationReader;
-	@Mock
-	Blacklist blacklist;
-	@Mock
-	DMaaPAuthenticator<NsaSimpleApiKey> dmaaPAuthenticator;
-	@Mock
-	DMaaPAAFAuthenticator dmaapAAFauthenticator;
-	@Mock
-	NsaApiKey user;
-	@Mock
-	NsaSimpleApiKey nsaSimpleApiKey;
-	@Mock
-	DMaaPKafkaMetaBroker dmaapKafkaMetaBroker;
-	@Mock
-	Topic createdTopic;
-	@Mock
-	ConsumerFactory factory;
-
-	@Before
-	public void setUp() throws Exception {
-		MockitoAnnotations.initMocks(this);
-		String source = "source of my InputStream";
-		iStream = new ByteArrayInputStream(source.getBytes("UTF-8"));
-
-		MockHttpServletRequest request = new MockHttpServletRequest();
-		MockHttpServletResponse response = new MockHttpServletResponse();
-		dMaapContext.setRequest(request);
-		dMaapContext.setResponse(response);
-		when(blacklist.contains(anyString())).thenReturn(false);
-		when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
-		dMaapContext.setConfigReader(configurationReader);
-
-		service.setErrorMessages(pErrorMessages);
-		PowerMockito.mockStatic(AJSCPropertiesMap.class);
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "timeout")).thenReturn("100");
-
-		AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "timeout");
-
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testGetEvents() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException {
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testGetEventsBlackListErr() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException {
-		when(blacklist.contains(anyString())).thenReturn(true);
-		when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
-		dMaapContext.setConfigReader(configurationReader);
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testGetEventsNoTopicError() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException {
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(null);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testGetEventsuserNull() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException {
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		MockHttpServletRequest mockRequest = new MockHttpServletRequest();
-		mockRequest.addHeader("Authorization", "passed");
-		dMaapContext.setRequest(mockRequest);
-		dMaapContext.getRequest().getHeader("Authorization");
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testGetEventsExcp2() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException {
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		when(configurationReader.getfRateLimiter()).thenThrow(new ConcurrentModificationException("Error occurred"));
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testPushEvents() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
-			invalidSettingValue, loadException {
-
-		// AdminUtils.createTopic(configurationReader.getZk(), "testTopic", 10,
-		// 1, new Properties());
-
-		configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
-
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-
-		service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
-
-		service.getEvents(dMaapContext, "testTopic", "CG1", "23");
-
-		/*
-		 * String trueValue = "True";
-		 * assertTrue(trueValue.equalsIgnoreCase("True"));
-		 */
-
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testPushEventsBlackListedIp() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
-			invalidSettingValue, loadException {
-
-		// AdminUtils.createTopic(configurationReader.getZk(), "testTopic", 10,
-		// 1, new Properties());
-		when(blacklist.contains(anyString())).thenReturn(true);
-		when(configurationReader.getfIpBlackList()).thenReturn(blacklist);
-		configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-
-		service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
-
-	}
-
-	@Test(expected = NullPointerException.class)
-	public void testPushEventsNoUser() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
-			invalidSettingValue, loadException {
-
-		configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
-
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(null);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		MockHttpServletRequest mockRequest = new MockHttpServletRequest();
-		mockRequest.addHeader("Authorization", "passed");
-		mockRequest.addHeader("Authorization", "passed");
-		dMaapContext.setRequest(mockRequest);
-		dMaapContext.getRequest().getHeader("Authorization");
-		service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
-
-	}
-
-	@Test(expected = CambriaApiException.class)
-	public void testPushEventsWtTransaction() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
-			invalidSettingValue, loadException {
-
-		configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
-
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "transidUEBtopicreqd")).thenReturn("true");
-
-		service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
-
-	}
-	
-	@Test(expected = CambriaApiException.class)
-	public void testPushEventsWtTransactionError() throws DMaaPAccessDeniedException, CambriaApiException, ConfigDbException,
-			TopicExistsException, AccessDeniedException, UnavailableException, IOException, missingReqdSetting,
-			invalidSettingValue, loadException {
-
-		configurationReader.setfRateLimiter(new DMaaPCambriaLimiter(new PropertyReader()));
-
-		when(dmaaPAuthenticator.authenticate(dMaapContext)).thenReturn(nsaSimpleApiKey);
-		when(configurationReader.getfSecurityManager()).thenReturn(dmaaPAuthenticator);
-		when(configurationReader.getfMetaBroker()).thenReturn(dmaapKafkaMetaBroker);
-		when(dmaapKafkaMetaBroker.getTopic("testTopic")).thenReturn(createdTopic);
-		PowerMockito.when(configurationReader.getfConsumerFactory()).thenReturn(factory);
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "transidUEBtopicreqd")).thenReturn("true");
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "event.batch.length")).thenReturn("0");
-		when(configurationReader.getfPublisher()).thenThrow(new ConcurrentModificationException("Error occurred"));
-
-		service.pushEvents(dMaapContext, "testTopic", iStream, "3", "12:00:00");
-
-	}
-	
-	@Test
-	public void testIsTransEnabled1() {
-
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"transidUEBtopicreqd")).thenReturn("true");
-		  assertTrue(service.isTransEnabled());
-
-	}
-	@Test
-	public void testIsTransEnabled2() {
-
-		when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
-				"transidUEBtopicreqd")).thenReturn("false");
-		  assertFalse(service.isTransEnabled());
-
-	}
-
-}
diff --git a/src/test/java/org/onap/dmaap/mr/cambria/service/impl/JUnitTestSuite.java b/src/test/java/org/onap/dmaap/mr/cambria/service/impl/JUnitTestSuite.java
index ec4b0e2..7536127 100644
--- a/src/test/java/org/onap/dmaap/mr/cambria/service/impl/JUnitTestSuite.java
+++ b/src/test/java/org/onap/dmaap/mr/cambria/service/impl/JUnitTestSuite.java
@@ -25,12 +25,13 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Suite;
 import org.junit.runners.Suite.SuiteClasses;
 import org.apache.log4j.Logger;
+import org.onap.dmaap.dmf.mr.service.impl.EventsServiceImplTest;
 import org.onap.dmaap.dmf.mr.service.impl.TopicServiceImplTest;
 
 @RunWith(Suite.class)
 @SuiteClasses({ UIServiceImplTest.class, AdminServiceImplemTest.class, ApiKeysServiceImplTest.class, 
 	ShowConsumerCacheTest.class,TopicServiceImplTest.class, TransactionServiceImplTest.class, MMServiceImplTest.class,
-	BaseTransactionDbImplTest.class,  MetricsServiceImplTest.class,EventsServiceImplTest.class})
+	BaseTransactionDbImplTest.class,  MetricsServiceImplTest.class, EventsServiceImplTest.class})
 public class JUnitTestSuite {
 	private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);