- move sdc-distribution-client instantiation out of the ModelController into a separate class
- add integration test with embedded kafka (not fully implemented)
Issue-ID: AAI-3818
Change-Id: I0b5dd118d9145372ddf123319b58829d0ef9275a
Signed-off-by: Fiete Ostkamp <Fiete.Ostkamp@telekom.de>
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-aop</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
- <version>3.3.1</version>
+ <!-- <version>3.3.1</version> -->
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.kafka</groupId>
+ <artifactId>spring-kafka</artifactId>
+ <!-- <version>3.1.2</version> -->
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.kafka</groupId>
+ <artifactId>spring-kafka-test</artifactId>
+ <!-- <version>3.1.1</version> -->
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.onap.aai</groupId>
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom AG Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.modelloader.config;
+
+import java.util.Date;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import org.onap.aai.cl.api.Logger;
+import org.onap.aai.cl.eelf.LoggerFactory;
+import org.onap.aai.modelloader.notification.EventCallback;
+import org.onap.aai.modelloader.service.ModelLoaderMsgs;
+import org.onap.aai.modelloader.service.SdcConnectionJob;
+import org.onap.sdc.api.IDistributionClient;
+import org.onap.sdc.api.results.IDistributionClientResult;
+import org.onap.sdc.utils.DistributionActionResultEnum;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@ConditionalOnProperty(value = "ml.distribution.connection.enabled", havingValue = "true", matchIfMissing = true)
+public class DistributionClientStartupConfig {
+
+ private static final Logger logger = LoggerFactory.getInstance().getLogger(DistributionClientStartupConfig.class);
+
+ private final IDistributionClient client;
+ private final ModelLoaderConfig config;
+ private final EventCallback eventCallback;
+
+ public DistributionClientStartupConfig(IDistributionClient client, ModelLoaderConfig config,
+ EventCallback eventCallback) {
+ this.client = client;
+ this.config = config;
+ this.eventCallback = eventCallback;
+ }
+
+ @EventListener(ApplicationReadyEvent.class)
+ protected void initSdcClient() {
+ // Initialize distribution client
+ logger.debug(ModelLoaderMsgs.INITIALIZING, "Initializing distribution client...");
+ IDistributionClientResult initResult = client.init(config, eventCallback);
+
+ if (initResult.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
+ // Start distribution client
+ logger.debug(ModelLoaderMsgs.INITIALIZING, "Starting distribution client...");
+ IDistributionClientResult startResult = client.start();
+ if (startResult.getDistributionActionResult() == DistributionActionResultEnum.SUCCESS) {
+ logger.info(ModelLoaderMsgs.INITIALIZING, "Connection to SDC established");
+ } else {
+ String errorMsg = "Failed to start distribution client: " + startResult.getDistributionMessageResult();
+ logger.error(ModelLoaderMsgs.ASDC_CONNECTION_ERROR, errorMsg);
+
+ // Kick off a timer to retry the SDC connection
+ Timer timer = new Timer();
+ TimerTask task = new SdcConnectionJob(client, config, eventCallback, timer);
+ timer.schedule(task, new Date(), 60000);
+ }
+ } else {
+ String errorMsg = "Failed to initialize distribution client: " + initResult.getDistributionMessageResult();
+ logger.error(ModelLoaderMsgs.ASDC_CONNECTION_ERROR, errorMsg);
+
+ // Kick off a timer to retry the SDC connection
+ Timer timer = new Timer();
+ TimerTask task = new SdcConnectionJob(client, config, eventCallback, timer);
+ timer.schedule(task, new Date(), 60000);
+ }
+ }
+}
import java.util.Timer;
import java.util.TimerTask;
+import javax.annotation.PostConstruct;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
this.artifactDownloadManager = artifactDownloadManager;
}
+ @PostConstruct
+ protected void start() {
+ if (!config.getASDCConnectionDisabled()) {
+ initSdcClient();
+ }
+ }
+
/**
* Responsible for stopping the connection to the distribution client before the resource is destroyed.
*/
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom AG Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.modelloader.distribution;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.aspectj.lang.annotation.After;
+import org.aspectj.lang.annotation.Aspect;
+import org.springframework.stereotype.Component;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Aspect
+@Getter
+@Setter
+@Component
+/**
+ * Aspect to asynchronously wait for the EventCallback being called after a distribution event from sdc was published
+ */
+public class EventCallbackAspect {
+
+ private CountDownLatch countDownLatch;
+
+ @After(
+ "execution(* org.onap.aai.modelloader.notification.EventCallback.activateCallback(..))")
+ private void afterEventCallbackCalled() {
+ if (countDownLatch != null) {
+ countDownLatch.countDown();
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/**
+ * ============LICENSE_START=======================================================
+ * org.onap.aai
+ * ================================================================================
+ * Copyright © 2024 Deutsche Telekom AG Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.aai.modelloader.distribution;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.annotation.DirtiesContext;
+
+@DirtiesContext
+@SpringBootTest(properties = { "model-loader.sdc.connection.enabled=true"})
+@EmbeddedKafka(partitions = 1, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })
+public class NotificationIntegrationTest {
+
+ @Autowired EventCallbackAspect eventCallbackAspect;
+
+ @Autowired
+ private KafkaTemplate<String, String> kafkaTemplate;
+
+ @Value("${test.topic}")
+ private String topic;
+
+ @Test
+ @Disabled("This test is not yet implemented")
+ public void thatActivateCallbackIsCalled()
+ throws Exception {
+ String data = "Smth";
+
+ // TODO: send distribution event here
+ kafkaTemplate.send(topic, data);
+
+ // TODO: mock distribution client requests to /sdc/v1/artifactTypes
+ // TODO: mock distribution client requests to /sdc/v1/distributionKafkaData
+
+ // aspect will wait for EventCallback.activateCallback to be called
+ eventCallbackAspect.setCountDownLatch(new CountDownLatch(1));
+ boolean callbackCalled = eventCallbackAspect.getCountDownLatch().await(10, TimeUnit.SECONDS);
+ assertTrue(callbackCalled);
+ }
+}
@BeforeEach
public void setupMocks() {
- MockitoAnnotations.initMocks(this);
+ MockitoAnnotations.openMocks(this);
}
@Test
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.onap.aai.modelloader.fixture.NotificationDataFixtureBuilder.getNotificationDataWithInvalidType;
import static org.onap.aai.modelloader.fixture.NotificationDataFixtureBuilder.getNotificationDataWithModelQuerySpec;
configProperties.put("ml.babel.TRUSTSTORE_FILE", "src/test/resources/auth/aai-client-dummy.p12");
configProperties.put("ml.babel.BASE_URL", url);
configProperties.put("ml.babel.GENERATE_ARTIFACTS_URL", "generate");
+ configProperties.put("ml.aai.RESTCLIENT_CONNECT_TIMEOUT", "12000");
+ configProperties.put("ml.aai.RESTCLIENT_READ_TIMEOUT", "12000");
BabelServiceClient client =
new HttpsBabelServiceClientFactory().create(new ModelLoaderConfig(configProperties, "."));
List<BabelArtifact> result =
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.IOException;
--- /dev/null
+CONFIG_HOME=src/test/resources
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.group-id=aai
+spring.kafka.consumer.client-id=aai-model-loader
+
+test.topic=SDC-DISTR-NOTIF-TOPIC
+
+spring.sleuth.enabled=false
+
+ml.distribution.connection.enabled=false # avoid having the distribution client running in the background (requires active kafka)
\ No newline at end of file
ml.distribution.KEYSTORE_PASSWORD=
ml.distribution.KEYSTORE_FILE=
ml.distribution.PASSWORD=Aa123456
-ml.distribution.POLLING_INTERVAL=30
-ml.distribution.POLLING_TIMEOUT=20
+ml.distribution.POLLING_INTERVAL=5
+ml.distribution.POLLING_TIMEOUT=3
ml.distribution.USER=ci
ml.distribution.ARTIFACT_TYPES=MODEL_QUERY_SPEC,TOSCA_CSAR
Code Review / aai / model-loader.git / commitdiff