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

   1 /*-
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 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.junit.jupiter.api.Assertions.assertEquals;
  20 import static org.junit.jupiter.api.Assertions.assertTrue;
  21 import static org.mockito.ArgumentMatchers.any;
  22 import static org.mockito.Mockito.doReturn;
  23 import static org.mockito.Mockito.mock;
  24 import static org.mockito.Mockito.spy;
  25 import static org.mockito.Mockito.times;
  26 import static org.mockito.Mockito.verify;
  27 import static org.mockito.Mockito.verifyNoMoreInteractions;
  28 import static org.mockito.Mockito.when;
  29
  30 import java.io.ByteArrayInputStream;
  31 import java.io.InputStream;
  32 import java.net.URI;
  33 import java.nio.file.Path;
  34 import java.nio.file.Paths;
  35 import java.time.Duration;
  36 import java.util.ArrayList;
  37 import java.util.HashMap;
  38 import java.util.List;
  39 import java.util.Map;
  40
  41 import org.apache.http.Header;
  42 import org.apache.http.HttpResponse;
  43 import org.apache.http.StatusLine;
  44 import org.apache.http.client.methods.HttpPut;
  45 import org.apache.http.client.methods.HttpUriRequest;
  46 import org.junit.jupiter.api.BeforeAll;
  47 import org.junit.jupiter.api.Test;
  48 import org.mockito.ArgumentCaptor;
  49 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
  50 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
  51 import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel;
  52 import org.onap.dcaegen2.collectors.datafile.model.ImmutableConsumerDmaapModel;
  53 import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerReactiveHttpClient;
  54 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
  55 import org.springframework.http.HttpStatus;
  56 import org.springframework.web.util.DefaultUriBuilderFactory;
  57 import org.springframework.web.util.UriBuilder;
  58
  59 import reactor.test.StepVerifier;
  60
  61 /**
  62  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/17/18
  63  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  64  */
  65 class DataRouterPublisherTest {
  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_FILE_NAME = "A20161224.1030-1045.bin.gz";
  73     private static final String FTPES_ADDRESS = "ftpes://192.168.0.101:22/ftp/rop/" + PM_FILE_NAME;
  74     private static final String LOCAL_FILE_NAME = SOURCE_NAME + "_" + PM_FILE_NAME;
  75
  76     private static final String COMPRESSION = "gzip";
  77     private static final String FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
  78     private static final String FILE_FORMAT_VERSION = "V10";
  79     private static final String X_DMAAP_DR_META = "X-DMAAP-DR-META";
  80
  81     private static final String HOST = "54.45.33.2";
  82     private static final String HTTPS_SCHEME = "https";
  83     private static final int PORT = 1234;
  84     private static final String APPLICATION_OCTET_STREAM_CONTENT_TYPE = "application/octet-stream";
  85     private static final String PUBLISH_TOPIC = "publish";
  86     private static final String FEED_ID = "1";
  87     private static final String FILE_CONTENT = "Just a string.";
  88
  89     private static ConsumerDmaapModel consumerDmaapModel;
  90     private static DmaapProducerReactiveHttpClient httpClientMock;
  91     private static AppConfig appConfig;
  92     private static DmaapPublisherConfiguration publisherConfigurationMock = mock(DmaapPublisherConfiguration.class);
  93     private final Map<String, String> contextMap = new HashMap<>();
  94     private static DataRouterPublisher publisherTaskUnderTestSpy;
  95
  96     /**
  97      * Sets up data for tests.
  98      */
  99     @BeforeAll
 100     public static void setUp() {
 101         when(publisherConfigurationMock.dmaapHostName()).thenReturn(HOST);
 102         when(publisherConfigurationMock.dmaapProtocol()).thenReturn(HTTPS_SCHEME);
 103         when(publisherConfigurationMock.dmaapPortNumber()).thenReturn(PORT);
 104
 105         consumerDmaapModel = ImmutableConsumerDmaapModel.builder() //
 106                 .productName(PRODUCT_NAME) //
 107                 .vendorName(VENDOR_NAME) //
 108                 .lastEpochMicrosec(LAST_EPOCH_MICROSEC) //
 109                 .sourceName(SOURCE_NAME) //
 110                 .startEpochMicrosec(START_EPOCH_MICROSEC) //
 111                 .timeZoneOffset(TIME_ZONE_OFFSET) //
 112                 .name(PM_FILE_NAME) //
 113                 .location(FTPES_ADDRESS) //
 114                 .internalLocation(Paths.get("target/" + LOCAL_FILE_NAME)) //
 115                 .compression("gzip") //
 116                 .fileFormatType(FILE_FORMAT_TYPE) //
 117                 .fileFormatVersion(FILE_FORMAT_VERSION) //
 118                 .build(); //
 119         appConfig = mock(AppConfig.class);
 120         publisherTaskUnderTestSpy = spy(new DataRouterPublisher(appConfig));
 121     }
 122
 123     @Test
 124     public void whenPassedObjectFits_ReturnsCorrectStatus() throws Exception {
 125         prepareMocksForTests(null, Integer.valueOf(HttpStatus.OK.value()));
 126         StepVerifier
 127                 .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
 128                 .expectNext(consumerDmaapModel) //
 129                 .verifyComplete();
 130
 131         ArgumentCaptor<HttpUriRequest> requestCaptor = ArgumentCaptor.forClass(HttpUriRequest.class);
 132         verify(httpClientMock).getBaseUri();
 133         verify(httpClientMock).addUserCredentialsToHead(any(HttpUriRequest.class));
 134         verify(httpClientMock).getDmaapProducerResponseWithRedirect(requestCaptor.capture(), any());
 135         verifyNoMoreInteractions(httpClientMock);
 136
 137         HttpPut actualPut = (HttpPut) requestCaptor.getValue();
 138         URI actualUri = actualPut.getURI();
 139         assertEquals(HTTPS_SCHEME, actualUri.getScheme());
 140         assertEquals(HOST, actualUri.getHost());
 141         assertEquals(PORT, actualUri.getPort());
 142         Path actualPath = Paths.get(actualUri.getPath());
 143         assertTrue(PUBLISH_TOPIC.equals(actualPath.getName(0).toString()));
 144         assertTrue(FEED_ID.equals(actualPath.getName(1).toString()));
 145         assertTrue(LOCAL_FILE_NAME.equals(actualPath.getName(2).toString()));
 146
 147         Header[] contentHeaders = actualPut.getHeaders("content-type");
 148         assertEquals(APPLICATION_OCTET_STREAM_CONTENT_TYPE, contentHeaders[0].getValue());
 149
 150         Header[] metaHeaders = actualPut.getHeaders(X_DMAAP_DR_META);
 151         Map<String, String> metaHash = getMetaDataAsMap(metaHeaders);
 152         assertTrue(10 == metaHash.size());
 153         assertEquals(PRODUCT_NAME, metaHash.get("productName"));
 154         assertEquals(VENDOR_NAME, metaHash.get("vendorName"));
 155         assertEquals(LAST_EPOCH_MICROSEC, metaHash.get("lastEpochMicrosec"));
 156         assertEquals(SOURCE_NAME, metaHash.get("sourceName"));
 157         assertEquals(START_EPOCH_MICROSEC, metaHash.get("startEpochMicrosec"));
 158         assertEquals(TIME_ZONE_OFFSET, metaHash.get("timeZoneOffset"));
 159         assertEquals(COMPRESSION, metaHash.get("compression"));
 160         assertEquals(FTPES_ADDRESS, metaHash.get("location"));
 161         assertEquals(FILE_FORMAT_TYPE, metaHash.get("fileFormatType"));
 162         assertEquals(FILE_FORMAT_VERSION, metaHash.get("fileFormatVersion"));
 163     }
 164
 165     @Test
 166     void whenPassedObjectFits_firstFailsWithExceptionThenSucceeds() throws Exception {
 167         prepareMocksForTests(new DatafileTaskException("Error"), HttpStatus.OK.value());
 168
 169         StepVerifier
 170                 .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 2, Duration.ofSeconds(0), contextMap))
 171                 .expectNext(consumerDmaapModel) //
 172                 .verifyComplete();
 173     }
 174
 175     @Test
 176     public void whenPassedObjectFits_firstFailsThenSucceeds() throws Exception {
 177         prepareMocksForTests(null, Integer.valueOf(HttpStatus.BAD_GATEWAY.value()),
 178                 Integer.valueOf(HttpStatus.OK.value()));
 179
 180         StepVerifier
 181                 .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
 182                 .expectNext(consumerDmaapModel) //
 183                 .verifyComplete();
 184
 185         verify(httpClientMock, times(2)).getBaseUri();
 186         verify(httpClientMock, times(2)).addUserCredentialsToHead(any(HttpUriRequest.class));
 187         verify(httpClientMock, times(2)).getDmaapProducerResponseWithRedirect(any(HttpUriRequest.class), any());
 188         verifyNoMoreInteractions(httpClientMock);
 189     }
 190
 191     @Test
 192     public void whenPassedObjectFits_firstFailsThenFails() throws Exception {
 193         prepareMocksForTests(null, Integer.valueOf(HttpStatus.BAD_GATEWAY.value()),
 194                 Integer.valueOf((HttpStatus.BAD_GATEWAY.value())));
 195
 196         StepVerifier
 197                 .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap))
 198                 .expectErrorMessage("Retries exhausted: 1/1") //
 199                 .verify();
 200
 201         verify(httpClientMock, times(2)).getBaseUri();
 202         verify(httpClientMock, times(2)).addUserCredentialsToHead(any(HttpUriRequest.class));
 203         verify(httpClientMock, times(2)).getDmaapProducerResponseWithRedirect(any(HttpUriRequest.class), any());
 204         verifyNoMoreInteractions(httpClientMock);
 205     }
 206
 207     @SafeVarargs
 208     final void prepareMocksForTests(Exception exception, Integer firstResponse, Integer... nextHttpResponses)
 209             throws Exception {
 210         httpClientMock = mock(DmaapProducerReactiveHttpClient.class);
 211         when(appConfig.getDmaapPublisherConfiguration()).thenReturn(publisherConfigurationMock);
 212         doReturn(publisherConfigurationMock).when(publisherTaskUnderTestSpy).resolveConfiguration();
 213         doReturn(httpClientMock).when(publisherTaskUnderTestSpy).resolveClient();
 214
 215         UriBuilder uriBuilder = new DefaultUriBuilderFactory().builder().scheme(HTTPS_SCHEME).host(HOST).port(PORT);
 216         when(httpClientMock.getBaseUri()).thenReturn(uriBuilder);
 217
 218         HttpResponse httpResponseMock = mock(HttpResponse.class);
 219         if (exception == null) {
 220             when(httpClientMock.getDmaapProducerResponseWithRedirect(any(HttpUriRequest.class), any()))
 221                     .thenReturn(httpResponseMock);
 222         } else {
 223             when(httpClientMock.getDmaapProducerResponseWithRedirect(any(HttpUriRequest.class), any()))
 224                     .thenThrow(exception).thenReturn(httpResponseMock);
 225         }
 226         StatusLine statusLineMock = mock(StatusLine.class);
 227         when(httpResponseMock.getStatusLine()).thenReturn(statusLineMock);
 228         when(statusLineMock.getStatusCode()).thenReturn(firstResponse, nextHttpResponses);
 229
 230         InputStream fileStream = new ByteArrayInputStream(FILE_CONTENT.getBytes());
 231         doReturn(fileStream).when(publisherTaskUnderTestSpy).createInputStream(Paths.get("target", LOCAL_FILE_NAME));
 232     }
 233
 234     private Map<String, String> getMetaDataAsMap(Header[] metaHeaders) {
 235         Map<String, String> metaHash = new HashMap<>();
 236         String actualMetaData = metaHeaders[0].getValue();
 237         actualMetaData = actualMetaData.substring(1, actualMetaData.length() - 1);
 238         actualMetaData = actualMetaData.replace("\"", "");
 239         String[] commaSplitedMetaData = actualMetaData.split(",");
 240         for (int i = 0; i < commaSplitedMetaData.length; i++) {
 241             String[] keyValuePair = commaSplitedMetaData[i].split(":");
 242             if (keyValuePair.length > 2) {
 243                 List<String> arrayKeyValuePair = new ArrayList<>(keyValuePair.length);
 244                 for (int j = 1; j < keyValuePair.length; j++) {
 245                     arrayKeyValuePair.add(keyValuePair[j]);
 246                 }
 247                 keyValuePair[1] = String.join(":", arrayKeyValuePair);
 248             }
 249             metaHash.put(keyValuePair[0], keyValuePair[1]);
 250         }
 251         return metaHash;
 252     }
 253 }