Improve JUnit tests for the Datafile Collector.
Add tests of logging.
Remove dependency to old JUnit version.
Improve some log messages.
Improve code coverage.
Change-Id: Ie8204619ab7fe4294c4ac0dd8f030973653ef4a3
Issue-ID: DCAEGEN2-1490
Signed-off-by: elinuxhenrik <henrik.b.andersson@est.tech>
tls_info:
cert_directory: '/opt/app/datafile/etc/cert/'
use_tls: true
+
<groupId>com.jcraft</groupId>
<artifactId>jsch</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-configuration-processor</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <!-- Actuator dependencies -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
- </dependency>
+ </dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
</dependency>
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-configuration-processor</artifactId>
- <optional>true</optional>
- </dependency>
<!--TESTS DEPENDENCIES -->
<dependency>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.github.stefanbirkner</groupId>
- <artifactId>fake-sftp-server-rule</artifactId>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
import com.google.gson.TypeAdapterFactory;
-
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
-
import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;
-
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.stereotype.Component;
-
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
}
/**
- * parse configuration
+ * Parse configuration.
*
- * @param serviceConfigRootObject
- * @param dmaapConfigRootObject if there is no dmaapConfigRootObject, the dmaap feeds are taken
- * from the serviceConfigRootObject
+ * @param serviceConfigRootObject the DFC service's configuration
+ * @param dmaapConfigRootObject if there is no dmaapConfigRootObject, the dmaap feeds are taken from the
+ * serviceConfigRootObject
* @return this which is updated if successful
*/
private AppConfig parseCloudConfig(JsonObject serviceConfigRootObject, JsonObject dmaapConfigRootObject) {
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
-
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
.passWord(getAsString(feedConfig, "password")) //
.userName(getAsString(feedConfig, "username")) //
.trustStorePath(getAsString(serviceConfigurationRoot, DMAAP_SECURITY_TRUST_STORE_PATH)) //
- .trustStorePasswordPath(getAsString(serviceConfigurationRoot, DMAAP_SECURITY_TRUST_STORE_PASS_PATH)) //
+ .trustStorePasswordPath(
+ getAsString(serviceConfigurationRoot, DMAAP_SECURITY_TRUST_STORE_PASS_PATH)) //
.keyStorePath(getAsString(serviceConfigurationRoot, DMAAP_SECURITY_KEY_STORE_PATH)) //
.keyStorePasswordPath(getAsString(serviceConfigurationRoot, DMAAP_SECURITY_KEY_STORE_PASS_PATH)) //
.enableDmaapCertAuth(
private JsonObject getFeedConfig(String feedName) throws DatafileTaskException {
JsonElement elem = dmaapConfigurationRoot.get(feedName);
if (elem == null) {
- elem = get(serviceConfigurationRoot, feedName); // Fallback, try to find it under
- // serviceConfigurationRoot
+ elem = get(serviceConfigurationRoot, feedName); // Fallback, try to find it under serviceConfigurationRoot
}
return elem.getAsJsonObject();
}
import java.net.MalformedURLException;
import java.net.URL;
-
import org.immutables.gson.Gson;
import org.immutables.value.Value;
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
throw new DatafileTaskException("The path has incorrect syntax: " + urlPath);
}
- final String dmaapTopicName = tokens[1] + "/" + tokens[2]; // ex. // /events/unauthenticated.VES_NOTIFICATION_OUTPUT
+ final String dmaapTopicName = tokens[1] + "/" + tokens[2]; // e.g. /events/unauthenticated.VES_NOTIFICATION_OUTPUT
final String consumerGroup = tokens[3]; // ex. OpenDcae-c12
final String consumerId = tokens[4]; // ex. C12
return new DmaapConsumerUrlPath(dmaapTopicName, consumerGroup, consumerId);
package org.onap.dcaegen2.collectors.datafile.configuration;
+import io.swagger.annotations.ApiOperation;
import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
-
import javax.annotation.PostConstruct;
-
import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
import org.onap.dcaegen2.collectors.datafile.tasks.ScheduledTasks;
import org.slf4j.Logger;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
-
-import io.swagger.annotations.ApiOperation;
import reactor.core.publisher.Mono;
/**
public enum Scheme {
FTPS, SFTP;
+ public static final String DFC_DOES_NOT_SUPPORT_PROTOCOL_ERROR_MSG = "DFC does not support protocol ";
+ public static final String SUPPORTED_PROTOCOLS_ERROR_MESSAGE =
+ ". Supported protocols are FTPES, FTPS, and SFTP";
+
/**
* Get a <code>Scheme</code> from a string.
*
} else if ("SFTP".equalsIgnoreCase(schemeString)) {
result = Scheme.SFTP;
} else {
- throw new DatafileTaskException(
- "DFC does not support protocol " + schemeString + ". Supported protocols are FTPES , FTPS, and SFTP");
+ throw new DatafileTaskException(DFC_DOES_NOT_SUPPORT_PROTOCOL_ERROR_MSG + schemeString
+ + SUPPORTED_PROTOCOLS_ERROR_MESSAGE);
}
return result;
}
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.SftpException;
-
import java.nio.file.Path;
import java.util.Optional;
-
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SftpClient implements FileCollectClient {
private static final Logger logger = LoggerFactory.getLogger(SftpClient.class);
- private static final int FTPS_DEFAULT_PORT = 22;
+ private static final int SFTP_DEFAULT_PORT = 22;
private final FileServerData fileServerData;
- private Session session = null;
- private ChannelSftp sftpChannel = null;
+ protected Session session = null;
+ protected ChannelSftp sftpChannel = null;
public SftpClient(FileServerData fileServerData) {
this.fileServerData = fileServerData;
sftpChannel.get(remoteFile, localFile.toString());
logger.trace("File {} Download Successfull from xNF", localFile.getFileName());
} catch (SftpException e) {
- boolean retry = e.id != ChannelSftp.SSH_FX_NO_SUCH_FILE && e.id != ChannelSftp.SSH_FX_PERMISSION_DENIED && e.id != ChannelSftp.SSH_FX_OP_UNSUPPORTED;
+ boolean retry = e.id != ChannelSftp.SSH_FX_NO_SUCH_FILE && e.id != ChannelSftp.SSH_FX_PERMISSION_DENIED
+ && e.id != ChannelSftp.SSH_FX_OP_UNSUPPORTED;
throw new DatafileTaskException("Unable to get file from xNF. Data: " + fileServerData, e, retry);
}
}
} catch (JSchException e) {
boolean retry = !e.getMessage().contains("Auth fail");
- throw new DatafileTaskException("Could not open Sftp client" + e, e, retry);
+ throw new DatafileTaskException("Could not open Sftp client. " + e, e, retry);
}
}
- private static int getPort(Optional<Integer> port) {
- return port.isPresent() ? port.get() : FTPS_DEFAULT_PORT;
+ private int getPort(Optional<Integer> port) {
+ return port.isPresent() ? port.get() : SFTP_DEFAULT_PORT;
}
- private static Session setUpSession(FileServerData fileServerData) throws JSchException {
- JSch jsch = new JSch();
+ private Session setUpSession(FileServerData fileServerData) throws JSchException {
+ JSch jsch = createJsch();
- Session newSession =
- jsch.getSession(fileServerData.userId(), fileServerData.serverAddress(), getPort(fileServerData.port()));
+ Session newSession = jsch.getSession(fileServerData.userId(), fileServerData.serverAddress(),
+ getPort(fileServerData.port()));
newSession.setConfig("StrictHostKeyChecking", "no");
newSession.setPassword(fileServerData.password());
newSession.connect();
return newSession;
}
- private static ChannelSftp getChannel(Session session) throws JSchException {
+ private ChannelSftp getChannel(Session session) throws JSchException {
Channel channel = session.openChannel("sftp");
channel.connect();
return (ChannelSftp) channel;
}
+
+ protected JSch createJsch() {
+ return new JSch();
+ }
}
import com.google.gson.FieldAttributes;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
-
import java.util.Set;
/**
*/
public abstract class JsonSerializer {
- private JsonSerializer() {}
-
private static Gson gson = new GsonBuilder() //
.serializeNulls() //
.addSerializationExclusionStrategy(new FilePublishInformationExclusionStrategy()) //
.create(); //
+ private JsonSerializer() {}
+
/**
* Serializes a <code>filePublishInformation</code>.
*
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
-
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.StreamSupport;
-
import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
import org.onap.dcaegen2.collectors.datafile.model.FileData;
import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
-
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class JsonMessageParser {
private static final Logger logger = LoggerFactory.getLogger(JsonMessageParser.class);
+ public static final String ERROR_MSG_VES_EVENT_PARSING = "VES event parsing. ";
+
private static final String COMMON_EVENT_HEADER = "commonEventHeader";
private static final String EVENT_NAME = "eventName";
private static final String LAST_EPOCH_MICROSEC = "lastEpochMicrosec";
}
}
- logger.error("VES event parsing. Missing arrayOfNamedHashMap in message. {}", message);
+ logger.error(ERROR_MSG_VES_EVENT_PARSING + "Missing arrayOfNamedHashMap in message. {}", message);
return Mono.empty();
}
- logger.error("VES event parsing. FileReady event has incorrect JsonObject. {}", message);
+ logger.error(ERROR_MSG_VES_EVENT_PARSING + "FileReady event has incorrect JsonObject. {}", message);
return Mono.empty();
}
if (missingValues.isEmpty() && isChangeTypeCorrect(changeType)) {
return Optional.of(messageMetaData);
} else {
- String errorMessage = "VES event parsing.";
+ String errorMessage = ERROR_MSG_VES_EVENT_PARSING;
if (!missingValues.isEmpty()) {
- errorMessage += " Missing data: " + missingValues;
+ errorMessage += "Missing data: " + missingValues + ".";
}
if (!isChangeTypeCorrect(changeType)) {
- errorMessage += " Change type is wrong: " + changeType + " expected: " + FILE_READY_CHANGE_TYPE;
+ errorMessage += " Change type is wrong: " + changeType + " Expected: " + FILE_READY_CHANGE_TYPE;
}
errorMessage += " Message: {}";
logger.error(errorMessage, message);
JsonObject data = fileInfo.getAsJsonObject(HASH_MAP);
String location = getValueFromJson(data, LOCATION, missingValues);
+ if (StringUtils.isEmpty(location)) {
+ logger.error(ERROR_MSG_VES_EVENT_PARSING + "File information wrong. Missing location. Data: {} {}",
+ messageMetaData, fileInfo);
+ return Optional.empty();
+ }
Scheme scheme;
try {
scheme = Scheme.getSchemeFromString(URI.create(location).getScheme());
} catch (Exception e) {
- logger.error("VES event parsing.", e);
+ logger.error(ERROR_MSG_VES_EVENT_PARSING + "{}. Location: {} Data: {}", e.getMessage(), location,
+ messageMetaData, e);
return Optional.empty();
}
FileData fileData = ImmutableFileData.builder() //
if (missingValues.isEmpty()) {
return Optional.of(fileData);
}
- logger.error("VES event parsing. File information wrong. Missing data: {} Data: {}", missingValues, fileInfo);
+ logger.error(ERROR_MSG_VES_EVENT_PARSING + "File information wrong. Missing data: {} Data: {}", missingValues,
+ fileInfo);
return Optional.empty();
}
/**
- * Gets data from the event name. Defined as:
- * {DomainAbbreviation}_{productName}-{vendorName}_{Description}, example:
- * Noti_RnNode-Ericsson_FileReady
+ * Gets data from the event name. Defined as: {DomainAbbreviation}_{productName}-{vendorName}_{Description},
+ * example: Noti_RnNode-Ericsson_FileReady
*
* @param dataType The type of data to get, {@link DmaapConsumerJsonParser.EventNameDataType}.
* @param eventName The event name to get the data from.
return eventArray[dataType.index];
} else {
missingValues.add(dataType.toString());
- logger.error("Can not get {} from eventName, eventName is not in correct format: {}", dataType, eventName);
+ logger.error(
+ ERROR_MSG_VES_EVENT_PARSING
+ + "Can not get {} from eventName, eventName is not in correct format: {}",
+ dataType, eventName);
}
return "";
}
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
-
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URI;
import java.nio.file.Path;
import java.time.Duration;
-
import org.apache.commons.io.IOUtils;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.web.util.DefaultUriBuilderFactory;
-
import reactor.core.publisher.Mono;
/**
logger.trace("{}", response);
return Mono.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode()));
} catch (Exception e) {
- logger.warn("Unable to send file to DataRouter. Data: {}", publishInfo.getInternalLocation(), e);
+ logger.warn("Publishing file {} to DR unsuccessful.", publishInfo.getName(), e);
return Mono.error(e);
}
}
put.addHeader(X_DMAAP_DR_META, metaData.toString());
URI uri = new DefaultUriBuilderFactory(
datafileAppConfig.getPublisherConfiguration(publishInfo.getChangeIdentifier()).publishUrl()) //
- .builder() //
- .pathSegment(publishInfo.getName()) //
- .build();
+ .builder() //
+ .pathSegment(publishInfo.getName()) //
+ .build();
put.setURI(uri);
MappedDiagnosticContext.appendTraceInfo(put);
}
}
- private static Mono<FilePublishInformation> handleHttpResponse(HttpStatus response, FilePublishInformation publishInfo) {
+ private static Mono<FilePublishInformation> handleHttpResponse(HttpStatus response,
+ FilePublishInformation publishInfo) {
MDC.setContextMap(publishInfo.getContext());
if (HttpUtils.isSuccessfulResponseCode(response.value())) {
- logger.trace("Publish to DR successful!");
+ logger.trace("Publishing file {} to DR successful!", publishInfo.getName());
return Mono.just(publishInfo);
} else {
- logger.warn("Publish to DR unsuccessful, response code: {}", response);
- return Mono.error(new Exception("Publish to DR unsuccessful, response code: " + response));
+ logger.warn("Publishing file {} to DR unsuccessful. Response code: {}", publishInfo.getName(), response);
+ return Mono.error(new Exception(
+ "Publishing file " + publishInfo.getName() + " to DR unsuccessful. Response code: " + response));
}
}
import java.time.Duration;
import java.util.Map;
import java.util.Optional;
-
import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
import org.onap.dcaegen2.collectors.datafile.configuration.FtpesConfig;
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
-
import reactor.core.publisher.Mono;
/**
import java.time.Instant;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
-
import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
import org.onap.dcaegen2.collectors.datafile.model.FileData;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
/**
- * This implements the main flow of the data file collector. Fetch file ready events from the
- * message router, fetch new files from the PNF publish these in the data router.
+ * This implements the main flow of the data file collector. Fetch file ready events from the message router, fetch new
+ * files from the PNF publish these in the data router.
*/
@Component
public class ScheduledTasks {
try {
if (getCurrentNumberOfTasks() > MAX_TASKS_FOR_POLLING || this.threadPoolQueueSize.get() > 0) {
logger.info(
- "Skipping consuming new files; current number of tasks: {}, number of subscriptions: {}, published files: {}, number of queued VES events: {}",
+ "Skipping consuming new files; current number of tasks: {}, number of subscriptions: {}, "
+ + "published files: {}, number of queued VES events: {}",
getCurrentNumberOfTasks(), this.currentNumberOfSubscriptions.get(), publishedFilesCache.size(),
threadPoolQueueSize.get());
return;
currentNumberOfSubscriptions.decrementAndGet();
});
} catch (Exception e) {
- logger.error("Unexpected exception: ", e);
+ logger.error("Unexpected exception: {}", e.toString(), e);
}
}
/**
* Fetch more messages from the message router. This is done in a polling/blocking fashion.
*/
- private Flux<FileReadyMessage> fetchMoreFileReadyMessages() {
+ Flux<FileReadyMessage> fetchMoreFileReadyMessages() {
logger.info(
"Consuming new file ready messages, current number of tasks: {}, published files: {}, "
+ "number of subscriptions: {}",
-/*
+/*-
* ============LICENSE_START======================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
+ * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. 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
import org.slf4j.MDC;
/**
- * PublishRedirectStrategy implementation
- * that automatically redirects all HEAD, GET, POST, PUT, and DELETE requests.
- * This strategy relaxes restrictions on automatic redirection of
- * POST methods imposed by the HTTP specification.
+ * PublishRedirectStrategy implementation that automatically redirects all HEAD, GET, POST, PUT, and DELETE requests.
+ * This strategy relaxes restrictions on automatic redirection of POST methods imposed by the HTTP specification.
*
*/
@Contract(threading = ThreadingBehavior.IMMUTABLE)
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Map<String, String> contextMap;
+ /**
+ * Redirectable methods.
+ */
+ private static final String[] REDIRECT_METHODS = new String[] { //
+ HttpPut.METHOD_NAME, //
+ HttpGet.METHOD_NAME, //
+ HttpPost.METHOD_NAME, //
+ HttpHead.METHOD_NAME, //
+ HttpDelete.METHOD_NAME //
+ };
+
/**
* Constructor PublishRedirectStrategy.
*
this.contextMap = contextMap;
}
- /**
- * Redirectable methods.
- */
- private static final String[] REDIRECT_METHODS = new String[] { //
- HttpPut.METHOD_NAME, //
- HttpGet.METHOD_NAME, //
- HttpPost.METHOD_NAME, //
- HttpHead.METHOD_NAME, //
- HttpDelete.METHOD_NAME //
- };
-
@Override
protected boolean isRedirectable(final String method) {
for (final String m : REDIRECT_METHODS) {
@Override
public HttpUriRequest getRedirect(final HttpRequest request, final HttpResponse response, final HttpContext context)
- throws ProtocolException {
+ throws ProtocolException {
MDC.setContextMap(contextMap);
final URI uri = getLocationURI(request, response, context);
logger.trace("getRedirect...: {}", request);
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import com.google.common.base.Charsets;
import com.google.common.io.Resources;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonSyntaxException;
-
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Properties;
-
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapConsumerConfiguration;
import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapPublisherConfiguration;
-
-import ch.qos.logback.classic.spi.ILoggingEvent;
-import ch.qos.logback.core.read.ListAppender;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
// Given
appConfigUnderTest.setFilepath("/temp.json");
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
+
// When
appConfigUnderTest.loadConfigurationFromFile();
// Then
+ assertTrue("Error message missing in log.",
+ logAppender.list.toString().contains("[WARN] Local configuration file not loaded: /temp.json"));
+ logAppender.stop();
+
Assertions.assertNull(appConfigUnderTest.getDmaapConsumerConfiguration());
assertThatThrownBy(() -> appConfigUnderTest.getPublisherConfiguration(CHANGE_IDENTIFIER))
.hasMessageContaining("No PublishingConfiguration loaded, changeIdentifier: PM_MEAS_FILES");
-
Assertions.assertNull(appConfigUnderTest.getFtpesConfiguration());
}
@Test
public void whenPeriodicConfigRefreshNoConsul() {
- ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
doReturn(Mono.just(properties())).when(appConfigUnderTest).readEnvironmentVariables(any(), any());
Mono<JsonObject> err = Mono.error(new IOException());
doReturn(err).when(cloudConfigurationProvider).callForServiceConfigurationReactive(any());
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
Flux<AppConfig> task = appConfigUnderTest.createRefreshConfigurationTask(1L, context);
StepVerifier //
doReturn(Mono.just(properties())).when(appConfigUnderTest).readEnvironmentVariables(any(), any());
Mono<JsonObject> json = Mono.just(getJsonRootObject());
- Mono<JsonObject> err = Mono.error(new IOException()); // no config entry created by the
- // dmaap plugin
+ Mono<JsonObject> err = Mono.error(new IOException()); // no config entry created by the dmaap plugin
doReturn(json, err).when(cloudConfigurationProvider).callForServiceConfigurationReactive(any());
package org.onap.dcaegen2.collectors.datafile.controller;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
assertEquals(logAppender.list.get(0).getMarker().getName(), "ENTRY");
assertNotNull(logAppender.list.get(0).getMDCPropertyMap().get("InvocationID"));
assertNotNull(logAppender.list.get(0).getMDCPropertyMap().get("RequestID"));
- assertEquals("[INFO] Heartbeat request", logAppender.list.get(0).toString());
+ assertTrue("Info missing in log", logAppender.list.toString().contains("[INFO] Heartbeat request"));
assertEquals(logAppender.list.get(1).getMarker().getName(), "EXIT");
- assertEquals("[INFO] Heartbeat request", logAppender.list.get(1).toString());
logAppender.stop();
}
}
-/*
+/*-
* ============LICENSE_START======================================================================
* Copyright (C) 2018-2019 Nordix Foundation. 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
+ * 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.
+ * 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.dcaegen2.collectors.datafile.ftp;
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.commons.io.IOUtils.toByteArray;
-import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
-import com.github.stefanbirkner.fakesftpserver.rule.FakeSftpServerRule;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.SftpException;
import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
import java.nio.file.Paths;
-import org.junit.Rule;
-import org.junit.Test;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
+@ExtendWith(MockitoExtension.class)
public class SftpClientTest {
+ private static final String HOST = "127.0.0.1";
+ private static final int SFTP_PORT = 1021;
private static final String USERNAME = "bob";
private static final String PASSWORD = "123";
- private static final String DUMMY_CONTENT = "dummy content";
- private static final Path LOCAL_DUMMY_FILE = Paths.get("target/dummy.txt");
- private static final String REMOTE_DUMMY_FILE = "/dummy_directory/dummy_file.txt";
- private static final JSch JSCH = new JSch();
- private static final int TIMEOUT = 2000;
- @Rule
- public final FakeSftpServerRule sftpServer = new FakeSftpServerRule().addUser(USERNAME, PASSWORD);
+ @Mock
+ private JSch jschMock;
+
+ @Mock
+ private Session sessionMock;
+
+ @Mock
+ private ChannelSftp channelMock;
@Test
- public void collectFile_withOKresponse()
- throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
+ public void openWithPort_success()
+ throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
- .serverAddress("127.0.0.1") //
+ .serverAddress(HOST) //
.userId(USERNAME) //
.password(PASSWORD) //
- .port(sftpServer.getPort()) //
+ .port(SFTP_PORT) //
.build();
- try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
- sftpClient.open();
- sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
- byte[] file = downloadFile(sftpServer, REMOTE_DUMMY_FILE);
-
- sftpClient.collectFile(REMOTE_DUMMY_FILE, LOCAL_DUMMY_FILE);
- byte[] localFile = Files.readAllBytes(LOCAL_DUMMY_FILE.toFile().toPath());
- assertThat(new String(file, UTF_8)).isEqualTo(DUMMY_CONTENT);
- assertThat(new String(localFile, UTF_8)).isEqualTo(DUMMY_CONTENT);
- }
+
+ SftpClient sftpClientSpy = spy(new SftpClient(expectedFileServerData));
+
+ doReturn(jschMock).when(sftpClientSpy).createJsch();
+ when(jschMock.getSession(anyString(), anyString(), anyInt())).thenReturn(sessionMock);
+ when(sessionMock.openChannel(anyString())).thenReturn(channelMock);
+
+ sftpClientSpy.open();
+
+ verify(jschMock).getSession(USERNAME, HOST, SFTP_PORT);
+ verify(sessionMock).setConfig("StrictHostKeyChecking", "no");
+ verify(sessionMock).setPassword(PASSWORD);
+ verify(sessionMock).connect();
+ verify(sessionMock).openChannel("sftp");
+ verifyNoMoreInteractions(sessionMock);
+
+ verify(channelMock).connect();
+ verifyNoMoreInteractions(channelMock);
}
@Test
- public void collectFile_withWrongUserName_shouldFail() throws DatafileTaskException, IOException {
+ public void openWithoutPort_success()
+ throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
- .serverAddress("127.0.0.1") //
- .userId("wrong") //
+ .serverAddress(HOST) //
+ .userId(USERNAME) //
.password(PASSWORD) //
- .port(sftpServer.getPort()) //
+ .port(Optional.empty()) //
.build();
- try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
- sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
+ SftpClient sftpClientSpy = spy(new SftpClient(expectedFileServerData));
- assertThatThrownBy(() -> sftpClient.open())
- .hasMessageContaining("Could not open Sftp clientcom.jcraft.jsch.JSchException: Auth fail");
- }
+ doReturn(jschMock).when(sftpClientSpy).createJsch();
+ when(jschMock.getSession(anyString(), anyString(), anyInt())).thenReturn(sessionMock);
+ when(sessionMock.openChannel(anyString())).thenReturn(channelMock);
+
+ sftpClientSpy.open();
+
+ verify(jschMock).getSession(USERNAME, HOST, 22);
}
@Test
- public void collectFile_withWrongFileName_shouldFail()
- throws IOException, JSchException, SftpException, DatafileTaskException {
+ public void open_throwsException()
+ throws DatafileTaskException, IOException, JSchException, SftpException, Exception {
FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
- .serverAddress("127.0.0.1") //
+ .serverAddress(HOST) //
.userId(USERNAME) //
.password(PASSWORD) //
- .port(sftpServer.getPort()) //
+ .port(SFTP_PORT) //
.build();
- try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
- sftpServer.putFile(REMOTE_DUMMY_FILE, DUMMY_CONTENT, UTF_8);
- sftpClient.open();
- assertThatThrownBy(() -> sftpClient.collectFile("wrong", LOCAL_DUMMY_FILE))
- .hasMessageStartingWith("Unable to get file from xNF. Data: FileServerData{serverAddress=127.0.0.1, "
- + "userId=bob, password=123, port=");
- }
- }
+ SftpClient sftpClientSpy = spy(new SftpClient(expectedFileServerData));
- private static Session connectToServer(FakeSftpServerRule sftpServer) throws JSchException {
- return connectToServerAtPort(sftpServer.getPort());
- }
+ doReturn(jschMock).when(sftpClientSpy).createJsch();
+ when(jschMock.getSession(anyString(), anyString(), anyInt())).thenThrow(new JSchException("Failed"));
- private static Session connectToServerAtPort(int port) throws JSchException {
- Session session = createSessionWithCredentials(USERNAME, PASSWORD, port);
- session.connect(TIMEOUT);
- return session;
+ assertThatThrownBy(() -> sftpClientSpy.open())
+ .hasMessageStartingWith("Could not open Sftp client. com.jcraft.jsch.JSchException: Failed");
}
- private static ChannelSftp connectSftpChannel(Session session) throws JSchException {
- ChannelSftp channel = (ChannelSftp) session.openChannel("sftp");
- channel.connect();
- return channel;
- }
+ @SuppressWarnings("resource")
+ @Test
+ public void collectFile_succes() throws DatafileTaskException, SftpException {
+ FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
+ .serverAddress(HOST) //
+ .userId(USERNAME) //
+ .password(PASSWORD) //
+ .port(SFTP_PORT) //
+ .build();
+ SftpClient sftpClient = new SftpClient(expectedFileServerData);
+
+ sftpClient.sftpChannel = channelMock;
- private static Session createSessionWithCredentials(String username, String password, int port)
- throws JSchException {
- Session session = JSCH.getSession(username, "127.0.0.1", port);
- session.setConfig("StrictHostKeyChecking", "no");
- session.setPassword(password);
- return session;
+ sftpClient.collectFile("remote.xml", Paths.get("local.xml"));
+
+ verify(channelMock).get("remote.xml", "local.xml");
+ verifyNoMoreInteractions(channelMock);
}
- private static byte[] downloadFile(FakeSftpServerRule server, String path)
- throws JSchException, SftpException, IOException {
- Session session = connectToServer(server);
- ChannelSftp channel = connectSftpChannel(session);
- try {
- InputStream is = channel.get(path);
- return toByteArray(is);
- } finally {
- channel.disconnect();
- session.disconnect();
+ @Test
+ public void collectFile_throwsExceptionWithoutRetry()
+ throws IOException, JSchException, SftpException, DatafileTaskException {
+ FileServerData expectedFileServerData = ImmutableFileServerData.builder() //
+ .serverAddress(HOST) //
+ .userId(USERNAME) //
+ .password(PASSWORD) //
+ .port(SFTP_PORT) //
+ .build();
+
+ try (SftpClient sftpClient = new SftpClient(expectedFileServerData)) {
+ sftpClient.sftpChannel = channelMock;
+ doThrow(new SftpException(ChannelSftp.SSH_FX_NO_SUCH_FILE, "Failed")).when(channelMock).get(anyString(),
+ anyString());
+
+ assertThatThrownBy(() -> sftpClient.collectFile("remoteFile", Paths.get("localFile")))
+ .isInstanceOf(DatafileTaskException.class)
+ .hasMessageStartingWith("Unable to get file from xNF. Data: FileServerData{serverAddress=" + HOST
+ + ", " + "userId=" + USERNAME + ", password=####, port=" + SFTP_PORT);
}
}
+
+ @Test
+ public void close_succes() throws DatafileTaskException, SftpException {
+ SftpClient sftpClient = new SftpClient(null);
+
+ sftpClient.session = sessionMock;
+ sftpClient.sftpChannel = channelMock;
+
+ sftpClient.close();
+
+ verify(sessionMock).disconnect();
+ verifyNoMoreInteractions(sessionMock);
+
+ verify(channelMock).exit();;
+ verifyNoMoreInteractions(channelMock);
+ }
}
package org.onap.dcaegen2.collectors.datafile.service;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.spy;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
-
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
-
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
-
+import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
* @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
*/
class JsonMessageParserTest {
+ private static final String ERROR_LOG_TAG = "[ERROR] ";
+
private static final String NR_RADIO_ERICSSON_EVENT_NAME = "Noti_NrRadio-Ericsson_FileReady";
private static final String PRODUCT_NAME = "NrRadio";
private static final String VENDOR_NAME = "Ericsson";
private static final String FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
private static final String FILE_FORMAT_VERSION = "V10";
private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
- private static final String INCORRECT_CHANGE_IDENTIFIER = "INCORRECT_PM_MEAS_FILES";
private static final String CHANGE_TYPE = "FileReady";
private static final String INCORRECT_CHANGE_TYPE = "IncorrectFileReady";
private static final String NOTIFICATION_FIELDS_VERSION = "1.0";
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
+ StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
+ .expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue(logAppender.list.toString()
+ .contains("[ERROR] VES event parsing. File information wrong. " + "Missing location."));
+ assertTrue(logAppender.list.get(0).toString().contains("sourceName=5GRAN_DU"));
+ }
+
+ @Test
+ void whenPassingCorrectJsonWrongScheme_noMessage() {
+ AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() //
+ .name(PM_FILE_NAME) //
+ .location("http://location.xml") //
+ .compression(GZIP_COMPRESSION) //
+ .fileFormatType(FILE_FORMAT_TYPE) //
+ .fileFormatVersion(FILE_FORMAT_VERSION) //
+ .build();
+ JsonMessage message = new JsonMessage.JsonMessageBuilder() //
+ .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
+ .changeIdentifier(CHANGE_IDENTIFIER) //
+ .changeType(CHANGE_TYPE) //
+ .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION) //
+ .addAdditionalField(additionalField) //
+ .build();
+
+ String messageString = message.toString();
+ String parsedString = message.getParsed();
+ JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
+ JsonElement jsonElement = new JsonParser().parse(parsedString);
+ Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
+ .getJsonObjectFromAnArray(jsonElement);
+
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + Scheme.DFC_DOES_NOT_SUPPORT_PROTOCOL_ERROR_MSG + "http"
+ + Scheme.SUPPORTED_PROTOCOLS_ERROR_MESSAGE + ". Location: http://location.xml"));
+ assertTrue("Missing sourceName in log", logAppender.list.toString().contains("sourceName=5GRAN_DU"));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectComplete().verify();
+
+ assertTrue("Error missing in log", logAppender.list.toString().contains(ERROR_LOG_TAG
+ + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "Can not get PRODUCT_NAME from eventName, eventName is not in correct format: Faulty event name"));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "File information wrong. Missing data: [name] Data: "
+ + message.getAdditionalFields().get(0).toString()));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString().contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "Missing arrayOfNamedHashMap in message. " + message.getParsed()));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "File information wrong. Missing data: [compression] Data: "
+ + message.getAdditionalFields().get(0).toString()));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).verifyComplete();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "File information wrong. Missing data: [fileFormatType] Data: "
+ + message.getAdditionalFields().get(0).toString()));
}
@Test
void whenPassingJsonWithoutMandatoryHeaderInformation_noFileData() {
JsonMessage message = new JsonMessage.JsonMessageBuilder() //
.eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
- .changeIdentifier("PM_MEAS_FILES_INVALID") //
- .changeType("FileReady_INVALID") //
- .notificationFieldsVersion("1.0_INVALID") //
.build();
String incorrectMessageString = message.toString();
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(incorrectMessageString)))
.expectSubscription().expectComplete().verify();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + "Missing data: [changeIdentifier, changeType, notificationFieldsVersion]. "
+ + "Change type is wrong: Expected: FileReady Message: " + message.getParsed()));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just("[{}]"))).expectSubscription()
.expectComplete().verify();
+
+ assertTrue("Error missing in log",
+ logAppender.list.toString().contains(ERROR_LOG_TAG + "Incorrect JsonObject - missing header. "));
}
@Test
Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
.getJsonObjectFromAnArray(jsonElement);
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
.expectSubscription().expectNextCount(0).expectComplete().verify();
- }
- @Test
- void whenPassingCorrectJsonWithIncorrectChangeIdentifier_noFileData() {
- AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() //
- .name(PM_FILE_NAME) //
- .location(LOCATION) //
- .compression(GZIP_COMPRESSION) //
- .fileFormatVersion(FILE_FORMAT_VERSION) //
- .build();
- JsonMessage message = new JsonMessage.JsonMessageBuilder() //
- .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
- .changeIdentifier(INCORRECT_CHANGE_IDENTIFIER) //
- .changeType(CHANGE_TYPE) //
- .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION) //
- .addAdditionalField(additionalField) //
- .build();
-
- String messageString = message.toString();
- String parsedString = message.getParsed();
- JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
- JsonElement jsonElement = new JsonParser().parse(parsedString);
- Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
- .getJsonObjectFromAnArray(jsonElement);
-
- StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
- .expectSubscription().expectComplete().verify();
+ assertTrue("Error missing in log",
+ logAppender.list.toString()
+ .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
+ + " Change type is wrong: " + INCORRECT_CHANGE_TYPE + " Expected: FileReady Message: "
+ + message.getParsed()));
}
}
package org.onap.dcaegen2.collectors.datafile.tasks;
+import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.net.URI;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.StatusLine;
import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerHttpClient;
+import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
import org.springframework.http.HttpStatus;
-
import reactor.test.StepVerifier;
/**
void whenPassedObjectFits_firstFailsWithExceptionThenSucceeds() throws Exception {
prepareMocksForTests(new DatafileTaskException("Error"), HttpStatus.OK.value());
- StepVerifier //
- .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 2, Duration.ofSeconds(0)))
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DataRouterPublisher.class);
+ StepVerifier.create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 2, Duration.ofSeconds(0)))
.expectNext(filePublishInformation) //
.verifyComplete();
+
+ assertTrue("Warning missing in log", logAppender.list.toString()
+ .contains("[WARN] Publishing file " + PM_FILE_NAME + " to DR unsuccessful."));
}
@Test
prepareMocksForTests(null, Integer.valueOf(HttpStatus.BAD_GATEWAY.value()),
Integer.valueOf((HttpStatus.BAD_GATEWAY.value())));
- StepVerifier //
- .create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 1, Duration.ofSeconds(0)))
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(DataRouterPublisher.class);
+ StepVerifier.create(publisherTaskUnderTestSpy.publishFile(filePublishInformation, 1, Duration.ofSeconds(0)))
.expectErrorMessage("Retries exhausted: 1/1") //
.verify();
+ assertTrue("Warning missing in log", logAppender.list.toString().contains("[WARN] Publishing file "
+ + PM_FILE_NAME + " to DR unsuccessful. Response code: " + HttpStatus.BAD_GATEWAY));
+
verify(httpClientMock, times(2)).addUserCredentialsToHead(any(HttpUriRequest.class));
verify(httpClientMock, times(2)).getDmaapProducerResponseWithRedirect(any(HttpUriRequest.class), any());
verifyNoMoreInteractions(httpClientMock);
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
-
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.StatusLine;
@BeforeAll
- public static void setUp() throws DatafileTaskException {
+ private static void setUp() throws DatafileTaskException {
when(publisherConfigurationMock.publishUrl()).thenReturn(PUBLISH_URL);
appConfigMock = mock(AppConfig.class);
package org.onap.dcaegen2.collectors.datafile.tasks;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import java.nio.file.Paths;
import java.time.Duration;
+import java.time.Instant;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-
+import org.apache.commons.lang3.StringUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileReadyMessage;
import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
-
+import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables;
+import org.slf4j.MDC;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
private AppConfig appConfig = mock(AppConfig.class);
- private ScheduledTasks testedObject = spy(new ScheduledTasks(appConfig));
+ private ScheduledTasks testedObject;
private int uniqueValue = 0;
private DMaaPMessageConsumer consumerMock;
@BeforeEach
private void setUp() throws DatafileTaskException {
+ testedObject = spy(new ScheduledTasks(appConfig));
+
+ consumerMock = mock(DMaaPMessageConsumer.class);
+ publishedCheckerMock = mock(PublishedChecker.class);
+ fileCollectorMock = mock(FileCollector.class);
+ dataRouterMock = mock(DataRouterPublisher.class);
+
+ doReturn(consumerMock).when(testedObject).createConsumerTask();
+ doReturn(publishedCheckerMock).when(testedObject).createPublishedChecker();
+ doReturn(fileCollectorMock).when(testedObject).createFileCollector();
+ doReturn(dataRouterMock).when(testedObject).createDataRouterPublisher();
+ }
+
+ private void setUpConfiguration() throws DatafileTaskException {
final PublisherConfiguration dmaapPublisherConfiguration = ImmutablePublisherConfiguration.builder() //
.publishUrl(publishUrl) //
.logUrl("") //
doReturn(dmaapPublisherConfiguration).when(appConfig).getPublisherConfiguration(CHANGE_IDENTIFIER);
doReturn(dmaapConsumerConfiguration).when(appConfig).getDmaapConsumerConfiguration();
doReturn(true).when(appConfig).isFeedConfigured(CHANGE_IDENTIFIER);
-
- consumerMock = mock(DMaaPMessageConsumer.class);
- publishedCheckerMock = mock(PublishedChecker.class);
- fileCollectorMock = mock(FileCollector.class);
- dataRouterMock = mock(DataRouterPublisher.class);
-
- doReturn(consumerMock).when(testedObject).createConsumerTask();
- doReturn(publishedCheckerMock).when(testedObject).createPublishedChecker();
- doReturn(fileCollectorMock).when(testedObject).createFileCollector();
- doReturn(dataRouterMock).when(testedObject).createDataRouterPublisher();
}
private MessageMetaData messageMetaData() {
private FileData fileData(int instanceNumber) {
return ImmutableFileData.builder() //
- .name("name" + instanceNumber) //
+ .name(PM_FILE_NAME + instanceNumber) //
.fileFormatType("") //
.fileFormatVersion("") //
.location("ftpes://192.168.0.101/ftp/rop/" + PM_FILE_NAME + instanceNumber) //
}
@Test
- public void notingToConsume() throws DatafileTaskException {
+ public void purgeFileCache() {
+ testedObject.publishedFilesCache.put(Paths.get("file.xml"));
+
+ testedObject.purgeCachedInformation(Instant.MAX);
+
+ assertEquals(0, testedObject.publishedFilesCacheSize());
+ }
+
+ @Test
+ public void nothingToConsume() throws DatafileTaskException {
+ setUpConfiguration();
+
doReturn(consumerMock).when(testedObject).createConsumerTask();
doReturn(Flux.empty()).when(consumerMock).getMessageRouterResponse();
verifyNoMoreInteractions(consumerMock);
}
+ @Test
+ public void skippingConsumeDueToCurrentNumberOfTasksGreaterThan50() {
+ doReturn(51).when(testedObject).getCurrentNumberOfTasks();
+
+ testedObject.executeDatafileMainTask();
+
+ verifyNoMoreInteractions(consumerMock);
+ }
+
+ @Test
+ public void executeDatafileMainTask_successfulCase() throws DatafileTaskException {
+ setUpConfiguration();
+
+ final int noOfEvents = 1;
+ final int noOfFilesPerEvent = 1;
+
+ Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(noOfEvents, noOfFilesPerEvent, true);
+ doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
+
+ doReturn(false).when(publishedCheckerMock).isFilePublished(anyString(), any(), any());
+
+ Mono<FilePublishInformation> collectedFile = Mono.just(filePublishInformation());
+ doReturn(collectedFile).when(fileCollectorMock).collectFile(notNull(), anyLong(), notNull(), notNull());
+ doReturn(collectedFile).when(dataRouterMock).publishFile(notNull(), anyLong(), notNull());
+
+ testedObject.executeDatafileMainTask();
+
+ await().untilAsserted(() -> assertEquals(0, testedObject.getCurrentNumberOfSubscriptions()));
+
+ assertFalse(StringUtils.isBlank(MDC.get(MdcVariables.REQUEST_ID)));
+
+ verify(appConfig).getDmaapConsumerConfiguration();
+ verify(appConfig).isFeedConfigured(CHANGE_IDENTIFIER);
+ verifyNoMoreInteractions(appConfig);
+ }
+
+ @Test
+ public void executeDatafileMainTask_unconfiguredChangeIdentifier() throws DatafileTaskException {
+ final PublisherConfiguration dmaapPublisherConfiguration = ImmutablePublisherConfiguration.builder() //
+ .publishUrl(publishUrl) //
+ .logUrl("") //
+ .userName("userName") //
+ .passWord("passWord") //
+ .trustStorePath("trustStorePath") //
+ .trustStorePasswordPath("trustStorePasswordPath") //
+ .keyStorePath("keyStorePath") //
+ .keyStorePasswordPath("keyStorePasswordPath") //
+ .enableDmaapCertAuth(true) //
+ .changeIdentifier("Different changeIdentifier") //
+ .build(); //
+ final ConsumerConfiguration dmaapConsumerConfiguration = ImmutableConsumerConfiguration.builder() //
+ .topicUrl("topicUrl").trustStorePath("trustStorePath") //
+ .trustStorePasswordPath("trustStorePasswordPath") //
+ .keyStorePath("keyStorePath") //
+ .keyStorePasswordPath("keyStorePasswordPath") //
+ .enableDmaapCertAuth(true) //
+ .build();
+
+ doReturn(dmaapPublisherConfiguration).when(appConfig).getPublisherConfiguration(CHANGE_IDENTIFIER);
+ doReturn(dmaapConsumerConfiguration).when(appConfig).getDmaapConsumerConfiguration();
+ doReturn(false).when(appConfig).isFeedConfigured(CHANGE_IDENTIFIER);
+ final int noOfEvents = 1;
+ final int noOfFilesPerEvent = 1;
+
+ Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(noOfEvents, noOfFilesPerEvent, true);
+ doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
+
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ScheduledTasks.class);
+ testedObject.executeDatafileMainTask();
+
+ await().untilAsserted(() -> assertEquals(0, testedObject.getCurrentNumberOfSubscriptions()));
+
+ assertTrue("Error missing in log", logAppender.list.toString().contains(
+ "[INFO] No feed is configured for: " + CHANGE_IDENTIFIER + ", file ignored: " + PM_FILE_NAME + "1"));
+ }
+
+ @Test
+ public void createMainTask_consumeFail() {
+ MDC.setContextMap(contextMap);
+ doReturn(Flux.error(new Exception("Failed"))).when(consumerMock).getMessageRouterResponse();
+
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ScheduledTasks.class);
+ StepVerifier //
+ .create(testedObject.createMainTask(contextMap)) //
+ .expectSubscription() //
+ .expectNextCount(0) //
+ .expectComplete() //
+ .verify(); //
+
+ assertTrue("Error missing in log", logAppender.list.toString().contains(
+ "[ERROR] Polling for file ready message failed, " + "exception: java.lang.Exception: Failed"));
+ }
+
@Test
public void consume_successfulCase() throws DatafileTaskException {
+ setUpConfiguration();
+
final int noOfEvents = 200;
final int noOfFilesPerEvent = 200;
final int noOfFiles = noOfEvents * noOfFilesPerEvent;
@Test
public void consume_fetchFailedOnce() throws DatafileTaskException {
+ setUpConfiguration();
+
Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(2, 2, true); // 4 files
doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
@Test
public void consume_publishFailedOnce() throws DatafileTaskException {
+ setUpConfiguration();
Flux<FileReadyMessage> fileReadyMessages = fileReadyMessageFlux(2, 2, true); // 4 files
doReturn(fileReadyMessages).when(consumerMock).getMessageRouterResponse();
.when(dataRouterMock) //
.publishFile(notNull(), anyLong(), notNull());
+ ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(ScheduledTasks.class);
StepVerifier //
.create(testedObject.createMainTask(contextMap)) //
.expectSubscription() //
.expectComplete() //
.verify(); //
+ assertTrue("Error missing in log", logAppender.list.toString().contains("[ERROR] File publishing failed: "));
+
assertEquals(0, testedObject.getCurrentNumberOfTasks());
verify(consumerMock, times(1)).getMessageRouterResponse();
verify(fileCollectorMock, times(4)).collectFile(notNull(), anyLong(), notNull(), notNull());
@Test
public void consume_successfulCase_sameFileNames() throws DatafileTaskException {
+ setUpConfiguration();
+
final int noOfEvents = 1;
final int noOfFilesPerEvent = 100;
private String notificationFieldsVersion;
private List<AdditionalField> arrayOfAdditionalFields;
+ public List<AdditionalField> getAdditionalFields() {
+ return arrayOfAdditionalFields;
+ }
@Override
public String toString() {
public String getParsed() {
StringBuffer additionalFieldsString = new StringBuffer();
if (arrayOfAdditionalFields.size() > 0) {
- additionalFieldsString.append("\"arrayOfNamedHashMap\": [");
+ additionalFieldsString.append("\"arrayOfNamedHashMap\":[");
for (Iterator<AdditionalField> iterator = arrayOfAdditionalFields.iterator(); iterator.hasNext();) {
AdditionalField additionalField = iterator.next();
additionalFieldsString.append(additionalField.toString());
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+
import java.util.HashMap;
import java.util.Map;
import org.apache.http.Header;
<!--TEST SETTINGS -->
<surefire.redirectTestOutputToFile>true</surefire.redirectTestOutputToFile>
<junit-jupiter.version>5.1.0</junit-jupiter.version>
- <junit-vintage.version>5.1.0</junit-vintage.version>
+ <junit.version>5.3.2</junit.version>
<junit-platform.version>1.1.0</junit-platform.version>
- <mockito.version>2.23.0</mockito.version>
+ <mockito.version>2.23.4</mockito.version>
<testng.version>6.14.2</testng.version>
- <stefanbirkner.version>2.0.1</stefanbirkner.version>
<jcraft.version>0.1.54</jcraft.version>
<springfox.version>2.8.0</springfox.version>
+ <awaitility.version>3.1.6</awaitility.version>
<!-- Plugin versions -->
<maven-resources-plugin.version>3.1.0</maven-resources-plugin.version>
<artifactId>dmaap-client</artifactId>
<version>${sdk.version}</version>
</dependency>
- <dependency>
- <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
- <artifactId>common-dependency</artifactId>
- <version>${sdk.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpasyncclient</artifactId>
<version>${apache.httpcomponents.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>
+ <artifactId>common-dependency</artifactId>
+ <version>${sdk.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${junit-jupiter.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.junit.vintage</groupId>
- <artifactId>junit-vintage-engine</artifactId>
- <version>${junit-vintage.version}</version>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-junit-jupiter</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>com.github.stefanbirkner</groupId>
- <artifactId>fake-sftp-server-rule</artifactId>
- <version>${stefanbirkner.version}</version>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
</dependency>
<!--REQUIRED TO GENERATE DOCUMENTATION -->
Code Review / dcaegen2 / collectors / datafile.git / commitdiff