X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=dmaap-adaptor-java%2Fsrc%2Ftest%2Fjava%2Forg%2Foran%2Fdmaapadapter%2FApplicationTest.java;h=66601750805f9188422d96eb7b20555bb692f128;hb=5ce0ce048084ad1504292a4315a0ba7446ee42bc;hp=2a82ff8fe862c499a2c988963b99142809b5aa90;hpb=e702c19d34c342e75993e169c261f9087cc0643f;p=nonrtric.git diff --git a/dmaap-adaptor-java/src/test/java/org/oran/dmaapadapter/ApplicationTest.java b/dmaap-adaptor-java/src/test/java/org/oran/dmaapadapter/ApplicationTest.java index 2a82ff8f..66601750 100644 --- a/dmaap-adaptor-java/src/test/java/org/oran/dmaapadapter/ApplicationTest.java +++ b/dmaap-adaptor-java/src/test/java/org/oran/dmaapadapter/ApplicationTest.java @@ -22,9 +22,8 @@ package org.oran.dmaapadapter; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertTrue; -import com.google.gson.Gson; -import com.google.gson.GsonBuilder; import com.google.gson.JsonParser; import java.io.FileOutputStream; @@ -35,6 +34,7 @@ import java.nio.file.Paths; import org.json.JSONObject; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.oran.dmaapadapter.clients.AsyncRestClient; @@ -44,13 +44,15 @@ import org.oran.dmaapadapter.configuration.ImmutableHttpProxyConfig; import org.oran.dmaapadapter.configuration.ImmutableWebClientConfig; import org.oran.dmaapadapter.configuration.WebClientConfig; import org.oran.dmaapadapter.configuration.WebClientConfig.HttpProxyConfig; +import org.oran.dmaapadapter.controllers.ProducerCallbacksController; import org.oran.dmaapadapter.r1.ConsumerJobInfo; -import org.oran.dmaapadapter.repository.InfoType; +import org.oran.dmaapadapter.r1.ProducerJobInfo; import org.oran.dmaapadapter.repository.InfoTypes; +import org.oran.dmaapadapter.repository.Job; import org.oran.dmaapadapter.repository.Jobs; +import org.oran.dmaapadapter.tasks.KafkaJobDataConsumer; +import org.oran.dmaapadapter.tasks.KafkaTopicConsumers; import org.oran.dmaapadapter.tasks.ProducerRegstrationTask; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.context.SpringBootTest.WebEnvironment; @@ -60,26 +62,28 @@ import org.springframework.boot.web.server.LocalServerPort; import org.springframework.boot.web.servlet.server.ServletWebServerFactory; import org.springframework.context.annotation.Bean; import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit.jupiter.SpringExtension; +import org.springframework.web.reactive.function.client.WebClientResponseException; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; @ExtendWith(SpringExtension.class) -@SpringBootTest(webEnvironment = WebEnvironment.DEFINED_PORT) +@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT) @TestPropertySource(properties = { // "server.ssl.key-store=./config/keystore.jks", // "app.webclient.trust-store=./config/truststore.jks", // "app.configuration-filepath=./src/test/resources/test_application_configuration.json"// }) class ApplicationTest { - private static final Logger logger = LoggerFactory.getLogger(ApplicationTest.class); @Autowired private ApplicationConfig applicationConfig; - @Autowired - private ProducerRegstrationTask producerRegstrationTask; - @Autowired private Jobs jobs; @@ -90,16 +94,22 @@ class ApplicationTest { private ConsumerController consumerController; @Autowired - private EcsSimulatorController ecsSimulatorController; + private IcsSimulatorController icsSimulatorController; + + @Autowired + KafkaTopicConsumers kafkaTopicConsumers; + + @Autowired + ProducerRegstrationTask producerRegistrationTask; + + private com.google.gson.Gson gson = new com.google.gson.GsonBuilder().create(); @LocalServerPort int localServerHttpPort; - private static Gson gson = new GsonBuilder().create(); - static class TestApplicationConfig extends ApplicationConfig { @Override - public String getEcsBaseUrl() { + public String getIcsBaseUrl() { return thisProcessUrl(); } @@ -139,12 +149,16 @@ class ApplicationTest { } } + @BeforeEach + void setPort() { + this.applicationConfig.setLocalServerHttpPort(this.localServerHttpPort); + } + @AfterEach void reset() { this.consumerController.testResults.reset(); - this.ecsSimulatorController.testResults.reset(); + this.icsSimulatorController.testResults.reset(); this.jobs.clear(); - this.types.clear(); } private AsyncRestClient restClient(boolean useTrustValidation) { @@ -176,8 +190,7 @@ class ApplicationTest { } private ConsumerJobInfo consumerJobInfo() { - InfoType type = this.types.getAll().iterator().next(); - return consumerJobInfo(type.getId(), "EI_JOB_ID"); + return consumerJobInfo("DmaapInformationType", "EI_JOB_ID"); } private Object jsonObject() { @@ -218,20 +231,130 @@ class ApplicationTest { } @Test - void testWholeChain() throws Exception { - await().untilAsserted(() -> assertThat(producerRegstrationTask.isRegisteredInEcs()).isTrue()); + void testResponseCodes() throws Exception { + String supervisionUrl = baseUrl() + ProducerCallbacksController.SUPERVISION_URL; + ResponseEntity resp = restClient().getForEntity(supervisionUrl).block(); + assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK); + + String jobUrl = baseUrl() + ProducerCallbacksController.JOB_URL; + resp = restClient().deleteForEntity(jobUrl + "/junk").block(); + assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK); + + ProducerJobInfo info = new ProducerJobInfo(null, "id", "typeId", "targetUri", "owner", "lastUpdated"); + String body = gson.toJson(info); + testErrorCode(restClient().post(jobUrl, body, MediaType.APPLICATION_JSON), HttpStatus.NOT_FOUND, + "Could not find type"); + } + + @Test + void testReceiveAndPostDataFromKafka() { + final String JOB_ID = "ID"; + final String TYPE_ID = "KafkaInformationType"; + await().untilAsserted(() -> assertThat(icsSimulatorController.testResults.registrationInfo).isNotNull()); + assertThat(icsSimulatorController.testResults.registrationInfo.supportedTypeIds).hasSize(this.types.size()); + + // Create a job + Job.Parameters param = new Job.Parameters("", new Job.BufferTimeout(123, 456), 1); + String targetUri = baseUrl() + ConsumerController.CONSUMER_TARGET_URL; + ConsumerJobInfo kafkaJobInfo = + new ConsumerJobInfo(TYPE_ID, jsonObject(gson.toJson(param)), "owner", targetUri, ""); + + this.icsSimulatorController.addJob(kafkaJobInfo, JOB_ID, restClient()); + await().untilAsserted(() -> assertThat(this.jobs.size()).isEqualTo(1)); + + KafkaJobDataConsumer kafkaConsumer = this.kafkaTopicConsumers.getConsumers().get(TYPE_ID, JOB_ID); + + // Handle received data from Kafka, check that it has been posted to the + // consumer + kafkaConsumer.start(Flux.just("data")); + + ConsumerController.TestResults consumer = this.consumerController.testResults; + await().untilAsserted(() -> assertThat(consumer.receivedBodies.size()).isEqualTo(1)); + assertThat(consumer.receivedBodies.get(0)).isEqualTo("[\"data\"]"); + + // Test send an exception + kafkaConsumer.start(Flux.error(new NullPointerException())); + + // Test regular restart of stopped + kafkaConsumer.stop(); + this.kafkaTopicConsumers.restartNonRunningTopics(); + await().untilAsserted(() -> assertThat(kafkaConsumer.isRunning()).isTrue()); + + // Delete the job + this.icsSimulatorController.deleteJob(JOB_ID, restClient()); + await().untilAsserted(() -> assertThat(this.jobs.size()).isZero()); + } + + @Test + void testReceiveAndPostDataFromDmaap() throws Exception { + final String JOB_ID = "ID"; - this.ecsSimulatorController.addJob(consumerJobInfo(), restClient()); + // Register producer, Register types + await().untilAsserted(() -> assertThat(icsSimulatorController.testResults.registrationInfo).isNotNull()); + assertThat(icsSimulatorController.testResults.registrationInfo.supportedTypeIds).hasSize(this.types.size()); + assertThat(producerRegistrationTask.isRegisteredInIcs()).isTrue(); + producerRegistrationTask.supervisionTask().block(); + // Create a job + this.icsSimulatorController.addJob(consumerJobInfo(), JOB_ID, restClient()); await().untilAsserted(() -> assertThat(this.jobs.size()).isEqualTo(1)); + // Return two messages from DMAAP and verify that these are sent to the owner of + // the job (consumer) DmaapSimulatorController.dmaapResponses.add("DmaapResponse1"); DmaapSimulatorController.dmaapResponses.add("DmaapResponse2"); - ConsumerController.TestResults consumer = this.consumerController.testResults; await().untilAsserted(() -> assertThat(consumer.receivedBodies.size()).isEqualTo(2)); assertThat(consumer.receivedBodies.get(0)).isEqualTo("DmaapResponse1"); + String jobUrl = baseUrl() + ProducerCallbacksController.JOB_URL; + String jobs = restClient().get(jobUrl).block(); + assertThat(jobs).contains(JOB_ID); + + // Delete the job + this.icsSimulatorController.deleteJob(JOB_ID, restClient()); + await().untilAsserted(() -> assertThat(this.jobs.size()).isZero()); + } + + @Test + void testReRegister() throws Exception { + // Wait foir register types and producer + await().untilAsserted(() -> assertThat(icsSimulatorController.testResults.registrationInfo).isNotNull()); + assertThat(icsSimulatorController.testResults.registrationInfo.supportedTypeIds).hasSize(this.types.size()); + + // Clear the registration, should trigger a re-register + icsSimulatorController.testResults.reset(); + await().untilAsserted(() -> assertThat(icsSimulatorController.testResults.registrationInfo).isNotNull()); + assertThat(icsSimulatorController.testResults.registrationInfo.supportedTypeIds).hasSize(this.types.size()); + + // Just clear the registerred types, should trigger a re-register + icsSimulatorController.testResults.types.clear(); + await().untilAsserted( + () -> assertThat(icsSimulatorController.testResults.registrationInfo.supportedTypeIds).hasSize(2)); + } + + public static void testErrorCode(Mono request, HttpStatus expStatus, String responseContains) { + testErrorCode(request, expStatus, responseContains, true); + } + + public static void testErrorCode(Mono request, HttpStatus expStatus, String responseContains, + boolean expectApplicationProblemJsonMediaType) { + StepVerifier.create(request) // + .expectSubscription() // + .expectErrorMatches( + t -> checkWebClientError(t, expStatus, responseContains, expectApplicationProblemJsonMediaType)) // + .verify(); } + private static boolean checkWebClientError(Throwable throwable, HttpStatus expStatus, String responseContains, + boolean expectApplicationProblemJsonMediaType) { + assertTrue(throwable instanceof WebClientResponseException); + WebClientResponseException responseException = (WebClientResponseException) throwable; + assertThat(responseException.getStatusCode()).isEqualTo(expStatus); + assertThat(responseException.getResponseBodyAsString()).contains(responseContains); + if (expectApplicationProblemJsonMediaType) { + assertThat(responseException.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_PROBLEM_JSON); + } + return true; + } }