43502b494e243325e3deec4362056bfb3262fbf1
[dcaegen2/collectors/datafile.git] /
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 import org.onap.dcaegen2.collectors.datafile.config.DmaapConsumerConfiguration;
33 import org.onap.dcaegen2.collectors.datafile.config.ImmutableDmaapConsumerConfiguration;
34 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
35 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
36 import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapEmptyResponseException;
37 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
38 import org.onap.dcaegen2.collectors.datafile.model.FileData;
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.service.DmaapConsumerJsonParser;
42 import org.onap.dcaegen2.collectors.datafile.service.consumer.DmaapConsumerReactiveHttpClient;
43 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
44 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
45
46 import reactor.core.publisher.Flux;
47 import reactor.core.publisher.Mono;
48 import reactor.test.StepVerifier;
49
50 /**
51  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
52  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
53  */
54 class DmaapConsumerTaskImplTest {
55     private static final String PM_MEAS_CHANGE_IDENTIFIER = "PM_MEAS_FILES";
56     private static final String FILE_READY_CHANGE_TYPE = "FileReady";
57     private static final String FTPES_SCHEME = "ftpes://";
58     private static final String SFTP_SCHEME = "sftp://";
59     private static final String SERVER_ADDRESS = "192.168.0.101";
60     private static final String PORT_22 = "22";
61     private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz";
62     private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME;
63     private static final String LOCAL_FILE_LOCATION = "target/" + PM_FILE_NAME;
64     private static final String FTPES_LOCATION = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
65     private static final String SFTP_LOCATION = SFTP_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION;
66     private static final String GZIP_COMPRESSION = "gzip";
67     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
68     private static final String FILE_FORMAT_VERSION = "V10";
69
70     private static List<ConsumerDmaapModel> listOfConsumerDmaapModel = new ArrayList<ConsumerDmaapModel>();
71
72     private static AppConfig appConfig;
73     private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
74     private DmaapConsumerTaskImpl dmaapConsumerTask;
75     private DmaapConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient;
76
77     private static String ftpesMessage;
78     private static FileData ftpesFileData;
79
80     private static String sftpMessage;
81     private static FileData sftpFileData;
82
83     @BeforeAll
84     public static void setUp() {
85         dmaapConsumerConfiguration = new ImmutableDmaapConsumerConfiguration.Builder().consumerGroup("OpenDCAE-c12")
86                 .consumerId("c12").dmaapContentType("application/json").dmaapHostName("54.45.33.2")
87                 .dmaapPortNumber(1234).dmaapProtocol("https").dmaapUserName("Datafile").dmaapUserPassword("Datafile")
88                 .dmaapTopicName("unauthenticated.NOTIFICATION").timeoutMS(-1).messageLimit(-1).build();
89
90         appConfig = mock(AppConfig.class);
91
92         AdditionalField ftpesAdditionalField =
93                 new JsonMessage.AdditionalFieldBuilder().location(FTPES_LOCATION).compression(GZIP_COMPRESSION)
94                         .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
95         JsonMessage ftpesJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
96                 .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
97                 .addAdditionalField(ftpesAdditionalField).build();
98         ftpesMessage = ftpesJsonMessage.toString();
99         ftpesFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
100                 .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(FTPES_LOCATION)
101                 .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
102                 .fileFormatVersion(FILE_FORMAT_VERSION).build();
103
104         AdditionalField sftpAdditionalField =
105                 new JsonMessage.AdditionalFieldBuilder().location(SFTP_LOCATION).compression(GZIP_COMPRESSION)
106                         .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
107         JsonMessage sftpJsonMessage = new JsonMessage.JsonMessageBuilder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
108                 .changeType(FILE_READY_CHANGE_TYPE).notificationFieldsVersion("1.0")
109                 .addAdditionalField(sftpAdditionalField).build();
110         sftpMessage = sftpJsonMessage.toString();
111         sftpFileData = ImmutableFileData.builder().changeIdentifier(PM_MEAS_CHANGE_IDENTIFIER)
112                 .changeType(FILE_READY_CHANGE_TYPE).name(PM_FILE_NAME).location(SFTP_LOCATION)
113                 .compression(GZIP_COMPRESSION).fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE)
114                 .fileFormatVersion(FILE_FORMAT_VERSION).build();
115
116
117         ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder().name(PM_FILE_NAME)
118                 .location(LOCAL_FILE_LOCATION).compression(GZIP_COMPRESSION)
119                 .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE).fileFormatVersion(FILE_FORMAT_VERSION).build();
120         listOfConsumerDmaapModel.add(consumerDmaapModel);
121     }
122
123     @Test
124     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
125         prepareMocksForDmaapConsumer("", null);
126
127         StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
128                 .expectError(DmaapEmptyResponseException.class).verify();
129
130         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
131     }
132
133     @Test
134     public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
135         prepareMocksForDmaapConsumer(ftpesMessage, ftpesFileData);
136
137         StepVerifier.create(dmaapConsumerTask.execute(ftpesMessage)).expectNext(ftpesFileData).verifyComplete();
138
139         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
140         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
141     }
142
143     @Test
144     public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
145         prepareMocksForDmaapConsumer(sftpMessage, sftpFileData);
146
147         StepVerifier.create(dmaapConsumerTask.execute(ftpesMessage)).expectNext(sftpFileData).verifyComplete();
148
149         verify(dmaapConsumerReactiveHttpClient, times(1)).getDmaapConsumerResponse();
150         verifyNoMoreInteractions(dmaapConsumerReactiveHttpClient);
151     }
152
153     private void prepareMocksForDmaapConsumer(String message, FileData fileDataAfterConsume) {
154         Mono<String> messageAsMono = Mono.just(message);
155         DmaapConsumerJsonParser dmaapConsumerJsonParserMock = mock(DmaapConsumerJsonParser.class);
156         dmaapConsumerReactiveHttpClient = mock(DmaapConsumerReactiveHttpClient.class);
157         when(dmaapConsumerReactiveHttpClient.getDmaapConsumerResponse()).thenReturn(messageAsMono);
158
159         if (!message.isEmpty()) {
160             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono)).thenReturn(Flux.just(fileDataAfterConsume));
161         } else {
162             when(dmaapConsumerJsonParserMock.getJsonObject(messageAsMono))
163                     .thenReturn(Flux.error(new DmaapEmptyResponseException()));
164         }
165
166         dmaapConsumerTask =
167                 spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerReactiveHttpClient, dmaapConsumerJsonParserMock));
168         when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
169         doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
170     }
171 }