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.BeforeAll;
  31 import org.junit.jupiter.api.Test;
  32
  33 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
  34 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  35 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapEmptyResponseException;
  36 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  37 import org.onap.dcaegen2.collectors.datafile.model.FileData;
  38 import org.onap.dcaegen2.collectors.datafile.model.FileMetaData;
  39 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
  40 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
  41 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileMetaData;
  42 import org.onap.dcaegen2.collectors.datafile.service.DmaapConsumerJsonParser;
  43
  44 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
  45 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
  46
  47 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
  48 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapConsumerConfiguration;
  49 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
  50 import reactor.core.publisher.Flux;
  51 import reactor.core.publisher.Mono;
  52 import reactor.test.StepVerifier;
  53
  54 /**
  55  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
  56  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  57  */
  58 class DmaapConsumerTaskImplTest {
  59     private static final String NR_RADIO_ERICSSON_EVENT_NAME = "Noti_NrRadio-Ericsson_FileReady";
  60     private static final String PRODUCT_NAME = "NrRadio";
  61     private static final String VENDOR_NAME = "Ericsson";
  62     private static final String LAST_EPOCH_MICROSEC = "8745745764578";
  63     private static final String SOURCE_NAME = "oteNB5309";
  64     private static final String START_EPOCH_MICROSEC = "8745745764578";
  65     private static final String TIME_ZONE_OFFSET = "UTC+05:00";
  66     private static final String PM_MEAS_CHANGE_IDENTIFIER = "PM_MEAS_FILES";
  67     private static final String FILE_READY_CHANGE_TYPE = "FileReady";
  68     private static final String FTPES_SCHEME = "ftpes://";
  69     private static final String SFTP_SCHEME = "sftp://";
  70     private static final String SERVER_ADDRESS = "192.168.0.101";
  71     private static final String PORT_22 = "22";
  72     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
  73     private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
  74     private static final String LOCAL_FILE_LOCATION = "target/" + PM_FILE_NAME;
  75     private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  76     private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  77     private static final String GZIP_COMPRESSION = "gzip";
  78     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
  79     private static final String FILE_FORMAT_VERSION = "V10";
  80
  81     private static List<ConsumerDmaapModel> listOfConsumerDmaapModel = new ArrayList<ConsumerDmaapModel>();
  82
  83     private static AppConfig appConfig;
  84     private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
  85     private DmaapConsumerTaskImpl dmaapConsumerTask;
  86     private DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
  87
  88     private static String ftpesMessage;
  89     private static FileData ftpesFileData;
  90
  91     private static String sftpMessage;
  92     private static FileData sftpFileData;
  93
  94     @BeforeAll
  95     public static void setUp() {
  96         //@formatter:off
  97         dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder()
  98                 .consumerGroup("OpenDCAE-c12")
  99                 .consumerId("c12")
 100                 .dmaapContentType("application/json")
 101                 .dmaapHostName("54.45.33.2")
 102                 .dmaapPortNumber(1234).dmaapProtocol("https")
 103                 .dmaapUserName("Datafile")
 104                 .dmaapUserPassword("Datafile")
 105                 .dmaapTopicName("unauthenticated.NOTIFICATION")
 106                 .timeoutMs(-1)
 107                 .messageLimit(-1)
 108                 .trustStorePath("trustStorePath")
 109                 .trustStorePasswordPath("trustStorePasswordPath")
 110                 .keyStorePath("keyStorePath")
 111                 .keyStorePasswordPath("keyStorePasswordPath")
 112                 .enableDmaapCertAuth(true)
 113                 .build();
 114
 115         appConfig = mock(AppConfig.class);
 116
 117         AdditionalField ftpesAdditionalField = new JsonMessage.AdditionalFieldBuilder()
 118                 .location(FTPES_LOCATION)
 119                 .compression(GZIP_COMPRESSION)
 120                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
 121                 .fileFormatVersion(FILE_FORMAT_VERSION)
 122                 .build();
 123
 124         JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder()
 125                 .eventName(NR_RADIO_ERICSSON_EVENT_NAME)
 126                 .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
 127                 .changeType(FILE_READY_CHANGE_TYPE)
 128                 .notificationFieldsVersion("1.0")
 129                 .addAdditionalField(ftpesAdditionalField)
 130                 .build();
 131
 132         ftpesMessage = ftpesJsonMessage.toString();
 133         FileMetaData fileMetaData = ImmutableFileMetaData.builder()
 134                 .productName(PRODUCT_NAME)
 135                 .vendorName(VENDOR_NAME)
 136                 .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
 137                 .sourceName(SOURCE_NAME)
 138                 .startEpochMicrosec(START_EPOCH_MICROSEC)
 139                 .timeZoneOffset(TIME_ZONE_OFFSET)
 140                 .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
 141                 .changeType(FILE_READY_CHANGE_TYPE)
 142                 .build();
 143         ftpesFileData = ImmutableFileData.builder()
 144                 .fileMetaData(fileMetaData)
 145                 .name(PM_FILE_NAME)
 146                 .location(FTPES_LOCATION)
 147                 .compression(GZIP_COMPRESSION)
 148                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
 149                 .fileFormatVersion(FILE_FORMAT_VERSION)
 150                 .build();
 151
 152         AdditionalField sftpAdditionalField = new JsonMessage.AdditionalFieldBuilder()
 153                 .location(SFTP_LOCATION)
 154                 .compression(GZIP_COMPRESSION)
 155                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
 156                 .fileFormatVersion(FILE_FORMAT_VERSION)
 157                 .build();
 158         JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder()
 159                 .eventName(NR_RADIO_ERICSSON_EVENT_NAME)
 160                 .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
 161                 .changeType(FILE_READY_CHANGE_TYPE)
 162                 .notificationFieldsVersion("1.0")
 163                 .addAdditionalField(sftpAdditionalField)
 164                 .build();
 165         sftpMessage = sftpJsonMessage.toString();
 166         sftpFileData = ImmutableFileData.builder()
 167                 .fileMetaData(fileMetaData)
 168                 .name(PM_FILE_NAME)
 169                 .location(SFTP_LOCATION)
 170                 .compression(GZIP_COMPRESSION)
 171                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
 172                 .fileFormatVersion(FILE_FORMAT_VERSION)
 173                 .build();
 174
 175
 176         ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder()
 177                 .productName(PRODUCT_NAME)
 178                 .vendorName(VENDOR_NAME)
 179                 .lastEpochMicrosec(LAST_EPOCH_MICROSEC)
 180                 .sourceName(SOURCE_NAME)
 181                 .startEpochMicrosec(START_EPOCH_MICROSEC)
 182                 .timeZoneOffset(TIME_ZONE_OFFSET)
 183                 .name(PM_FILE_NAME)
 184                 .location(FTPES_LOCATION)
 185                 .internalLocation(LOCAL_FILE_LOCATION)
 186                 .compression(GZIP_COMPRESSION)
 187                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
 188                 .fileFormatVersion(FILE_FORMAT_VERSION)
 189                 .build();
 190         listOfConsumerDmaapModel.add(consumerDmaapModel);
 191         //@formatter:on
 192     }
 193
 194     @Test
 195     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
 196         prepareMocksForDmaapConsumer("", null);
 197
 198         StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
 199                 .expectError(DmaapEmptyResponseException.class).verify();
 200
 201         verify(dmaapConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
 202     }
 203
 204     @Test
 205     public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
 206         prepareMocksForDmaapConsumer(ftpesMessage, ftpesFileData);
 207
 208         StepVerifier.create(dmaapConsumerTask.execute(ftpesMessage)).expectNext(ftpesFileData).verifyComplete();
 209
 210         verify(dmaapConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
 211         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
 212     }
 213
 214     @Test
 215     public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
 216         prepareMocksForDmaapConsumer(sftpMessage, sftpFileData);
 217
 218         StepVerifier.create(dmaapConsumerTask.execute(ftpesMessage)).expectNext(sftpFileData).verifyComplete();
 219
 220         verify(dmaapConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
 221         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
 222     }
 223
 224     private void prepareMocksForDmaapConsumer(String message, FileData fileDataAfterConsume) {
 225         Mono<String> messageAsMono = Mono.just(message);
 226         DmaapConsumerJsonParser dmaapConsumerJsonParserMock = mock(DmaapConsumerJsonParser.class);
 227         dmaapConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
 228         when(dmaapConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(messageAsMono);
 229
 230         if (!message.isEmpty()) {
 231             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono)).thenReturn(Flux.just(fileDataAfterConsume));
 232         } else {
 233             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono))
 234                     .thenReturn(Flux.error(new DmaapEmptyResponseException()));
 235         }
 236
 237         dmaapConsumerTask =
 238                 spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerReactiveHttpClient, dmaapConsumerJsonParserMock));
 239         when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
 240         doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
 241     }
 242 }