gerrit.onap Code Review - dcaegen2/collectors/datafile.git/blob

   1 /*
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018 Nordix Foundation. All rights reserved.
   4  * ===============================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   6  * in compliance with the License. You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software distributed under the License
  11  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  12  * or implied. See the License for the specific language governing permissions and limitations under
  13  * the License.
  14  * ============LICENSE_END========================================================================
  15  */
  16
  17 package org.onap.dcaegen2.collectors.datafile.tasks;
  18
  19 import static org.mockito.Mockito.doReturn;
  20 import static org.mockito.Mockito.mock;
  21 import static org.mockito.Mockito.spy;
  22 import static org.mockito.Mockito.times;
  23 import static org.mockito.Mockito.verify;
  24 import static org.mockito.Mockito.verifyNoMoreInteractions;
  25 import static org.mockito.Mockito.when;
  26
  27 import java.util.ArrayList;
  28 import java.util.List;
  29
  30 import org.junit.jupiter.api.Assertions;
  31 import org.junit.jupiter.api.BeforeAll;
  32 import org.junit.jupiter.api.Test;
  33 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
  34 import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
  35 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
  36 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  37 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapEmptyResponseException;
  38 import org.onap.dcaegen2.collectors.datafile.ftp.FileCollector;
  39 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  40 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
  41 import org.onap.dcaegen2.collectors.datafile.service.DmaapConsumerJsonParser;
  42 import org.onap.dcaegen2.collectors.datafile.service.FileData;
  43 import org.onap.dcaegen2.collectors.datafile.service.ImmutableFileData;
  44 import org.onap.dcaegen2.collectors.datafile.service.consumer.DmaapConsumerReactiveHttpClient;
  45 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
  46 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
  47
  48 import reactor.core.publisher.Mono;
  49 import reactor.test.StepVerifier;
  50
  51 /**
  52  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
  53  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  54  */
  55 class DmaapConsumerTaskImplTest {
  56     private static final String PM_MEAS_CHANGE_IDINTIFIER = "PM_MEAS_FILES";
  57     private static final String FILE_READY_CHANGE_TYPE = "FileReady";
  58     private static final String FTPES_SCHEME = "ftpes://";
  59     private static final String SFTP_SCHEME = "sftp://";
  60     private static final String SERVER_ADDRESS = "192.168.0.101";
  61     private static final String PORT_22 = "22";
  62     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
  63     private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
  64     private static final String LOCAL_FILE_LOCATION = "target/" + PM_FILE_NAME;
  65     private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  66     private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  67     private static final String GZIP_COMPRESSION = "gzip";
  68     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
  69     private static final String FILE_FORMAT_VERSION = "V10";
  70
  71     private static List<ConsumerDmaapModel> listOfConsumerDmaapModel = new ArrayList<ConsumerDmaapModel>();
  72
  73     private static AppConfig appConfig;
  74     private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
  75     private DmaapConsumerTaskImpl dmaapConsumerTask;
  76     private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
  77
  78     private static FileCollector fileCollectorMock;
  79
  80     private static String ftpesMessage;
  81     private static List<FileData> ftpesFileDataAfterConsume = new ArrayList<FileData>();
  82
  83     private static String sftpMessage;
  84     private static List<FileData> sftpFileDataAfterConsume = new ArrayList<FileData>();
  85
  86     @BeforeAll
  87     public static void setUp() {
  88         dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder().consumerGroup("OpenDCAE-c12")
  89                 .consumerId("c12").dmaapContentType("application/json").dmaapHostName("54.45.33.2")
  90                 .dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
  91                 .dmaapTopicName("unauthenticated.NOTIFICATION").timeoutMS(-1).messageLimit(-1).build();
  92
  93         appConfig = mock(AppConfig.class);
  94
  95         AdditionalField ftpesAdditionalField =
  96                 new JsonMessage.AdditionalFieldBuilder().location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
  97                         .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
  98         JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
  99                 .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
 100                 .addAdditionalField(ftpesAdditionalField).build();
 101         ftpesMessage = ftpesJsonMessage.toString();
 102         FileData ftpesFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
 103                 .changeType(FILE_READY_CHANGE_TYPE).location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
 104                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
 105         ftpesFileDataAfterConsume.add(ftpesFileData);
 106
 107         AdditionalField sftpAdditionalField =
 108                 new JsonMessage.AdditionalFieldBuilder().location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
 109                         .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
 110         JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
 111                 .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
 112                 .addAdditionalField(sftpAdditionalField).build();
 113         sftpMessage = sftpJsonMessage.toString();
 114         FileData sftpFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDINTIFIER)
 115                 .changeType(FILE_READY_CHANGE_TYPE).location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
 116                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
 117         sftpFileDataAfterConsume.add(sftpFileData);
 118
 119
 120         ImmutableConsumerDmaapModel consumerDmaapModel =
 121                 ImmutableConsumerDmaapModel.builder().location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
 122                         .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
 123         listOfConsumerDmaapModel.add(consumerDmaapModel);
 124
 125         fileCollectorMock = mock(FileCollector.class);
 126     }
 127
 128     @Test
 129     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
 130         // given
 131         prepareMocksForDmaapConsumer("", new ArrayList<FileData>());
 132
 133         // then
 134         StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
 135                 .expectError(DmaapEmptyResponseException.class).verify();
 136
 137         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
 138     }
 139
 140     @Test
 141     public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
 142         // given
 143         prepareMocksForDmaapConsumer(ftpesMessage, ftpesFileDataAfterConsume);
 144         // when
 145         final List<ConsumerDmaapModel> arrayOfResponse = dmaapConsumerTask.execute("Sample input").block();
 146         // then
 147         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
 148         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
 149         verify(fileCollectorMock, times(1)).getFilesFromSender(ftpesFileDataAfterConsume);
 150         verifyNoMoreInteractions(fileCollectorMock);
 151         Assertions.assertEquals(listOfConsumerDmaapModel, arrayOfResponse);
 152
 153     }
 154
 155     @Test
 156     public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
 157         // given
 158         prepareMocksForDmaapConsumer(sftpMessage, sftpFileDataAfterConsume);
 159         // when
 160         final List<ConsumerDmaapModel> arrayOfResponse = dmaapConsumerTask.execute("Sample input").block();
 161         // then
 162         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
 163         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
 164         verify(fileCollectorMock, times(1)).getFilesFromSender(sftpFileDataAfterConsume);
 165         verifyNoMoreInteractions(fileCollectorMock);
 166         Assertions.assertEquals(listOfConsumerDmaapModel, arrayOfResponse);
 167
 168     }
 169
 170     private void prepareMocksForDmaapConsumer(String message, List<FileData> fileDataAfterConsume) {
 171         Mono<String> messageAsMono = Mono.just(message);
 172         DmaapConsumerJsonParser dmaapConsumerJsonParserMock = mock(DmaapConsumerJsonParser.class);
 173         dmaapConsumerReactiveHttpClient = mock(DmaapConsumerReactiveHttpClient.class);
 174         when(dmaapConsumerReactiveHttpClient.getDmaapConsumerResponse()).thenReturn(messageAsMono);
 175
 176         if (!message.isEmpty()) {
 177             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono)).thenReturn(Mono.just(fileDataAfterConsume));
 178         } else {
 179             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono))
 180                     .thenReturn(Mono.error(new DmaapEmptyResponseException()));
 181         }
 182         when(fileCollectorMock.getFilesFromSender(fileDataAfterConsume))
 183                 .thenReturn(Mono.just(listOfConsumerDmaapModel));
 184
 185         dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerReactiveHttpClient,
 186                 dmaapConsumerJsonParserMock, fileCollectorMock));
 187         when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
 188         doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
 189     }
 190 }