2 * ============LICENSE_START====================================
3 * DCAEGEN2-SERVICES-SDK
4 * =========================================================
5 * Copyright (C) 2019 Nokia. All rights reserved.
6 * =========================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=====================================
21 package org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl;
23 import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.DummyHttpServer.sendResource;
24 import static org.onap.dcaegen2.services.sdk.rest.services.cbs.client.impl.DummyHttpServer.sendString;
26 import com.google.gson.JsonObject;
27 import io.vavr.collection.Stream;
28 import java.time.Duration;
29 import org.junit.jupiter.api.AfterAll;
30 import org.junit.jupiter.api.BeforeAll;
31 import org.junit.jupiter.api.Test;
32 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
33 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
34 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
35 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.EnvProperties;
36 import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.ImmutableEnvProperties;
37 import reactor.core.publisher.Flux;
38 import reactor.core.publisher.Mono;
39 import reactor.test.StepVerifier;
42 * @author <a href="mailto:piotr.jaszczyk@nokia.com">Piotr Jaszczyk</a>
43 * @since February 2019
45 class CbsClientImplIT {
47 private static final String CONSUL_RESPONSE = "[\n"
49 + " \"ServiceAddress\": \"HOST\",\n"
50 + " \"ServiceName\": \"the_cbs\",\n"
51 + " \"ServicePort\": PORT\n"
54 private static final String SAMPLE_CONFIG = "/sample_config.json";
55 private static final String SAMPLE_CONFIG_KEY = "keystore.path";
56 private static final String EXPECTED_CONFIG_VALUE = "/var/run/security/keystore.p12";
57 private static EnvProperties sampleEnvironment;
58 private static DummyHttpServer server;
62 server = DummyHttpServer.start(routes ->
63 routes.get("/v1/catalog/service/the_cbs", (req, resp) -> sendString(resp, lazyConsulResponse()))
64 .get("/service_component/dcae-component", (req, resp) -> sendResource(resp, SAMPLE_CONFIG)));
65 sampleEnvironment = ImmutableEnvProperties.builder()
66 .appName("dcae-component")
68 .consulHost(server.host())
69 .consulPort(server.port())
74 static void tearDown() {
79 void testCbsClientWithSingleCall() {
81 final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
82 final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
85 final Mono<JsonObject> result = sut.flatMap(cbsClient -> cbsClient.get(diagnosticContext));
88 StepVerifier.create(result.map(this::sampleConfigValue))
89 .expectNext(EXPECTED_CONFIG_VALUE)
91 .verify(Duration.ofSeconds(5));
95 void testCbsClientWithPeriodicCall() {
97 final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
98 final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
101 final Flux<JsonObject> result = sut
102 .flatMapMany(cbsClient -> cbsClient.get(diagnosticContext, Duration.ZERO, Duration.ofMillis(10)));
105 final int itemsToTake = 5;
106 StepVerifier.create(result.take(itemsToTake).map(this::sampleConfigValue))
107 .expectNextSequence(Stream.of(EXPECTED_CONFIG_VALUE).cycle(itemsToTake))
109 .verify(Duration.ofSeconds(5));
113 void testCbsClientWithUpdatesCall() {
115 final Mono<CbsClient> sut = CbsClientFactory.createCbsClient(sampleEnvironment);
116 final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
117 final Duration period = Duration.ofMillis(10);
120 final Flux<JsonObject> result = sut
121 .flatMapMany(cbsClient -> cbsClient.updates(diagnosticContext, Duration.ZERO, period));
124 final Duration timeToCollectItemsFor = period.multipliedBy(50);
125 StepVerifier.create(result.take(timeToCollectItemsFor).map(this::sampleConfigValue))
126 .expectNext(EXPECTED_CONFIG_VALUE)
128 .verify(Duration.ofSeconds(5));
131 private String sampleConfigValue(JsonObject obj) {
132 return obj.get(SAMPLE_CONFIG_KEY).getAsString();
135 private static Mono<String> lazyConsulResponse() {
136 return Mono.just(CONSUL_RESPONSE)
137 .map(CbsClientImplIT::processConsulResponseTemplate);
140 private static String processConsulResponseTemplate(String resp) {
141 return resp.replaceAll("HOST", server.host())
142 .replaceAll("PORT", Integer.toString(server.port()));