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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation.
   4  * Copyright (C) 2020 Nokia. All rights reserved.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.dcaegen2.collectors.datafile.tasks;
  23
  24 import com.google.gson.JsonElement;
  25 import com.google.gson.JsonParser;
  26 import org.junit.jupiter.api.BeforeAll;
  27 import org.junit.jupiter.api.Test;
  28 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
  29 import org.onap.dcaegen2.collectors.datafile.configuration.ConsumerConfiguration;
  30 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  31 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
  32 import org.onap.dcaegen2.collectors.datafile.model.FileData;
  33 import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
  34 import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage;
  35 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileData;
  36 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFilePublishInformation;
  37 import org.onap.dcaegen2.collectors.datafile.model.ImmutableFileReadyMessage;
  38 import org.onap.dcaegen2.collectors.datafile.model.ImmutableMessageMetaData;
  39 import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
  40 import org.onap.dcaegen2.collectors.datafile.service.JsonMessageParser;
  41 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
  42 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
  43 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
  44 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterSubscribeRequest;
  45 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
  46 import reactor.core.publisher.Flux;
  47 import reactor.test.StepVerifier;
  48
  49 import java.nio.file.Path;
  50 import java.nio.file.Paths;
  51 import java.util.ArrayList;
  52 import java.util.HashMap;
  53 import java.util.List;
  54 import java.util.Optional;
  55
  56 import static org.mockito.ArgumentMatchers.any;
  57 import static org.mockito.Mockito.mock;
  58 import static org.mockito.Mockito.spy;
  59 import static org.mockito.Mockito.times;
  60 import static org.mockito.Mockito.verify;
  61 import static org.mockito.Mockito.verifyNoMoreInteractions;
  62 import static org.mockito.Mockito.when;
  63
  64 public class DMaaPMessageConsumerTest {
  65     private static final String NR_RADIO_ERICSSON_EVENT_NAME = "Noti_NrRadio-Ericsson_FileReady";
  66     private static final String PRODUCT_NAME = "NrRadio";
  67     private static final String VENDOR_NAME = "Ericsson";
  68     private static final String LAST_EPOCH_MICROSEC = "8745745764578";
  69     private static final String SOURCE_NAME = "oteNB5309";
  70     private static final String START_EPOCH_MICROSEC = "8745745764578";
  71     private static final String TIME_ZONE_OFFSET = "UTC+05:00";
  72     private static final String PM_MEAS_CHANGE_IDENTIFIER = "PM_MEAS_FILES";
  73     private static final String FILE_READY_CHANGE_TYPE = "FileReady";
  74     private static final String FTPES_SCHEME = "ftpes://";
  75     private static final String SFTP_SCHEME = "sftp://";
  76     private static final String SERVER_ADDRESS = "192.168.0.101";
  77     private static final String PORT_22 = "22";
  78     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
  79     private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
  80     private static final Path LOCAL_FILE_LOCATION = Paths.get("target/" + PM_FILE_NAME);
  81     private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  82     private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
  83     private static final String GZIP_COMPRESSION = "gzip";
  84     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
  85     private static final String FILE_FORMAT_VERSION = "V10";
  86     private static List<FilePublishInformation> listOfFilePublishInformation = new ArrayList<>();
  87     private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
  88
  89     private DMaaPMessageConsumer messageConsumer;
  90     private static String ftpesMessageString;
  91     private static JsonElement ftpesMessageJson;
  92     private static FileData ftpesFileData;
  93     private static FileReadyMessage expectedFtpesMessage;
  94
  95     private static String sftpMessageString;
  96     private static JsonElement sftpMessageJson;
  97     private static FileData sftpFileData;
  98     private static FileReadyMessage expectedSftpMessage;
  99
 100     private static AppConfig appConfig;
 101     private static ConsumerConfiguration dmaapConsumerConfiguration;
 102     private static MessageRouterSubscriber messageRouterSubscriber;
 103
 104     /**
 105      * Sets up data for the test.
 106      */
 107     @BeforeAll
 108     public static void setUp() {
 109
 110         appConfig = mock(AppConfig.class);
 111
 112         AdditionalField ftpesAdditionalField = new JsonMessage.AdditionalFieldBuilder() //
 113             .location(FTPES_LOCATION) //
 114             .compression(GZIP_COMPRESSION) //
 115             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
 116             .fileFormatVersion(FILE_FORMAT_VERSION) //
 117             .build();
 118
 119         JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder() //
 120             .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
 121             .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER) //
 122             .changeType(FILE_READY_CHANGE_TYPE) //
 123             .notificationFieldsVersion("1.0") //
 124             .addAdditionalField(ftpesAdditionalField) //
 125             .build();
 126
 127         ftpesMessageString = ftpesJsonMessage.toString();
 128         ftpesMessageJson = JsonParser.parseString(ftpesMessageString);
 129
 130         MessageMetaData messageMetaData = ImmutableMessageMetaData.builder() //
 131             .productName(PRODUCT_NAME) //
 132             .vendorName(VENDOR_NAME) //
 133             .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
 134             .sourceName(SOURCE_NAME) //
 135             .startEpochMicrosec(START_EPOCH_MICROSEC) //
 136             .timeZoneOffset(TIME_ZONE_OFFSET) //
 137             .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER) //
 138             .changeType(FILE_READY_CHANGE_TYPE) //
 139             .build();
 140         ftpesFileData = ImmutableFileData.builder() //
 141             .name(PM_FILE_NAME) //
 142             .location(FTPES_LOCATION) //
 143             .scheme(Scheme.FTPES) //
 144             .compression(GZIP_COMPRESSION) //
 145             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
 146             .fileFormatVersion(FILE_FORMAT_VERSION) //
 147             .messageMetaData(messageMetaData) //
 148             .build();
 149
 150         List<FileData> files = new ArrayList<>();
 151         files.add(ftpesFileData);
 152         expectedFtpesMessage = ImmutableFileReadyMessage.builder() //
 153             .files(files) //
 154             .build();
 155
 156         AdditionalField sftpAdditionalField = new JsonMessage.AdditionalFieldBuilder() //
 157             .location(SFTP_LOCATION) //
 158             .compression(GZIP_COMPRESSION) //
 159             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
 160             .fileFormatVersion(FILE_FORMAT_VERSION) //
 161             .build();
 162         JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder() //
 163             .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
 164             .changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER) //
 165             .changeType(FILE_READY_CHANGE_TYPE) //
 166             .notificationFieldsVersion("1.0") //
 167             .addAdditionalField(sftpAdditionalField) //
 168             .build();
 169         sftpMessageString = sftpJsonMessage.toString();
 170         sftpMessageJson = JsonParser.parseString(sftpMessageString);
 171         sftpFileData = ImmutableFileData.builder() //
 172             .name(PM_FILE_NAME) //
 173             .location(SFTP_LOCATION) //
 174             .scheme(Scheme.FTPES) //
 175             .compression(GZIP_COMPRESSION) //
 176             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
 177             .fileFormatVersion(FILE_FORMAT_VERSION) //
 178             .messageMetaData(messageMetaData) //
 179             .build();
 180
 181         ImmutableFilePublishInformation filePublishInformation = ImmutableFilePublishInformation.builder() //
 182             .productName(PRODUCT_NAME) //
 183             .vendorName(VENDOR_NAME) //
 184             .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
 185             .sourceName(SOURCE_NAME) //
 186             .startEpochMicrosec(START_EPOCH_MICROSEC) //
 187             .timeZoneOffset(TIME_ZONE_OFFSET) //
 188             .name(PM_FILE_NAME) //
 189             .location(FTPES_LOCATION) //
 190             .internalLocation(LOCAL_FILE_LOCATION) //
 191             .compression(GZIP_COMPRESSION) //
 192             .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) //
 193             .fileFormatVersion(FILE_FORMAT_VERSION) //
 194             .changeIdentifier(CHANGE_IDENTIFIER) //
 195             .context(new HashMap<String, String>()) //
 196             .build();
 197         listOfFilePublishInformation.add(filePublishInformation);
 198
 199         files = new ArrayList<>();
 200         files.add(sftpFileData);
 201         expectedSftpMessage = ImmutableFileReadyMessage.builder() //
 202             .files(files) //
 203             .build();
 204     }
 205
 206     @Test
 207     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
 208         prepareMocksForDmaapConsumer(Optional.empty(), null);
 209
 210         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
 211             .expectSubscription() //
 212             .expectError(DatafileTaskException.class) //
 213             .verify();
 214
 215         verify(messageRouterSubscriber, times(1))
 216                 .getElements(dmaapConsumerConfiguration.getMessageRouterSubscribeRequest());
 217     }
 218
 219     @Test
 220     public void whenFtpes_ReturnsCorrectResponse() {
 221         prepareMocksForDmaapConsumer(Optional.of(ftpesMessageJson), expectedFtpesMessage);
 222
 223         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
 224             .expectNext(expectedFtpesMessage) //
 225             .verifyComplete();
 226
 227
 228
 229         verify(messageRouterSubscriber, times(1))
 230                 .getElements(dmaapConsumerConfiguration.getMessageRouterSubscribeRequest());
 231         verifyNoMoreInteractions(messageRouterSubscriber);
 232     }
 233
 234     @Test
 235     public void whenSftp_ReturnsCorrectResponse() {
 236         prepareMocksForDmaapConsumer(Optional.of(sftpMessageJson), expectedSftpMessage);
 237
 238         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
 239             .expectNext(expectedSftpMessage) //
 240             .verifyComplete();
 241
 242         verify(messageRouterSubscriber, times(1))
 243                 .getElements(dmaapConsumerConfiguration.getMessageRouterSubscribeRequest());
 244         verifyNoMoreInteractions(messageRouterSubscriber);
 245     }
 246
 247     private void prepareMocksForDmaapConsumer(Optional<JsonElement> message,
 248         FileReadyMessage fileReadyMessageAfterConsume) {
 249         Flux<JsonElement> messageAsMono = message.isPresent() ? Flux.just(message.get()) : Flux.empty();
 250
 251         messageRouterSubscriber = mock(MessageRouterSubscriber.class);
 252         dmaapConsumerConfiguration = new ConsumerConfiguration(mock(MessageRouterSubscriberConfig.class),
 253                 messageRouterSubscriber, mock(MessageRouterSubscribeRequest.class));
 254
 255         JsonMessageParser jsonMessageParserMock = mock(JsonMessageParser.class);
 256         when(messageRouterSubscriber.getElements(dmaapConsumerConfiguration.getMessageRouterSubscribeRequest()))
 257                 .thenReturn(messageAsMono);
 258         when(appConfig.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
 259
 260         if (message.isPresent()) {
 261             when(jsonMessageParserMock.getMessagesFromJson(any())).thenReturn(Flux.just(fileReadyMessageAfterConsume));
 262         } else {
 263             when(jsonMessageParserMock.getMessagesFromJson(any()))
 264                 .thenReturn(Flux.error(new DatafileTaskException("problemas")));
 265         }
 266
 267         messageConsumer = spy(new DMaaPMessageConsumer(appConfig, jsonMessageParserMock));
 268     }
 269
 270 }