summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorTomek Kaminski <tomasz.kaminski@nokia.com>2019-07-10 12:17:35 +0200
committerTomek Kaminski <tomasz.kaminski@nokia.com>2019-07-16 13:41:21 +0200
commit08f656f11693e4b172f696d14eec5237217217d0 (patch)
tree4632cfac58774680f1ed94cadcd90fc3d368730c
parent360b04b616c672fd95921f5638e4d843355d082c (diff)
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>
Diffstat
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/exception/DMaaPErrorMessages.java4
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java30
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticator.java4
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImpl.java53
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/service/impl/ErrorResponseProvider.java147
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImpl.java510
-rw-r--r--src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java15
-rw-r--r--src/test/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImplTest.java117
-rw-r--r--src/test/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImplTest.java598
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/security/DMaaPAAFAuthenticatorImplTest.java83
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/security/JUnitTestSuite.java3
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/service/impl/EventsServiceImplTest.java312
-rw-r--r--src/test/java/org/onap/dmaap/mr/cambria/service/impl/JUnitTestSuite.java3
13 files changed, 1133 insertions, 746 deletions
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/exception/DMaaPErrorMessages.java b/src/main/java/org/onap/dmaap/dmf/mr/exception/DMaaPErrorMessages.java
index 7b68b42..95b0577 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/exception/DMaaPErrorMessages.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/exception/DMaaPErrorMessages.java
@@ -58,10 +58,10 @@ public class DMaaPErrorMessages {
private String topicsfailure="Failed to retrieve list of all topics.";
//@Value("${not.permitted.access.1}")
- private String notPermitted1="Access Denied.User does not have permission to perform";
+ private String notPermitted1="Access Denied.User does not have permission to perform ";
//@Value("${not.permitted.access.2}")
- private String notPermitted2="operation on Topic:";
+ private String notPermitted2=" operation on Topic:";
//@Value("${get.topic.details.failure}")
private String topicDetailsFail="Failed to retrieve details of topic:";
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java b/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
index 537fc22..2f7f1db 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/resources/CambriaOutboundEventStream.java
@@ -548,7 +548,35 @@ public class CambriaOutboundEventStream implements StreamWriter {
private ArrayList<Consumer> fKafkaConsumerList;
private boolean istransType = true;
// private static final Logger log =
-
+
private static final EELFLogger log = EELFManager.getInstance().getLogger(CambriaOutboundEventStream.class);
+
+ public int getfLimit() {
+ return fLimit;
+ }
+
+ public int getfTimeoutMs() {
+ return fTimeoutMs;
+ }
+
+ public boolean isfPretty() {
+ return fPretty;
+ }
+
+ public boolean isfWithMeta() {
+ return fWithMeta;
+ }
+
+ public boolean isAAFTopic() {
+ return isAAFTopic;
+ }
+
+ public boolean isIstransEnable() {
+ return istransEnable;
+ }
+
+ public boolean isIstransType() {
+ return istransType;
+ }
} \ No newline at end of file
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticator.java b/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticator.java
index a7f2376..12f0465 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticator.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticator.java
@@ -22,12 +22,8 @@
package org.onap.dmaap.dmf.mr.security;
import javax.servlet.http.HttpServletRequest;
-
import org.onap.dmaap.dmf.mr.CambriaApiException;
-
-
-
/**
*
* @author sneha.d.desai
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImpl.java b/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImpl.java
index 25644a7..0e6b0f6 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImpl.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/security/DMaaPAAFAuthenticatorImpl.java
@@ -21,6 +21,7 @@
*******************************************************************************/
package org.onap.dmaap.dmf.mr.security;
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
import javax.servlet.http.HttpServletRequest;
import org.onap.dmaap.dmf.mr.CambriaApiException;
@@ -34,47 +35,37 @@ import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
*/
public class DMaaPAAFAuthenticatorImpl implements DMaaPAAFAuthenticator {
+ private static final String NAMESPACE_PROPERTY = "defaultNSforUEB";
+ private static final String DEFAULT_NAMESPACE = "org.onap.dmaap.mr";
+ private static final String NAMESPACE_PREFIX = "org.onap";
+
/**
* @param req
* @param role
*/
@Override
public boolean aafAuthentication(HttpServletRequest req, String role) {
- boolean auth = false;
- if(req.isUserInRole(role))
- {
-
- auth = true;
- }
-
- return auth;
+ return req.isUserInRole(role);
}
@Override
public String aafPermissionString(String topicName, String action) throws CambriaApiException {
-
-
- String permission = "";
- String nameSpace ="";
- if(topicName.contains(".") && topicName.contains("org.onap")) {
-
- nameSpace = topicName.substring(0,topicName.lastIndexOf("."));
- }
- else {
- nameSpace = null;
- nameSpace= com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,"defaultNSforUEB");
-
- if(null==nameSpace)nameSpace="org.onap.dmaap.mr";
-
-
-
- }
-
- permission = nameSpace+".topic|:topic."+topicName+"|"+action;
- return permission;
-
+
+ String nameSpace = topicName.startsWith(NAMESPACE_PREFIX) ? parseNamespace(topicName) :
+ readNamespaceFromProperties();
+
+ nameSpace = !nameSpace.isEmpty()? nameSpace : DEFAULT_NAMESPACE;
+
+ return new StringBuilder(nameSpace).append(".topic|:topic.").append(topicName)
+ .append("|").append(action).toString();
+ }
+
+ String readNamespaceFromProperties() {
+ return AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,NAMESPACE_PROPERTY);
+ }
+
+ private String parseNamespace(String topicName) {
+ return topicName.substring(0,topicName.lastIndexOf('.'));
}
-
-
}
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/service/impl/ErrorResponseProvider.java b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/ErrorResponseProvider.java
new file mode 100644
index 0000000..50a611e
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/ErrorResponseProvider.java
@@ -0,0 +1,147 @@
+/*******************************************************************************
+ * ============LICENSE_START===================================================
+ * org.onap.dmaap
+ * ============================================================================
+ * Copyright © 2019 Nokia 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 com.google.common.base.Preconditions;
+import java.util.Date;
+import org.apache.http.HttpStatus;
+import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
+import org.onap.dmaap.dmf.mr.exception.DMaaPErrorMessages;
+import org.onap.dmaap.dmf.mr.exception.DMaaPResponseCode;
+import org.onap.dmaap.dmf.mr.exception.ErrorResponse;
+import org.onap.dmaap.dmf.mr.utils.Utils;
+
+class ErrorResponseProvider {
+
+ private String clientId;
+ private String topicName;
+ private String consumerGroup;
+ private String remoteHost;
+ private String publisherId;
+ private String publisherIp;
+ private DMaaPErrorMessages errorMessages;
+
+ private ErrorResponseProvider() {
+
+ }
+
+ ErrorResponse getIpBlacklistedError(String remoteAddr) {
+ return new ErrorResponse(HttpStatus.SC_FORBIDDEN,
+ DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
+ "Source address [" + remoteAddr + "] is blacklisted. Please contact the cluster management team.",
+ null, Utils.getFormattedDate(new Date()), topicName, publisherId,
+ publisherIp, consumerGroup + "/" + clientId, remoteHost);
+ }
+
+ ErrorResponse getTopicNotFoundError() {
+ return new ErrorResponse(HttpStatus.SC_NOT_FOUND,
+ DMaaPResponseCode.RESOURCE_NOT_FOUND.getResponseCode(),
+ errorMessages.getTopicNotExist() + "-[" + topicName + "]", null, Utils.getFormattedDate(new Date()),
+ topicName, publisherId, publisherIp, consumerGroup + "/" + clientId, remoteHost);
+ }
+
+ ErrorResponse getAafAuthorizationError(String permission, String action) {
+ return new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
+ DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
+ errorMessages.getNotPermitted1() + action + errorMessages.getNotPermitted2() + topicName + " on "
+ + permission,
+ null, Utils.getFormattedDate(new Date()), topicName, publisherId, publisherIp, consumerGroup + "/" + clientId,
+ remoteHost);
+ }
+
+ ErrorResponse getServiceUnavailableError(String msg) {
+ return new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
+ DMaaPResponseCode.SERVER_UNAVAILABLE.getResponseCode(),
+ errorMessages.getServerUnav() + msg, null, Utils.getFormattedDate(new Date()), topicName,
+ publisherId, publisherIp, consumerGroup + "/" + clientId, remoteHost);
+ }
+
+ ErrorResponse getConcurrentModificationError() {
+ return new ErrorResponse(HttpStatus.SC_CONFLICT,
+ DMaaPResponseCode.TOO_MANY_REQUESTS.getResponseCode(),
+ "Couldn't respond to client, possible of consumer requests from more than one server. Please contact MR team if you see this issue occurs continously", null,
+ Utils.getFormattedDate(new Date()), topicName, publisherId, publisherIp, consumerGroup + "/" + clientId, remoteHost);
+ }
+
+ ErrorResponse getGenericError(String msg) {
+ return new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
+ DMaaPResponseCode.SERVER_UNAVAILABLE.getResponseCode(),
+ "Couldn't respond to client, closing cambria consumer" + msg, null,
+ Utils.getFormattedDate(new Date()), topicName, publisherId, publisherIp, consumerGroup + "/" + clientId, remoteHost);
+ }
+
+ public static class Builder {
+
+ private String clientId;
+ private String topicName;
+ private String consumerGroup;
+ private String remoteHost;
+ private String publisherId;
+ private String publisherIp;
+ DMaaPErrorMessages errorMessages;
+
+ Builder withErrorMessages(DMaaPErrorMessages errorMessages) {
+ this.errorMessages = errorMessages;
+ return this;
+ }
+
+ Builder withTopic(String topic) {
+ this.topicName = topic;
+ return this;
+ }
+
+ Builder withClient(String client) {
+ this.clientId = client;
+ return this;
+ }
+
+ Builder withConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ return this;
+ }
+
+ Builder withRemoteHost(String remoteHost) {
+ this.remoteHost = remoteHost;
+ return this;
+ }
+
+ Builder withPublisherId(String publisherId) {
+ this.publisherId = publisherId;
+ return this;
+ }
+
+ Builder withPublisherIp(String publisherIp) {
+ this.publisherIp = publisherIp;
+ return this;
+ }
+
+ public ErrorResponseProvider build() {
+ Preconditions.checkArgument(errorMessages!=null);
+ ErrorResponseProvider errRespProvider = new ErrorResponseProvider();
+ errRespProvider.errorMessages = this.errorMessages;
+ errRespProvider.clientId = this.clientId;
+ errRespProvider.consumerGroup = this.consumerGroup;
+ errRespProvider.topicName = this.topicName;
+ errRespProvider.remoteHost = this.remoteHost;
+ errRespProvider.publisherId = this.publisherId;
+ errRespProvider.publisherIp = this.publisherIp;
+ return errRespProvider;
+ }
+ }
+}
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImpl.java b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImpl.java
index 11c544f..ec5bfc0 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImpl.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/service/impl/EventsServiceImpl.java
@@ -21,34 +21,32 @@
*******************************************************************************/
package org.onap.dmaap.dmf.mr.service.impl;
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+import com.att.nsa.configs.ConfigDbException;
+import com.att.nsa.drumlin.service.standards.MimeTypes;
+import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
+import com.att.nsa.security.NsaApiKey;
+import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
+import com.att.nsa.util.rrConvertor;
import java.io.IOException;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.ConcurrentModificationException;
import java.util.Date;
-import java.util.HashMap;
import java.util.LinkedList;
-import java.util.Properties;
-
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.core.MediaType;
-
+import org.apache.commons.lang.math.NumberUtils;
import org.apache.http.HttpStatus;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.TopicExistsException;
import org.json.JSONObject;
import org.json.JSONTokener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.stereotype.Service;
-
-import com.att.ajsc.filemonitor.AJSCPropertiesMap;
import org.onap.dmaap.dmf.mr.CambriaApiException;
import org.onap.dmaap.dmf.mr.backends.Consumer;
import org.onap.dmaap.dmf.mr.backends.ConsumerFactory;
@@ -56,7 +54,6 @@ 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.backends.Publisher.message;
-import org.onap.dmaap.dmf.mr.backends.kafka.KafkaLiveLockAvoider2;
import org.onap.dmaap.dmf.mr.beans.DMaaPCambriaLimiter;
import org.onap.dmaap.dmf.mr.beans.DMaaPContext;
import org.onap.dmaap.dmf.mr.beans.LogDetails;
@@ -65,7 +62,6 @@ import org.onap.dmaap.dmf.mr.exception.DMaaPAccessDeniedException;
import org.onap.dmaap.dmf.mr.exception.DMaaPErrorMessages;
import org.onap.dmaap.dmf.mr.exception.DMaaPResponseCode;
import org.onap.dmaap.dmf.mr.exception.ErrorResponse;
-
import org.onap.dmaap.dmf.mr.metabroker.Topic;
import org.onap.dmaap.dmf.mr.resources.CambriaEventSet;
import org.onap.dmaap.dmf.mr.resources.CambriaOutboundEventStream;
@@ -74,15 +70,10 @@ import org.onap.dmaap.dmf.mr.security.DMaaPAAFAuthenticatorImpl;
import org.onap.dmaap.dmf.mr.security.DMaaPAuthenticatorImpl;
import org.onap.dmaap.dmf.mr.service.EventsService;
import org.onap.dmaap.dmf.mr.utils.DMaaPResponseBuilder;
+import org.onap.dmaap.dmf.mr.utils.DMaaPResponseBuilder.StreamWriter;
import org.onap.dmaap.dmf.mr.utils.Utils;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-import com.att.nsa.configs.ConfigDbException;
-import com.att.nsa.drumlin.service.standards.MimeTypes;
-import com.att.nsa.drumlin.till.nv.rrNvReadable.missingReqdSetting;
-import com.att.nsa.security.NsaApiKey;
-import com.att.nsa.security.ReadWriteSecuredResource.AccessDeniedException;
-import com.att.nsa.util.rrConvertor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
/**
* This class provides the functinality to publish and subscribe message to
@@ -93,20 +84,20 @@ import com.att.nsa.util.rrConvertor;
*/
@Service
public class EventsServiceImpl implements EventsService {
- // private static final Logger LOG =
private static final EELFLogger LOG = EELFManager.getInstance().getLogger(EventsServiceImpl.class);
-
private static final String BATCH_LENGTH = "event.batch.length";
private static final String TRANSFER_ENCODING = "Transfer-Encoding";
+ private static final String TIMEOUT_PROPERTY = "timeout";
+ private static final String SUBSCRIBE_ACTION = "sub";
+ private static final String PUBLISH_ACTION = "pub";
+
@Autowired
private DMaaPErrorMessages errorMessages;
-
- //@Autowired
-
- // @Value("${metrics.send.cambria.topic}")
-
+ String getPropertyFromAJSCmap(String propertyKey) {
+ return AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, propertyKey);
+ }
public DMaaPErrorMessages getErrorMessages() {
return errorMessages;
@@ -129,222 +120,91 @@ public class EventsServiceImpl implements EventsService {
*/
@Override
public void getEvents(DMaaPContext ctx, String topic, String consumerGroup, String clientId)
- throws ConfigDbException, TopicExistsException, AccessDeniedException, UnavailableException,
- CambriaApiException, IOException, DMaaPAccessDeniedException {
+ throws ConfigDbException, AccessDeniedException, UnavailableException,
+ CambriaApiException, IOException {
+
final long startTime = System.currentTimeMillis();
final HttpServletRequest req = ctx.getRequest();
-
- boolean isAAFTopic = false;
- // was this host blacklisted?
- final String remoteAddr = Utils.getRemoteAddress(ctx);
- if (ctx.getConfigReader().getfIpBlackList().contains(remoteAddr)) {
-
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
- DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
- "Source address [" + remoteAddr + "] is blacklisted. Please contact the cluster management team.",
- null, Utils.getFormattedDate(new Date()), topic, Utils.getUserApiKey(ctx.getRequest()),
- ctx.getRequest().getRemoteHost(), null, null);
- LOG.info(errRes.toString());
- throw new CambriaApiException(errRes);
- }
-
- int limit = CambriaConstants.kNoLimit;
- if (req.getParameter("limit") != null) {
- limit = Integer.parseInt(req.getParameter("limit"));
- }
-
- int timeoutMs = CambriaConstants.kNoTimeout;
- String strtimeoutMS = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "timeout");
- if (strtimeoutMS != null)
- timeoutMs = Integer.parseInt(strtimeoutMS);
- // int timeoutMs = ctx.getConfigReader().getSettings().getInt("timeout",
-
- if (req.getParameter("timeout") != null) {
- timeoutMs = Integer.parseInt(req.getParameter("timeout"));
- }
-
- // By default no filter is applied if filter is not passed as a
- // parameter in the request URI
- String topicFilter = CambriaConstants.kNoFilter;
- if (null != req.getParameter("filter")) {
- topicFilter = req.getParameter("filter");
- }
- // pretty to print the messaages in new line
- String prettyval = "0";
- String strPretty = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "pretty");
- if (null != strPretty)
- prettyval = strPretty;
-
- String metaval = "0";
- String strmeta = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "meta");
- if (null != strmeta)
- metaval = strmeta;
-
- final boolean pretty = rrConvertor.convertToBooleanBroad(prettyval);
- // withMeta to print offset along with message
- final boolean withMeta = rrConvertor.convertToBooleanBroad(metaval);
-
final LogWrap logger = new LogWrap(topic, consumerGroup, clientId);
- logger.info("fetch: timeout=" + timeoutMs + ", limit=" + limit + ", filter=" + topicFilter + " from Remote host "+ctx.getRequest().getRemoteHost());
+ final String remoteHost = req.getRemoteHost();
+ ErrorResponseProvider errRespProvider = new ErrorResponseProvider.Builder().withErrorMessages(errorMessages)
+ .withTopic(topic).withConsumerGroup(consumerGroup).withClient(clientId).withRemoteHost(remoteHost).build();
- // is this user allowed to read this topic?
- final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(ctx);
- final Topic metatopic = ctx.getConfigReader().getfMetaBroker().getTopic(topic);
+ validateIpBlacklist(errRespProvider, ctx);
- if (metatopic == null) {
- // no such topic.
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_NOT_FOUND,
- DMaaPResponseCode.RESOURCE_NOT_FOUND.getResponseCode(),
- errorMessages.getTopicNotExist() + "-[" + topic + "]", null, Utils.getFormattedDate(new Date()),
- topic, null, null, consumerGroup + "/" + clientId, ctx.getRequest().getRemoteHost());
- LOG.info(errRes.toString());
- throw new CambriaApiException(errRes);
- }
- String metricTopicname = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "metrics.send.cambria.topic");
- if (null == metricTopicname)
- metricTopicname = "msgrtr.apinode.metrics.dmaap";
-
- boolean topicNameEnforced = false;
- String topicNameStd = null;
- topicNameStd = com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,
- "enforced.topic.name.AAF");
- if (null != topicNameStd && topic.startsWith(topicNameStd)) {
- topicNameEnforced = true;
+ final Topic metaTopic = ctx.getConfigReader().getfMetaBroker().getTopic(topic);
+ if (metaTopic == null) {
+ throw new CambriaApiException(errRespProvider.getTopicNotFoundError());
}
- if (null == ctx.getRequest().getHeader("Authorization") && !topic.equalsIgnoreCase(metricTopicname)) {
- if (null != metatopic.getOwner() && !("".equals(metatopic.getOwner()))) {
- // check permissions
- metatopic.checkUserRead(user);
- }
- }
- // if headers are not provided then user will be null
- if (topicNameEnforced ||(user == null && null != ctx.getRequest().getHeader("Authorization"))) {
- // the topic name will be sent by the client
-
- DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
- String permission = aaf.aafPermissionString(topic, "sub");
- if (!aaf.aafAuthentication(ctx.getRequest(), permission)) {
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
- DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
- errorMessages.getNotPermitted1() + " read " + errorMessages.getNotPermitted2() + topic + " on "
- + permission,
- null, Utils.getFormattedDate(new Date()), topic, null, null, consumerGroup + "/" + clientId,
- ctx.getRequest().getRemoteHost());
- LOG.info(errRes.toString());
- throw new DMaaPAccessDeniedException(errRes);
+ boolean isAAFTopic = authorizeClientWhenNeeded(ctx, metaTopic, topic, errRespProvider, SUBSCRIBE_ACTION);
- }
- isAAFTopic = true;
- }
final long elapsedMs1 = System.currentTimeMillis() - startTime;
logger.info("Time taken in getEvents Authorization " + elapsedMs1 + " ms for " + topic + " " + consumerGroup
+ " " + clientId);
- Consumer c = null;
-
- String lhostId = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "clusterhostid");
- if (null == lhostId) {
- try {
- lhostId = InetAddress.getLocalHost().getCanonicalHostName();
- } catch (UnknownHostException e) {
- LOG.info("Unknown Host Exception error occured while getting getting hostid");
- }
- }
- CambriaOutboundEventStream coes = null;
+ verifyHostId();
+ final boolean pretty = isPrettyPrintEnabled();
+ final boolean withMeta = isMetaOffsetEnabled();
+ int timeoutMs = getMessageTimeout(req);
+ int limit = getMessageLimit(req);
+ String topicFilter = (null != req.getParameter("filter")) ? req.getParameter("filter") : CambriaConstants.kNoFilter;
+ logger.info("fetch: timeout=" + timeoutMs + ", limit=" + limit + ", filter=" + topicFilter + " from Remote host "+ctx.getRequest().getRemoteHost());
+
+ Consumer consumer = null;
try {
final MetricsSet metricsSet = ctx.getConfigReader().getfMetrics();
final DMaaPCambriaLimiter rl = ctx.getConfigReader().getfRateLimiter();
- rl.onCall(topic, consumerGroup, clientId, ctx.getRequest().getRemoteHost());
- c = ctx.getConfigReader().getfConsumerFactory().getConsumerFor(topic, consumerGroup, clientId, timeoutMs,
- ctx.getRequest().getRemoteHost());
- coes = new CambriaOutboundEventStream.Builder(c).timeout(timeoutMs)
+ rl.onCall(topic, consumerGroup, clientId, remoteHost);
+ consumer = ctx.getConfigReader().getfConsumerFactory().getConsumerFor(topic, consumerGroup, clientId, timeoutMs,
+ remoteHost);
+ CambriaOutboundEventStream coes = new CambriaOutboundEventStream.Builder(consumer).timeout(timeoutMs)
.limit(limit).filter(topicFilter).pretty(pretty).withMeta(withMeta).build();
coes.setDmaapContext(ctx);
- coes.setTopic(metatopic);
- if (isTransEnabled() || isAAFTopic) {
- coes.setTransEnabled(true);
- } else {
- coes.setTransEnabled(false);
- }
+ coes.setTopic(metaTopic);
+ coes.setTransEnabled(isTransEnabled() || isAAFTopic);
coes.setTopicStyle(isAAFTopic);
final long elapsedMs2 = System.currentTimeMillis() - startTime;
logger.info("Time taken in getEvents getConsumerFor " + elapsedMs2 + " ms for " + topic + " "
+ consumerGroup + " " + clientId);
- DMaaPResponseBuilder.setNoCacheHeadings(ctx);
-
- DMaaPResponseBuilder.respondOkWithStream(ctx, MediaType.APPLICATION_JSON, coes);
+ respondOkWithStream(ctx, coes);
// No IOException thrown during respondOkWithStream, so commit the
// new offsets to all the brokers
- c.commitOffsets();
+ consumer.commitOffsets();
final int sent = coes.getSentCount();
-
- metricsSet.consumeTick(sent);
- rl.onSend(topic, consumerGroup, clientId, sent);
+ metricsSet.consumeTick(sent);
+ rl.onSend(topic, consumerGroup, clientId, sent);
final long elapsedMs = System.currentTimeMillis() - startTime;
- logger.info("Sent " + sent + " msgs in " + elapsedMs + " ms; committed to offset " + c.getOffset() + " for "
+ logger.info("Sent " + sent + " msgs in " + elapsedMs + " ms; committed to offset " + consumer.getOffset() + " for "
+ topic + " " + consumerGroup + " " + clientId + " on to the server "
- + ctx.getRequest().getRemoteHost());
+ + remoteHost);
} catch (UnavailableException excp) {
logger.warn(excp.getMessage(), excp);
-
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
- DMaaPResponseCode.SERVER_UNAVAILABLE.getResponseCode(),
- errorMessages.getServerUnav() + excp.getMessage(), null, Utils.getFormattedDate(new Date()), topic,
- null, null, consumerGroup + "-" + clientId, ctx.getRequest().getRemoteHost());
+ ErrorResponse errRes = errRespProvider.getServiceUnavailableError(excp.getMessage());
LOG.info(errRes.toString());
throw new CambriaApiException(errRes);
- } catch (java.util.ConcurrentModificationException excp1) {
- LOG.info(excp1.getMessage() + "on " + topic + " " + consumerGroup + " ****** " + clientId + " from Remote"+ctx.getRequest().getRemoteHost());
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_CONFLICT,
- DMaaPResponseCode.TOO_MANY_REQUESTS.getResponseCode(),
- "Couldn't respond to client, possible of consumer requests from more than one server. Please contact MR team if you see this issue occurs continously", null,
- Utils.getFormattedDate(new Date()), topic, null, null, clientId, ctx.getRequest().getRemoteHost());
+ } catch (ConcurrentModificationException excp1) {
+ LOG.info(excp1.getMessage() + "on " + topic + " " + consumerGroup + " ****** " + clientId + " from Remote"+remoteHost);
+ ErrorResponse errRes = errRespProvider.getConcurrentModificationError();
logger.info(errRes.toString());
throw new CambriaApiException(errRes);
- } catch (CambriaApiException excp) {
- LOG.info(excp.getMessage() + "on " + topic + " " + consumerGroup + " ****** " + clientId);
-
- throw excp;
- }
- catch (Exception excp) {
- // System.out.println(excp + "------------------ " + topic+"
- // "+consumerGroup+" "+clientId);
-
+ } catch (Exception excp) {
logger.info("Couldn't respond to client, closing cambria consumer " + " " + topic + " " + consumerGroup
+ " " + clientId + " " + HttpStatus.SC_SERVICE_UNAVAILABLE + " ****** " + excp);
-
- ctx.getConfigReader().getfConsumerFactory().destroyConsumer(topic, consumerGroup, clientId);
-
-
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_SERVICE_UNAVAILABLE,
- DMaaPResponseCode.SERVER_UNAVAILABLE.getResponseCode(),
- "Couldn't respond to client, closing cambria consumer" + excp.getMessage(), null,
- Utils.getFormattedDate(new Date()), topic, null, null, clientId, ctx.getRequest().getRemoteHost());
+ ctx.getConfigReader().getfConsumerFactory().destroyConsumer(topic, consumerGroup, clientId);
+ ErrorResponse errRes = errRespProvider.getGenericError(excp.getMessage());
logger.info(errRes.toString());
throw new CambriaApiException(errRes);
} finally {
- coes = null;
- // If no cache, close the consumer now that we're done with it.
- boolean kSetting_EnableCache = ConsumerFactory.kDefault_IsCacheEnabled;
- String strkSetting_EnableCache = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- ConsumerFactory.kSetting_EnableCache);
- if (null != strkSetting_EnableCache)
- kSetting_EnableCache = Boolean.parseBoolean(strkSetting_EnableCache);
- // if
- // (!ctx.getConfigReader().getSettings().getBoolean(ConsumerFactory.kSetting_EnableCache,
- // ConsumerFactory.kDefault_IsCacheEnabled) && (c != null)) {
- if (!kSetting_EnableCache && (c != null)) {
+ if (consumer != null && !isCacheEnabled()) {
try {
- c.close();
+ consumer.close();
} catch (Exception e) {
- logger.info("***Exception occured in getEvents finaly block while closing the consumer " + " "
+ logger.info("***Exception occurred in getEvents finally block while closing the consumer " + " "
+ topic + " " + consumerGroup + " " + clientId + " " + HttpStatus.SC_SERVICE_UNAVAILABLE
+ " " + e);
}
@@ -352,117 +212,155 @@ public class EventsServiceImpl implements EventsService {
}
}
- /**
- * @throws missingReqdSetting
- *
- */
- @Override
- public void pushEvents(DMaaPContext ctx, final String topic, InputStream msg, final String defaultPartition,
- final String requestTime) throws ConfigDbException, AccessDeniedException, TopicExistsException,
- CambriaApiException, IOException, missingReqdSetting, DMaaPAccessDeniedException {
-
- // is this user allowed to write to this topic?
- final long startMs = System.currentTimeMillis();
- final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(ctx);
- final Topic metatopic = ctx.getConfigReader().getfMetaBroker().getTopic(topic);
- boolean isAAFTopic = false;
-
- // was this host blacklisted?
+ private void validateIpBlacklist(ErrorResponseProvider errResponseProvider, DMaaPContext ctx) throws CambriaApiException {
final String remoteAddr = Utils.getRemoteAddress(ctx);
-
if (ctx.getConfigReader().getfIpBlackList().contains(remoteAddr)) {
-
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_FORBIDDEN,
- DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
- "Source address [" + remoteAddr + "] is blacklisted. Please contact the cluster management team.",
- null, Utils.getFormattedDate(new Date()), topic, Utils.getUserApiKey(ctx.getRequest()),
- ctx.getRequest().getRemoteHost(), null, null);
+ ErrorResponse errRes = errResponseProvider.getIpBlacklistedError(remoteAddr);
LOG.info(errRes.toString());
throw new CambriaApiException(errRes);
}
+ }
- String topicNameStd = null;
+ private boolean authorizeClientWhenNeeded(DMaaPContext ctx, Topic metaTopic, String topicName,
+ ErrorResponseProvider errRespProvider, String action) throws CambriaApiException, AccessDeniedException {
- // topicNameStd=
-
- topicNameStd = com.att.ajsc.beans.PropertiesMapBean.getProperty(CambriaConstants.msgRtr_prop,
- "enforced.topic.name.AAF");
- String metricTopicname = com.att.ajsc.filemonitor.AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "metrics.send.cambria.topic");
- if (null == metricTopicname)
- metricTopicname = "msgrtr.apinode.metrics.dmaap";
- boolean topicNameEnforced = false;
- if (null != topicNameStd && topic.startsWith(topicNameStd)) {
- topicNameEnforced = true;
+ boolean isAAFTopic = false;
+ String metricTopicName = getMetricTopicName();
+ if(!metricTopicName.equalsIgnoreCase(topicName)) {
+ if(isCadiEnabled() && isTopicNameEnforcedAaf(topicName)) {
+ isAAFTopic = true;
+ DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
+ String permission = aaf.aafPermissionString(topicName, action);
+ if (!aaf.aafAuthentication(ctx.getRequest(), permission)) {
+ ErrorResponse errRes = errRespProvider.getAafAuthorizationError(permission, action);
+ LOG.info(errRes.toString());
+ throw new DMaaPAccessDeniedException(errRes);
+
+ }
+ } else if( null != metaTopic.getOwner() && !metaTopic.getOwner().isEmpty()) {
+ final NsaApiKey user = DMaaPAuthenticatorImpl.getAuthenticatedUser(ctx);
+ if(SUBSCRIBE_ACTION.equals(action)) {
+ metaTopic.checkUserRead(user);
+ } else if(PUBLISH_ACTION.equals(action)) {
+ metaTopic.checkUserWrite(user);
+ }
+ }
}
+ return isAAFTopic;
+ }
+
+ boolean isCadiEnabled() {
+ return Utils.isCadiEnabled();
+ }
- // Here check if the user has rights to publish on the topic
- // ( This will be called when no auth is added or when UEB API Key
- // Authentication is used)
- // checkUserWrite(user) method will throw an error when there is no Auth
- // header added or when the
- // user has no publish rights
+ void respondOkWithStream(DMaaPContext ctx, StreamWriter coes) throws IOException{
+ DMaaPResponseBuilder.setNoCacheHeadings(ctx);
+ DMaaPResponseBuilder.respondOkWithStream(ctx, MediaType.APPLICATION_JSON, coes);
+ }
- if (null != metatopic && null != metatopic.getOwner() && !("".equals(metatopic.getOwner()))
- && null == ctx.getRequest().getHeader("Authorization") && !topic.equalsIgnoreCase(metricTopicname)) {
- metatopic.checkUserWrite(user);
- }
+ private int getMessageLimit(HttpServletRequest request) {
+ return NumberUtils.toInt(request.getParameter("limit"), CambriaConstants.kNoLimit);
+ }
- // if headers are not provided then user will be null
- if (topicNameEnforced || (user == null && null != ctx.getRequest().getHeader("Authorization")
- && !topic.equalsIgnoreCase(metricTopicname))) {
- // the topic name will be sent by the client
-
- DMaaPAAFAuthenticator aaf = new DMaaPAAFAuthenticatorImpl();
- String permission = aaf.aafPermissionString(topic, "pub");
- if (!aaf.aafAuthentication(ctx.getRequest(), permission)) {
- ErrorResponse errRes = new ErrorResponse(HttpStatus.SC_UNAUTHORIZED,
- DMaaPResponseCode.ACCESS_NOT_PERMITTED.getResponseCode(),
- errorMessages.getNotPermitted1() + " publish " + errorMessages.getNotPermitted2() + topic
- + " on " + permission,
- null, Utils.getFormattedDate(new Date()), topic, Utils.getUserApiKey(ctx.getRequest()),
- ctx.getRequest().getRemoteHost(), null, null);
- LOG.info(errRes.toString());
- throw new DMaaPAccessDeniedException(errRes);
+ private int getMessageTimeout(HttpServletRequest request) {
+ String timeoutMsAsString = getPropertyFromAJSCmap(TIMEOUT_PROPERTY);
+ int defaultTimeoutMs = timeoutMsAsString!=null ? NumberUtils.toInt(timeoutMsAsString, CambriaConstants.kNoTimeout) :
+ CambriaConstants.kNoTimeout;
+
+ String timeoutProperty = request.getParameter(TIMEOUT_PROPERTY);
+ return timeoutProperty != null ? NumberUtils.toInt(timeoutProperty, defaultTimeoutMs) : defaultTimeoutMs;
+ }
+
+ private boolean isPrettyPrintEnabled() {
+ return rrConvertor.convertToBooleanBroad(getPropertyFromAJSCmap("pretty"));
+ }
+
+ private boolean isMetaOffsetEnabled() {
+ return rrConvertor.convertToBooleanBroad(getPropertyFromAJSCmap( "meta"));
+ }
+
+ private boolean isTopicNameEnforcedAaf(String topicName) {
+ String topicNameStd = getPropertyFromAJSCmap("enforced.topic.name.AAF");
+ return !topicNameStd.isEmpty() && topicName.startsWith(topicNameStd);
+ }
+
+ private boolean isCacheEnabled() {
+ String cachePropsSetting = getPropertyFromAJSCmap(ConsumerFactory.kSetting_EnableCache);
+ return !cachePropsSetting.isEmpty() ? Boolean.parseBoolean(cachePropsSetting) : ConsumerFactory.kDefault_IsCacheEnabled;
+ }
+
+ private void verifyHostId() {
+ String lhostId = getPropertyFromAJSCmap("clusterhostid");
+ if (lhostId.isEmpty()) {
+ try {
+ InetAddress.getLocalHost().getCanonicalHostName();
+ } catch (UnknownHostException e) {
+ LOG.warn("Unknown Host Exception error occurred while getting getting hostid", e);
}
- isAAFTopic = true;
+
}
+ }
- final HttpServletRequest req = ctx.getRequest();
+ private String getMetricTopicName() {
+ String metricTopicFromProps = getPropertyFromAJSCmap("metrics.send.cambria.topic");
+ return !metricTopicFromProps.isEmpty() ? metricTopicFromProps : "msgrtr.apinode.metrics.dmaap";
+ }
- // check for chunked input
- boolean chunked = false;
- if (null != req.getHeader(TRANSFER_ENCODING)) {
- chunked = req.getHeader(TRANSFER_ENCODING).contains("chunked");
- }
- // get the media type, or set it to a generic value if it wasn't
- // provided
- String mediaType = req.getContentType();
- if (mediaType == null || mediaType.length() == 0) {
- mediaType = MimeTypes.kAppGenericBinary;
- }
+ /**
+ * @throws missingReqdSetting
+ *
+ */
+ @Override
+ public void pushEvents(DMaaPContext ctx, final String topic, InputStream msg, final String defaultPartition,
+ final String requestTime) throws ConfigDbException, AccessDeniedException,
+ CambriaApiException, IOException, missingReqdSetting {
- if (mediaType.contains("charset=UTF-8")) {
- mediaType = mediaType.replace("; charset=UTF-8", "").trim();
- }
+ final long startMs = System.currentTimeMillis();
+ String remoteHost = ctx.getRequest().getRemoteHost();
+ ErrorResponseProvider errRespProvider = new ErrorResponseProvider.Builder().withErrorMessages(errorMessages)
+ .withTopic(topic).withRemoteHost(remoteHost).withPublisherIp(remoteHost)
+ .withPublisherId(Utils.getUserApiKey(ctx.getRequest())).build();
- String istransidUEBtopicreqd = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "transidUEBtopicreqd");
- boolean istransidreqd = false;
- if (null != istransidUEBtopicreqd && istransidUEBtopicreqd.equalsIgnoreCase("true")) {
- istransidreqd = true;
+ validateIpBlacklist(errRespProvider, ctx);
+
+ final Topic metaTopic = ctx.getConfigReader().getfMetaBroker().getTopic(topic);
+ if (metaTopic == null) {
+ throw new CambriaApiException(errRespProvider.getTopicNotFoundError());
}
- if (isAAFTopic || istransidreqd) {
+ final boolean isAAFTopic = authorizeClientWhenNeeded(ctx, metaTopic, topic, errRespProvider, PUBLISH_ACTION);
+
+ final HttpServletRequest req = ctx.getRequest();
+ boolean chunked = isRequestedChunk(req);
+ String mediaType = getMediaType(req);
+ boolean transactionRequired = isTransactionIdRequired();
+
+ if (isAAFTopic || transactionRequired) {
pushEventsWithTransaction(ctx, msg, topic, defaultPartition, requestTime, chunked, mediaType);
} else {
pushEvents(ctx, topic, msg, defaultPartition, chunked, mediaType);
}
+
final long endMs = System.currentTimeMillis();
final long totalMs = endMs - startMs;
-
LOG.info("Overall Response time - Published " + " msgs in " + totalMs + " ms for topic " + topic);
+ }
+
+ private boolean isRequestedChunk(HttpServletRequest request) {
+ return null != request.getHeader(TRANSFER_ENCODING) &&
+ request.getHeader(TRANSFER_ENCODING).contains("chunked");
+ }
+
+ private String getMediaType(HttpServletRequest request) {
+ String mediaType = request.getContentType();
+ if (mediaType == null || mediaType.length() == 0) {
+ return MimeTypes.kAppGenericBinary;
+ }
+ return mediaType.replace("; charset=UTF-8", "").trim();
+ }
+ private boolean isTransactionIdRequired() {
+ return getPropertyFromAJSCmap("transidUEBtopicreqd").equalsIgnoreCase("true");
}
/**
@@ -481,7 +379,7 @@ public class EventsServiceImpl implements EventsService {
*/
private void pushEvents(DMaaPContext ctx, String topic, InputStream msg, String defaultPartition, boolean chunked,
String mediaType)
- throws ConfigDbException, AccessDeniedException, TopicExistsException, CambriaApiException, IOException {
+ throws ConfigDbException, AccessDeniedException, CambriaApiException, IOException {
final MetricsSet metricsSet = ctx.getConfigReader().getfMetrics();
// setup the event set
final CambriaEventSet events = new CambriaEventSet(mediaType, msg, chunked, defaultPartition);
@@ -490,8 +388,8 @@ public class EventsServiceImpl implements EventsService {
final long startMs = System.currentTimeMillis();
long count = 0;
long maxEventBatch = 1024L* 16;
- String batchlen = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, BATCH_LENGTH);
- if (null != batchlen)
+ String batchlen = getPropertyFromAJSCmap( BATCH_LENGTH);
+ if (null != batchlen && !batchlen.isEmpty())
maxEventBatch = Long.parseLong(batchlen);
// long maxEventBatch =
@@ -550,7 +448,7 @@ public class EventsServiceImpl implements EventsService {
final JSONObject response = new JSONObject();
response.put("count", count);
response.put("serverTimeMs", totalMs);
- DMaaPResponseBuilder.respondOk(ctx, response);
+ respondOk(ctx, response);
} catch (Exception excp) {
int status = HttpStatus.SC_NOT_FOUND;
@@ -590,7 +488,7 @@ public class EventsServiceImpl implements EventsService {
*/
private void pushEventsWithTransaction(DMaaPContext ctx, InputStream inputStream, final String topic,
final String partitionKey, final String requestTime, final boolean chunked, final String mediaType)
- throws ConfigDbException, AccessDeniedException, TopicExistsException, IOException, CambriaApiException {
+ throws ConfigDbException, AccessDeniedException, IOException, CambriaApiException {
final MetricsSet metricsSet = ctx.getConfigReader().getfMetrics();
@@ -601,8 +499,8 @@ public class EventsServiceImpl implements EventsService {
final long startMs = System.currentTimeMillis();
long count = 0;
long maxEventBatch = 1024L * 16;
- String evenlen = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, BATCH_LENGTH);
- if (null != evenlen)
+ String evenlen = getPropertyFromAJSCmap( BATCH_LENGTH);
+ if (null != evenlen && !evenlen.isEmpty())
maxEventBatch = Long.parseLong(evenlen);
// final long maxEventBatch =
@@ -639,9 +537,9 @@ public class EventsServiceImpl implements EventsService {
responseTransactionId = m.getLogDetails().getTransactionId();
- JSONObject jsonObject = new JSONObject();
- jsonObject.put("msgWrapMR", m.getMessage());
- jsonObject.put("transactionId", responseTransactionId);
+ //JSONObject jsonObject = new JSONObject();
+ //jsonObject.put("msgWrapMR", m.getMessage());
+ //jsonObject.put("transactionId", responseTransactionId);
// final KeyedMessage<String, String> data = new
// KeyedMessage<String, String>(topic, m.getKey(),
@@ -755,8 +653,9 @@ public class EventsServiceImpl implements EventsService {
// build a response
final JSONObject response = new JSONObject();
response.put("count", count);
+ response.put("transactionId", responseTransactionId);
response.put("serverTimeMs", totalMs);
- DMaaPResponseBuilder.respondOk(ctx, response);
+ respondOk(ctx, response);
} catch (Exception excp) {
int status = HttpStatus.SC_NOT_FOUND;
@@ -798,6 +697,10 @@ public class EventsServiceImpl implements EventsService {
msg.setLogDetails(logDetails);
}
+ void respondOk(DMaaPContext ctx, JSONObject response) throws IOException {
+ DMaaPResponseBuilder.respondOk(ctx, response);
+ }
+
/**
*
* @author anowarul.islam
@@ -837,8 +740,7 @@ public class EventsServiceImpl implements EventsService {
}
public boolean isTransEnabled() {
- String istransidUEBtopicreqd = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
- "transidUEBtopicreqd");
+ String istransidUEBtopicreqd = getPropertyFromAJSCmap("transidUEBtopicreqd");
boolean istransidreqd = false;
if ((null != istransidUEBtopicreqd && istransidUEBtopicreqd.equalsIgnoreCase("true"))) {
istransidreqd = true;
@@ -863,13 +765,5 @@ public class EventsServiceImpl implements EventsService {
return logDetails;
}
-
-
-
-
-
-
-
-
} \ No newline at end of file
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java b/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
index 3048251..40e6840 100644
--- a/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
+++ b/src/main/java/org/onap/dmaap/dmf/mr/utils/Utils.java
@@ -23,6 +23,7 @@ package org.onap.dmaap.dmf.mr.utils;
import java.io.IOException;
import java.io.InputStream;
+import java.security.Principal;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
@@ -46,7 +47,9 @@ public class Utils {
private static final String DATE_FORMAT = "dd-MM-yyyy::hh:mm:ss:SSS";
public static final String CAMBRIA_AUTH_HEADER = "X-CambriaAuth";
+ private static final String AUTH_HEADER = "Authorization";
private static final String BATCH_ID_FORMAT = "000000";
+ private static final String X509_ATTR = "javax.servlet.request.X509Certificate";
private static final EELFLogger log = EELFManager.getInstance().getLogger(Utils.class);
private Utils() {
@@ -75,15 +78,21 @@ public class Utils {
if (null != auth) {
final String[] splittedAuthKey = auth.split(":");
return splittedAuthKey[0];
- }else if (null!=request.getHeader("Authorization")){
+ }else if (null != request.getHeader(AUTH_HEADER) || null != request.getAttribute(X509_ATTR)){
/**
* AAF implementation enhancement
*/
- String user= request.getUserPrincipal().getName().toString();
- return user.substring(0, user.lastIndexOf("@"));
+ Principal principal = request.getUserPrincipal();
+ if(principal != null){
+ String name = principal.getName();
+ return name.substring(0, name.lastIndexOf('@'));
+ }
+ log.warn("No principal has been provided on HTTP request");
}
return null;
}
+
+
/**
* to format the batch sequence id
* @param batchId
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);