<artifactId>validation-api</artifactId>
<version>2.0.1.Final</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>3.3.1</version>
+ </dependency>
</dependencies>
</dependencyManagement>
</project>
"type": "object",
"properties": {
"operationId": {
- "description": "Used to distinguish multiple operations using same cmhandleId",
+ "description": "Used to distinguish multiple operations using same handle ids",
"type": "string"
},
"ids": {
"description": "Id's of the cmhandles",
- "type": "array"
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
},
"statusCode": {
"description": "which says success or failure (0-99) are for success and (100-199) are for failure",
"description": "Human readable message, Which says what the response has",
"type": "string"
},
- "responseContent": {
+ "result": {
"description": "Contains the requested data response.",
"type": "object",
"existingJavaType": "java.lang.Object",
"SubscriptionEventResponse" : {
"description": "The payload for subscription response event.",
"type": "object",
- "javaType": "org.onap.cps.ncmp.events.subscription1_0_0.SubscriptionEventResponse",
+ "javaType": "org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionEventResponse",
"properties": {
"data": {
"type": "object",
</configOptions>
</configuration>
</execution>
+ <execution>
+ <id>ncmp-openapi-yaml-gen</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>compile</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/docs/openapi/openapi.yml</inputSpec>
+ <generatorName>openapi-yaml</generatorName>
+ <configOptions>
+ <outputFile>openapi.yaml</outputFile>
+ </configOptions>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
+ <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<executions>
<execution>
</outputDirectory>
<resources>
<resource>
- <directory>${project.basedir}/target/generated-sources/swagger/</directory>
+ <directory>${project.basedir}/target/generated-sources/openapi/</directory>
<includes>
<include>openapi*.yaml</include>
</includes>
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api;
+
+import lombok.Getter;
+
+@Getter
+public enum NcmpEventResponseCode {
+
+ CODE_100("100", "cm handle id(s) not found"),
+ CODE_101("101", "cm handle(s) not ready");
+
+ private final String statusCode;
+ private final String statusMessage;
+
+ NcmpEventResponseCode(final String statusCode, final String statusMessage) {
+ this.statusCode = statusCode;
+ this.statusMessage = statusMessage;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import java.net.URI;
+import java.util.Map;
+import java.util.UUID;
+import lombok.Builder;
+import org.apache.commons.lang3.StringUtils;
+import org.onap.cps.ncmp.api.impl.utils.EventDateTimeFormatter;
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext;
+import org.onap.cps.utils.JsonObjectMapper;
+
+@Builder(buildMethodName = "setCloudEvent")
+public class NcmpCloudEventBuilder {
+
+ private Object event;
+ private Map<String, String> extensions;
+ private String type;
+ @Builder.Default
+ private static final String EVENT_SPEC_VERSION_V1 = "1.0.0";
+
+ /**
+ * Creates ncmp cloud event with provided attributes.
+ *
+ * @return Cloud Event
+ */
+ public CloudEvent build() {
+ final JsonObjectMapper jsonObjectMapper = CpsApplicationContext.getCpsBean(JsonObjectMapper.class);
+ final CloudEventBuilder cloudEventBuilder = CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withSource(URI.create("NCMP"))
+ .withType(type)
+ .withDataSchema(URI.create("urn:cps:" + type + ":" + EVENT_SPEC_VERSION_V1))
+ .withTime(EventDateTimeFormatter.toIsoOffsetDateTime(
+ EventDateTimeFormatter.getCurrentIsoFormattedDateTime()))
+ .withData(jsonObjectMapper.asJsonBytes(event));
+ extensions.entrySet().stream()
+ .filter(extensionEntry -> StringUtils.isNotBlank(extensionEntry.getValue()))
+ .forEach(extensionEntry ->
+ cloudEventBuilder.withExtension(extensionEntry.getKey(), extensionEntry.getValue()));
+ return cloudEventBuilder.build();
+ }
+}
private void generateAndSendResponse() {
final String subscriptionEventId = subscriptionClientId + subscriptionName;
if (forwardedSubscriptionEventCache.containsKey(subscriptionEventId)) {
- final Set<String> dmiNames = forwardedSubscriptionEventCache.get(subscriptionEventId);
- subscriptionEventResponseOutcome.sendResponse(subscriptionClientId, subscriptionName,
- dmiNames.isEmpty());
+ subscriptionEventResponseOutcome.sendResponse(subscriptionClientId, subscriptionName);
forwardedSubscriptionEventCache.remove(subscriptionEventId);
}
}
package org.onap.cps.ncmp.api.impl.events.avcsubscription;
import com.hazelcast.map.IMap;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import org.apache.kafka.common.header.Headers;
import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig;
import org.onap.cps.ncmp.api.impl.events.EventsPublisher;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
import org.onap.cps.ncmp.event.model.SubscriptionEvent;
import org.onap.cps.spi.exceptions.OperationNotYetSupportedException;
private final EventsPublisher<SubscriptionEvent> eventsPublisher;
private final IMap<String, Set<String>> forwardedSubscriptionEventCache;
private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome;
+ private final SubscriptionEventMapper subscriptionEventMapper;
+ private final SubscriptionPersistence subscriptionPersistence;
private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
@Value("${app.ncmp.avc.subscription-forward-topic-prefix}")
private String dmiAvcSubscriptionTopicPrefix;
final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName
= DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
+ findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings,
+ dmiPropertiesPerCmHandleIdPerServiceName);
+ }
+
+ private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders,
+ final List<String> cmHandleTargetsAsStrings,
+ final Map<String, Map<String, Map<String, String>>>
+ dmiPropertiesPerCmHandleIdPerServiceName) {
+ final List<String> cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream()
+ .map(Map.Entry::getValue).map(Map::keySet).flatMap(Set::stream).collect(Collectors.toList());
+
+ final List<String> targetCmHandlesDoesNotExistInDb = new ArrayList<>(cmHandleTargetsAsStrings);
+ targetCmHandlesDoesNotExistInDb.removeAll(cmHandlesThatExistsInDb);
+
final Set<String> dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet());
+
+ if (dmisToRespond.isEmpty() || !targetCmHandlesDoesNotExistInDb.isEmpty()) {
+ updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
+ }
if (dmisToRespond.isEmpty()) {
final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID();
final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
- subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName, true);
+ subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName);
} else {
startResponseTimeout(subscriptionEvent, dmisToRespond);
forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders);
+ "-"
+ dmiName;
}
+
+ private void updatesCmHandlesToRejectedAndPersistSubscriptionEvent(
+ final SubscriptionEvent subscriptionEvent,
+ final List<String> targetCmHandlesDoesNotExistInDb) {
+ final YangModelSubscriptionEvent yangModelSubscriptionEvent =
+ subscriptionEventMapper.toYangModelSubscriptionEvent(subscriptionEvent);
+ yangModelSubscriptionEvent.getPredicates()
+ .setTargetCmHandles(findRejectedCmHandles(targetCmHandlesDoesNotExistInDb,
+ yangModelSubscriptionEvent));
+ subscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEvent);
+ }
+
+ private static List<YangModelSubscriptionEvent.TargetCmHandle> findRejectedCmHandles(
+ final List<String> targetCmHandlesDoesNotExistInDb,
+ final YangModelSubscriptionEvent yangModelSubscriptionEvent) {
+ return yangModelSubscriptionEvent.getPredicates().getTargetCmHandles().stream()
+ .filter(targetCmHandle -> targetCmHandlesDoesNotExistInDb.contains(targetCmHandle.getCmHandleId()))
+ .map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.getCmHandleId(),
+ SubscriptionStatus.REJECTED)).collect(Collectors.toList());
+ }
}
package org.onap.cps.ncmp.api.impl.events.avcsubscription;
import com.hazelcast.map.IMap;
+import java.util.Collection;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig;
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
+import org.onap.cps.ncmp.api.impl.utils.DataNodeHelper;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
import org.onap.cps.ncmp.api.models.SubscriptionEventResponse;
+import org.onap.cps.spi.model.DataNode;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
log.info("subscription event response of clientId: {} is received.", clientId);
final String subscriptionName = subscriptionEventResponse.getSubscriptionName();
final String subscriptionEventId = clientId + subscriptionName;
- boolean isFullOutcomeResponse = false;
+ boolean createOutcomeResponse = false;
if (forwardedSubscriptionEventCache.containsKey(subscriptionEventId)) {
final Set<String> dmiNames = forwardedSubscriptionEventCache.get(subscriptionEventId);
dmiNames.remove(subscriptionEventResponse.getDmiName());
forwardedSubscriptionEventCache.put(subscriptionEventId, dmiNames,
ForwardedSubscriptionEventCacheConfig.SUBSCRIPTION_FORWARD_STARTED_TTL_SECS, TimeUnit.SECONDS);
- isFullOutcomeResponse = forwardedSubscriptionEventCache.get(subscriptionEventId).isEmpty();
-
- if (isFullOutcomeResponse) {
- forwardedSubscriptionEventCache.remove(subscriptionEventId);
- }
+ createOutcomeResponse = forwardedSubscriptionEventCache.get(subscriptionEventId).isEmpty();
}
if (subscriptionModelLoaderEnabled) {
updateSubscriptionEvent(subscriptionEventResponse);
}
- if (isFullOutcomeResponse && notificationFeatureEnabled) {
- subscriptionEventResponseOutcome.sendResponse(clientId, subscriptionName,
- isFullOutcomeResponse);
+ if (createOutcomeResponse
+ && notificationFeatureEnabled
+ && hasNoPendingCmHandles(clientId, subscriptionName)) {
+ subscriptionEventResponseOutcome.sendResponse(clientId, subscriptionName);
+ forwardedSubscriptionEventCache.remove(subscriptionEventId);
}
}
+ private boolean hasNoPendingCmHandles(final String clientId, final String subscriptionName) {
+ final Collection<DataNode> dataNodeSubscription = subscriptionPersistence.getCmHandlesForSubscriptionEvent(
+ clientId, subscriptionName);
+ final Map<String, SubscriptionStatus> cmHandleIdToStatusMap =
+ DataNodeHelper.getCmHandleIdToStatusMapFromDataNodes(
+ dataNodeSubscription);
+ return !cmHandleIdToStatusMap.values().contains(SubscriptionStatus.PENDING);
+ }
+
private void updateSubscriptionEvent(final SubscriptionEventResponse subscriptionEventResponse) {
final YangModelSubscriptionEvent yangModelSubscriptionEvent =
subscriptionEventResponseMapper
*
* @param subscriptionClientId client id of the subscription.
* @param subscriptionName name of the subscription.
- * @param isFullOutcomeResponse the flag to decide on complete or partial response to be generated.
*/
- public void sendResponse(final String subscriptionClientId, final String subscriptionName,
- final boolean isFullOutcomeResponse) {
+ public void sendResponse(final String subscriptionClientId, final String subscriptionName) {
final SubscriptionEventOutcome subscriptionEventOutcome = generateResponse(
- subscriptionClientId, subscriptionName, isFullOutcomeResponse);
+ subscriptionClientId, subscriptionName);
final Headers headers = new RecordHeaders();
final String subscriptionEventId = subscriptionClientId + subscriptionName;
outcomeEventsPublisher.publishEvent(subscriptionOutcomeEventTopic,
subscriptionEventId, headers, subscriptionEventOutcome);
}
- private SubscriptionEventOutcome generateResponse(final String subscriptionClientId, final String subscriptionName,
- final boolean isFullOutcomeResponse) {
- final Collection<DataNode> dataNodes = subscriptionPersistence.getDataNodesForSubscriptionEvent();
+ private SubscriptionEventOutcome generateResponse(final String subscriptionClientId,
+ final String subscriptionName) {
+ final Collection<DataNode> dataNodes =
+ subscriptionPersistence.getCmHandlesForSubscriptionEvent(subscriptionClientId, subscriptionName);
final List<Map<String, Serializable>> dataNodeLeaves = DataNodeHelper.getDataNodeLeaves(dataNodes);
final List<Collection<Serializable>> cmHandleIdToStatus =
DataNodeHelper.getCmHandleIdToStatus(dataNodeLeaves);
+ final Map<String, SubscriptionStatus> cmHandleIdToStatusMap =
+ DataNodeHelper.getCmHandleIdToStatusMap(cmHandleIdToStatus);
return formSubscriptionOutcomeMessage(cmHandleIdToStatus, subscriptionClientId, subscriptionName,
- isFullOutcomeResponse);
+ isFullOutcomeResponse(cmHandleIdToStatusMap));
}
+ private boolean isFullOutcomeResponse(final Map<String, SubscriptionStatus> cmHandleIdToStatusMap) {
+ return !cmHandleIdToStatusMap.values().contains(SubscriptionStatus.PENDING);
+ }
private SubscriptionEventOutcome formSubscriptionOutcomeMessage(
final List<Collection<Serializable>> cmHandleIdToStatus, final String subscriptionClientId,
cmHandleTransitionPairs.add(cmHandleTransitionPair);
}
});
-
return cmHandleTransitionPairs;
}
private void persistCmHandle(final YangModelCmHandle targetYangModelCmHandle,
final YangModelCmHandle currentYangModelCmHandle) {
- if (isNew(currentYangModelCmHandle.getCompositeState(), targetYangModelCmHandle.getCompositeState())) {
+ if (isNew(currentYangModelCmHandle.getCompositeState())) {
log.debug("Registering a new cm handle {}", targetYangModelCmHandle.getId());
inventoryPersistence.saveCmHandle(targetYangModelCmHandle);
} else if (isDeleted(targetYangModelCmHandle.getCompositeState())) {
final Map<String, CompositeState> compositeStatePerCmHandleId = new LinkedHashMap<>();
cmHandleTransitionPairs.forEach(cmHandleTransitionPair -> {
- if (isNew(cmHandleTransitionPair.getCurrentYangModelCmHandle().getCompositeState(),
- cmHandleTransitionPair.getTargetYangModelCmHandle().getCompositeState())) {
+ if (isNew(cmHandleTransitionPair.getCurrentYangModelCmHandle().getCompositeState()
+ )) {
newCmHandles.add(cmHandleTransitionPair.getTargetYangModelCmHandle());
} else if (!isDeleted(cmHandleTransitionPair.getTargetYangModelCmHandle().getCompositeState())) {
compositeStatePerCmHandleId.put(cmHandleTransitionPair.getTargetYangModelCmHandle().getId(),
CompositeStateUtils.setCompositeState(targetCmHandleState).accept(yangModelCmHandle.getCompositeState());
}
- private boolean isNew(final CompositeState existingCompositeState, final CompositeState targetCompositeState) {
- return (existingCompositeState == null && targetCompositeState.getCmHandleState() == ADVISED);
+ private boolean isNew(final CompositeState existingCompositeState) {
+ return (existingCompositeState == null);
}
private boolean isDeleted(final CompositeState targetCompositeState) {
final LcmEvent lcmEvent = new LcmEvent();
lcmEvent.setEventId(UUID.randomUUID().toString());
lcmEvent.setEventCorrelationId(eventCorrelationId);
- lcmEvent.setEventTime(EventDateTimeFormatter.getCurrentDateTime());
+ lcmEvent.setEventTime(EventDateTimeFormatter.getCurrentIsoFormattedDateTime());
lcmEvent.setEventSource("org.onap.ncmp");
lcmEvent.setEventType(lcmEventType.getEventType());
lcmEvent.setEventSchema("org.onap.ncmp:cmhandle-lcm-event");
import org.onap.cps.ncmp.api.impl.config.NcmpConfiguration;
import org.onap.cps.ncmp.api.impl.executor.TaskExecutor;
import org.onap.cps.ncmp.api.impl.utils.DmiServiceUrlBuilder;
-import org.onap.cps.ncmp.api.impl.utils.ResourceDataOperationRequestUtils;
+import org.onap.cps.ncmp.api.impl.utils.data.operation.ResourceDataOperationRequestUtils;
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
import org.onap.cps.ncmp.api.inventory.CmHandleState;
import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
= getDistinctCmHandleIdsFromDataOperationRequest(dataOperationRequest);
final Collection<YangModelCmHandle> yangModelCmHandles
- = getYangModelCmHandlesInReadyState(cmHandlesIds);
+ = inventoryPersistence.getYangModelCmHandles(cmHandlesIds);
final Map<String, List<DmiDataOperation>> operationsOutPerDmiServiceName
- = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(dataOperationRequest,
- yangModelCmHandles);
+ = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(topicParamInQuery,
+ requestId, dataOperationRequest, yangModelCmHandles);
buildDataOperationRequestUrlAndSendToDmiService(topicParamInQuery, requestId, operationsOutPerDmiServiceName);
}
dataOperationDefinition.getCmHandleIds().stream()).collect(Collectors.toSet());
}
- private Collection<YangModelCmHandle> getYangModelCmHandlesInReadyState(final Set<String> requestedCmHandleIds) {
- // TODO Need to publish an error response to client given topic.
- // Code should be implemented into https://jira.onap.org/browse/CPS-1614 (
- // NCMP : Error handling for non-ready cm handle state)
- return inventoryPersistence.getYangModelCmHandles(requestedCmHandleIds).stream()
- .filter(yangModelCmHandle -> yangModelCmHandle.getCompositeState().getCmHandleState()
- == CmHandleState.READY).collect(Collectors.toList());
- }
-
private void buildDataOperationRequestUrlAndSendToDmiService(final String topicParamInQuery,
final String requestId,
final Map<String, List<DmiDataOperation>>
}
return resultMap;
}
+
+ /**
+ * Extracts the mapping of cm handle id to status from data node collection.
+ *
+ * @param dataNodes as a collection
+ * @return cm handle id to status mapping
+ */
+ public static Map<String, SubscriptionStatus> getCmHandleIdToStatusMapFromDataNodes(
+ final Collection<DataNode> dataNodes) {
+ return getCmHandleIdToStatusMap(getCmHandleIdToStatus(getDataNodeLeaves(dataNodes)));
+ }
}
/*
* ============LICENSE_START=======================================================
- * Copyright (C) 2022 Nordix Foundation
+ * Copyright (C) 2022-2023 Nordix Foundation
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.cps.ncmp.api.impl.utils;
+import java.time.OffsetDateTime;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class EventDateTimeFormatter {
+public interface EventDateTimeFormatter {
- private static final String DATE_TIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+ String ISO_TIMESTAMP_PATTERN = "yyyy-MM-dd'T'HH:mm:ss.SSSZ";
+
+ DateTimeFormatter ISO_TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern(ISO_TIMESTAMP_PATTERN);
/**
* Gets current date time.
*
* @return the current date time
*/
- public static String getCurrentDateTime() {
- return ZonedDateTime.now()
- .format(DateTimeFormatter.ofPattern(DATE_TIME_FORMAT));
+ static String getCurrentIsoFormattedDateTime() {
+ return ZonedDateTime.now().format(ISO_TIMESTAMP_FORMATTER);
+ }
+
+ static OffsetDateTime toIsoOffsetDateTime(final String dateTimestampAsString) {
+ return StringUtils.isNotBlank(dateTimestampAsString)
+ ? OffsetDateTime.parse(dateTimestampAsString, ISO_TIMESTAMP_FORMATTER) : null;
}
}
+++ /dev/null
-/*
- * ============LICENSE_START=======================================================
- * Copyright (C) 2023 Nordix Foundation
- * ================================================================================
- * 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.
- *
- * SPDX-License-Identifier: Apache-2.0
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.cps.ncmp.api.impl.utils;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.onap.cps.ncmp.api.impl.operations.CmHandle;
-import org.onap.cps.ncmp.api.impl.operations.DmiDataOperation;
-import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
-import org.onap.cps.ncmp.api.models.DataOperationDefinition;
-import org.onap.cps.ncmp.api.models.DataOperationRequest;
-
-@Slf4j
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class ResourceDataOperationRequestUtils {
-
- private static final String UNKNOWN_SERVICE_NAME = null;
-
- /**
- * Create a list of DMI data operations per DMI service (name).
- *
- * @param dataOperationRequestIn incoming data operation request details
- * @param yangModelCmHandles involved cm handles represented as YangModelCmHandle (incl. metadata)
- *
- * @return {@code Map<String, List<DmiDataOperation>>} Create a list of DMI data operations operation
- * per DMI service (name).
- */
- public static Map<String, List<DmiDataOperation>> processPerDefinitionInDataOperationsRequest(
- final DataOperationRequest dataOperationRequestIn,
- final Collection<YangModelCmHandle> yangModelCmHandles) {
-
- final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName =
- DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
-
- final Map<String, String> dmiServiceNamesPerCmHandleId =
- getDmiServiceNamesPerCmHandleId(dmiPropertiesPerCmHandleIdPerServiceName);
-
- final Map<String, List<DmiDataOperation>> dmiDataOperationsOutPerDmiServiceName = new HashMap<>();
-
- for (final DataOperationDefinition dataOperationDefinitionIn :
- dataOperationRequestIn.getDataOperationDefinitions()) {
- for (final String cmHandleId : dataOperationDefinitionIn.getCmHandleIds()) {
- final String dmiServiceName = dmiServiceNamesPerCmHandleId.get(cmHandleId);
- final Map<String, String> cmHandleIdProperties
- = dmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).get(cmHandleId);
- if (cmHandleIdProperties == null) {
- publishErrorMessageToClientTopic(cmHandleId);
- } else {
- final DmiDataOperation dmiDataOperationOut = getOrAddDmiDataOperation(dmiServiceName,
- dataOperationDefinitionIn, dmiDataOperationsOutPerDmiServiceName);
- final CmHandle cmHandle = CmHandle.buildCmHandleWithProperties(cmHandleId, cmHandleIdProperties);
- dmiDataOperationOut.getCmHandles().add(cmHandle);
- }
- }
- }
- return dmiDataOperationsOutPerDmiServiceName;
- }
-
- private static void publishErrorMessageToClientTopic(final String requestedCmHandleId) {
- log.warn("cm handle {} not found", requestedCmHandleId);
- // TODO Need to publish an error response to client given topic.
- // Code should be implemented into https://jira.onap.org/browse/CPS-1583 (
- // NCMP : Handle non-existing cm handles)
- }
-
- private static Map<String, String> getDmiServiceNamesPerCmHandleId(
- final Map<String, Map<String, Map<String, String>>> dmiDmiPropertiesPerCmHandleIdPerServiceName) {
- final Map<String, String> dmiServiceNamesPerCmHandleId = new HashMap<>();
- for (final Map.Entry<String, Map<String, Map<String, String>>> dmiDmiPropertiesEntry
- : dmiDmiPropertiesPerCmHandleIdPerServiceName.entrySet()) {
- final String dmiServiceName = dmiDmiPropertiesEntry.getKey();
- final Set<String> cmHandleIds = dmiDmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).keySet();
- for (final String cmHandleId : cmHandleIds) {
- dmiServiceNamesPerCmHandleId.put(cmHandleId, dmiServiceName);
- }
- }
- dmiDmiPropertiesPerCmHandleIdPerServiceName.put(UNKNOWN_SERVICE_NAME, Collections.emptyMap());
- return dmiServiceNamesPerCmHandleId;
- }
-
- private static DmiDataOperation getOrAddDmiDataOperation(final String dmiServiceName,
- final DataOperationDefinition
- dataOperationDefinitionIn,
- final Map<String, List<DmiDataOperation>>
- dmiDataOperationsOutPerDmiServiceName) {
- dmiDataOperationsOutPerDmiServiceName
- .computeIfAbsent(dmiServiceName, dmiServiceNameAsKey -> new ArrayList<>());
- final List<DmiDataOperation> dmiDataOperationsOut
- = dmiDataOperationsOutPerDmiServiceName.get(dmiServiceName);
- final boolean isNewOperation = dmiDataOperationsOut.isEmpty()
- || !dmiDataOperationsOut.get(dmiDataOperationsOut.size() - 1).getOperationId()
- .equals(dataOperationDefinitionIn.getOperationId());
- if (isNewOperation) {
- final DmiDataOperation newDmiDataOperationOut =
- DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn);
- dmiDataOperationsOut.add(newDmiDataOperationOut);
- return newDmiDataOperationOut;
- }
- return dmiDataOperationsOut.get(dmiDataOperationsOut.size() - 1);
- }
-}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.context;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CpsApplicationContext implements ApplicationContextAware {
+
+ private static ApplicationContext applicationContext;
+
+ /**
+ * Returns the spring managed cps bean instance of the given class type if it exists.
+ * Returns null otherwise.
+ *
+ * @param cpsBeanClass cps class type
+ * @return requested bean instance
+ */
+ public static <T extends Object> T getCpsBean(final Class<T> cpsBeanClass) {
+ return applicationContext.getBean(cpsBeanClass);
+ }
+
+ @Override
+ public void setApplicationContext(final ApplicationContext cpsApplicationContext) {
+ setCpsApplicationContext(cpsApplicationContext);
+ }
+
+ private static synchronized void setCpsApplicationContext(final ApplicationContext cpsApplicationContext) {
+ CpsApplicationContext.applicationContext = cpsApplicationContext;
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.data.operation;
+
+import io.cloudevents.CloudEvent;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NcmpEventResponseCode;
+import org.onap.cps.ncmp.api.impl.events.NcmpCloudEventBuilder;
+import org.onap.cps.ncmp.events.async1_0_0.Data;
+import org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent;
+import org.onap.cps.ncmp.events.async1_0_0.Response;
+import org.springframework.util.MultiValueMap;
+
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class DataOperationEventCreator {
+
+ /**
+ * Creates data operation event.
+ *
+ * @param clientTopic topic the client wants to use for responses
+ * @param requestId unique identifier per request
+ * @param cmHandleIdsPerResponseCodesPerOperationId map of cm handles per operation response per response code
+ * @return Cloud Event
+ */
+ public static CloudEvent createDataOperationEvent(final String clientTopic,
+ final String requestId,
+ final MultiValueMap<String,
+ Map<NcmpEventResponseCode, List<String>>>
+ cmHandleIdsPerResponseCodesPerOperationId) {
+ final DataOperationEvent dataOperationEvent = new DataOperationEvent();
+ final Data data = createPayloadFromDataOperationResponses(cmHandleIdsPerResponseCodesPerOperationId);
+ dataOperationEvent.setData(data);
+ final Map<String, String> extensions = createDataOperationExtensions(requestId, clientTopic);
+ return NcmpCloudEventBuilder.builder().type(DataOperationEvent.class.getName())
+ .event(dataOperationEvent).extensions(extensions).setCloudEvent().build();
+ }
+
+ private static Data createPayloadFromDataOperationResponses(final MultiValueMap<String, Map<NcmpEventResponseCode,
+ List<String>>> cmHandleIdsPerOperationIdPerResponseCode) {
+ final Data data = new Data();
+ final List<org.onap.cps.ncmp.events.async1_0_0.Response> responses = new ArrayList<>();
+ cmHandleIdsPerOperationIdPerResponseCode.entrySet().forEach(cmHandleIdsPerOperationIdPerResponseCodeEntries ->
+ cmHandleIdsPerOperationIdPerResponseCodeEntries.getValue().forEach(cmHandleIdsPerResponseCodeEntries ->
+ responses.addAll(createResponseFromDataOperationResponses(
+ cmHandleIdsPerOperationIdPerResponseCodeEntries.getKey(),
+ cmHandleIdsPerResponseCodeEntries)
+ )));
+ data.setResponses(responses);
+ return data;
+ }
+
+ private static List<Response> createResponseFromDataOperationResponses(
+ final String operationId,
+ final Map<NcmpEventResponseCode, List<String>> cmHandleIdsPerResponseCodeEntries) {
+ final List<org.onap.cps.ncmp.events.async1_0_0.Response> responses = new ArrayList<>();
+ cmHandleIdsPerResponseCodeEntries.entrySet()
+ .forEach(cmHandleIdsPerResponseCodeEntry -> {
+ final Response response = new Response();
+ response.setOperationId(operationId);
+ response.setStatusCode(cmHandleIdsPerResponseCodeEntry.getKey().getStatusCode());
+ response.setStatusMessage(cmHandleIdsPerResponseCodeEntry.getKey().getStatusMessage());
+ response.setIds(cmHandleIdsPerResponseCodeEntry.getValue());
+ responses.add(response);
+ });
+ return responses;
+ }
+
+ private static Map<String, String> createDataOperationExtensions(final String requestId, final String clientTopic) {
+ final Map<String, String> extensions = new HashMap<>();
+ extensions.put("correlationid", requestId);
+ extensions.put("destination", clientTopic);
+ return extensions;
+ }
+}
--- /dev/null
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils.data.operation;
+
+import io.cloudevents.CloudEvent;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.NcmpEventResponseCode;
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher;
+import org.onap.cps.ncmp.api.impl.operations.CmHandle;
+import org.onap.cps.ncmp.api.impl.operations.DmiDataOperation;
+import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer;
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext;
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
+import org.onap.cps.ncmp.api.inventory.CmHandleState;
+import org.onap.cps.ncmp.api.models.DataOperationDefinition;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+
+@Slf4j
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class ResourceDataOperationRequestUtils {
+
+ private static final String UNKNOWN_SERVICE_NAME = null;
+
+ /**
+ * Create a list of DMI data operation per DMI service (name).
+ *
+ * @param topicParamInQuery client given topic
+ * @param requestId unique identifier per request
+ * @param dataOperationRequestIn incoming data operation request details
+ * @param yangModelCmHandles involved cm handles represented as YangModelCmHandle (incl. metadata)
+ * @return {@code Map<String, List<DmiBatchOperation>>} Create a list of DMI batch operation per DMI service (name).
+ */
+ public static Map<String, List<DmiDataOperation>> processPerDefinitionInDataOperationsRequest(
+ final String topicParamInQuery,
+ final String requestId,
+ final DataOperationRequest dataOperationRequestIn,
+ final Collection<YangModelCmHandle> yangModelCmHandles) {
+
+ final Map<String, List<DmiDataOperation>> dmiDataOperationsOutPerDmiServiceName = new HashMap<>();
+ final MultiValueMap<String, Map<NcmpEventResponseCode, List<String>>> cmHandleIdsPerOperationIdPerResponseCode
+ = new LinkedMultiValueMap<>();
+ final Set<String> nonReadyCmHandleIdsLookup = filterAndGetNonReadyCmHandleIds(yangModelCmHandles);
+
+ final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName =
+ DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
+
+ final Map<String, String> dmiServiceNamesPerCmHandleId =
+ getDmiServiceNamesPerCmHandleId(dmiPropertiesPerCmHandleIdPerServiceName);
+
+ for (final DataOperationDefinition dataOperationDefinitionIn :
+ dataOperationRequestIn.getDataOperationDefinitions()) {
+ final List<String> nonExistingCmHandleIds = new ArrayList<>();
+ final List<String> nonReadyCmHandleIds = new ArrayList<>();
+ for (final String cmHandleId : dataOperationDefinitionIn.getCmHandleIds()) {
+ if (nonReadyCmHandleIdsLookup.contains(cmHandleId)) {
+ nonReadyCmHandleIds.add(cmHandleId);
+ } else {
+ final String dmiServiceName = dmiServiceNamesPerCmHandleId.get(cmHandleId);
+ final Map<String, String> cmHandleIdProperties
+ = dmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).get(cmHandleId);
+ if (cmHandleIdProperties == null) {
+ nonExistingCmHandleIds.add(cmHandleId);
+ } else {
+ final DmiDataOperation dmiBatchOperationOut = getOrAddDmiBatchOperation(dmiServiceName,
+ dataOperationDefinitionIn, dmiDataOperationsOutPerDmiServiceName);
+ final CmHandle cmHandle = CmHandle.buildCmHandleWithProperties(cmHandleId,
+ cmHandleIdProperties);
+ dmiBatchOperationOut.getCmHandles().add(cmHandle);
+ }
+ }
+ }
+ populateCmHandleIdsPerOperationIdPerResponseCode(cmHandleIdsPerOperationIdPerResponseCode,
+ dataOperationDefinitionIn.getOperationId(), NcmpEventResponseCode.CODE_100, nonExistingCmHandleIds);
+ populateCmHandleIdsPerOperationIdPerResponseCode(cmHandleIdsPerOperationIdPerResponseCode,
+ dataOperationDefinitionIn.getOperationId(), NcmpEventResponseCode.CODE_101, nonReadyCmHandleIds);
+ }
+ if (!cmHandleIdsPerOperationIdPerResponseCode.isEmpty()) {
+ publishErrorMessageToClientTopic(topicParamInQuery, requestId, cmHandleIdsPerOperationIdPerResponseCode);
+ }
+ return dmiDataOperationsOutPerDmiServiceName;
+ }
+
+ @Async
+ private static void publishErrorMessageToClientTopic(final String clientTopic,
+ final String requestId,
+ final MultiValueMap<String,
+ Map<NcmpEventResponseCode, List<String>>>
+ cmHandleIdsPerOperationIdPerResponseCode) {
+ final CloudEvent dataOperationCloudEvent = DataOperationEventCreator.createDataOperationEvent(clientTopic,
+ requestId, cmHandleIdsPerOperationIdPerResponseCode);
+ final EventsPublisher<CloudEvent> eventsPublisher = CpsApplicationContext.getCpsBean(EventsPublisher.class);
+ eventsPublisher.publishCloudEvent(clientTopic, requestId, dataOperationCloudEvent);
+ }
+
+ private static Map<String, String> getDmiServiceNamesPerCmHandleId(
+ final Map<String, Map<String, Map<String, String>>> dmiDmiPropertiesPerCmHandleIdPerServiceName) {
+ final Map<String, String> dmiServiceNamesPerCmHandleId = new HashMap<>();
+ for (final Map.Entry<String, Map<String, Map<String, String>>> dmiDmiPropertiesEntry
+ : dmiDmiPropertiesPerCmHandleIdPerServiceName.entrySet()) {
+ final String dmiServiceName = dmiDmiPropertiesEntry.getKey();
+ final Set<String> cmHandleIds = dmiDmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).keySet();
+ for (final String cmHandleId : cmHandleIds) {
+ dmiServiceNamesPerCmHandleId.put(cmHandleId, dmiServiceName);
+ }
+ }
+ dmiDmiPropertiesPerCmHandleIdPerServiceName.put(UNKNOWN_SERVICE_NAME, Collections.emptyMap());
+ return dmiServiceNamesPerCmHandleId;
+ }
+
+ private static DmiDataOperation getOrAddDmiBatchOperation(final String dmiServiceName,
+ final DataOperationDefinition
+ dataOperationDefinitionIn,
+ final Map<String, List<DmiDataOperation>>
+ dmiBatchOperationsOutPerDmiServiceName) {
+ dmiBatchOperationsOutPerDmiServiceName
+ .computeIfAbsent(dmiServiceName, dmiServiceNameAsKey -> new ArrayList<>());
+ final List<DmiDataOperation> dmiBatchOperationsOut
+ = dmiBatchOperationsOutPerDmiServiceName.get(dmiServiceName);
+ final boolean isNewOperation = dmiBatchOperationsOut.isEmpty()
+ || !dmiBatchOperationsOut.get(dmiBatchOperationsOut.size() - 1).getOperationId()
+ .equals(dataOperationDefinitionIn.getOperationId());
+ if (isNewOperation) {
+ final DmiDataOperation newDmiBatchOperationOut =
+ DmiDataOperation.buildDmiDataOperationRequestBodyWithoutCmHandles(dataOperationDefinitionIn);
+ dmiBatchOperationsOut.add(newDmiBatchOperationOut);
+ return newDmiBatchOperationOut;
+ }
+ return dmiBatchOperationsOut.get(dmiBatchOperationsOut.size() - 1);
+ }
+
+ private static Set<String> filterAndGetNonReadyCmHandleIds(final Collection<YangModelCmHandle> yangModelCmHandles) {
+ return yangModelCmHandles.stream()
+ .filter(yangModelCmHandle -> yangModelCmHandle.getCompositeState().getCmHandleState()
+ != CmHandleState.READY).map(YangModelCmHandle::getId).collect(Collectors.toSet());
+ }
+
+ private static void populateCmHandleIdsPerOperationIdPerResponseCode(final MultiValueMap<String,
+ Map<NcmpEventResponseCode, List<String>>> cmHandleIdsPerOperationIdByResponseCode,
+ final String operationId,
+ final NcmpEventResponseCode
+ ncmpEventResponseCode,
+ final List<String> cmHandleIds) {
+ if (!cmHandleIds.isEmpty()) {
+ cmHandleIdsPerOperationIdByResponseCode.add(operationId, Map.of(ncmpEventResponseCode, cmHandleIds));
+ }
+ }
+}
response.operationId == 'some-operation-id'
response.statusCode == 'any-success-status-code'
response.statusMessage == 'Successfully applied changes'
- response.responseContent as String == '[some-key:some-value]'
+ response.result as String == '[some-key:some-value]'
}
def 'Filter an event with type #eventType'() {
def createConsumerRecord(eventTypeAsString) {
def jsonData = TestUtils.getResourceFileContent('dataOperationEvent.json')
- def testEventSentAsBytes = objectMapper.writeValueAsBytes(jsonObjectMapper.convertJsonString(jsonData, DataOperationEvent.class))
+ def testEventSentAsBytes = jsonObjectMapper.asJsonBytes(jsonObjectMapper.convertJsonString(jsonData, DataOperationEvent.class))
CloudEvent cloudEvent = getCloudEvent(eventTypeAsString, testEventSentAsBytes)
--- /dev/null
+/*
+ * ============LICENSE_START========================================================
+ * Copyright (c) 2023 Nordix Foundation.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events
+
+import ch.qos.logback.classic.Level
+import ch.qos.logback.classic.Logger
+import ch.qos.logback.core.read.ListAppender
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.clients.producer.RecordMetadata
+import org.apache.kafka.common.TopicPartition
+import org.onap.cps.ncmp.init.SubscriptionModelLoader
+import org.slf4j.LoggerFactory
+import org.springframework.kafka.support.SendResult
+import spock.lang.Specification
+
+class EventPublisherSpec extends Specification {
+
+ def objectUnderTest = new EventsPublisher(null, null)
+ def logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
+ def loggingListAppender
+
+ void setup() {
+ logger.setLevel(Level.DEBUG)
+ loggingListAppender = new ListAppender()
+ logger.addAppender(loggingListAppender)
+ loggingListAppender.start()
+ }
+
+ void cleanup() {
+ ((Logger) LoggerFactory.getLogger(SubscriptionModelLoader.class)).detachAndStopAllAppenders()
+ }
+
+ def 'Callback handling on success.'() {
+ given: 'a send result'
+ def producerRecord = new ProducerRecord('topic-1', 'my value')
+ def topicPartition = new TopicPartition('topic-2', 0)
+ def recordMetadata = new RecordMetadata(topicPartition, 0, 0, 0, 0, 0)
+ def sendResult = new SendResult(producerRecord, recordMetadata)
+ when: 'the callback handler processes success'
+ def callbackHandler = objectUnderTest.handleCallback('topic-3')
+ callbackHandler.onSuccess(sendResult)
+ then: 'an event is logged with level DEBUG'
+ def loggingEvent = getLoggingEvent()
+ loggingEvent.level == Level.DEBUG
+ and: 'it contains the topic (from the record metadata) and the "value" (from the producer record)'
+ loggingEvent.formattedMessage.contains('topic-2')
+ loggingEvent.formattedMessage.contains('my value')
+ }
+
+
+ def 'Callback handling on failure.'() {
+ when: 'the callback handler processes a failure'
+ def callbackHandler = objectUnderTest.handleCallback('my topic')
+ callbackHandler.onFailure(new Exception('my exception'))
+ then: 'an event is logged with level ERROR'
+ def loggingEvent = getLoggingEvent()
+ loggingEvent.level == Level.ERROR
+ and: 'it contains the topic and exception message'
+ loggingEvent.formattedMessage.contains('my topic')
+ loggingEvent.formattedMessage.contains('my exception')
+ }
+
+ def getLoggingEvent() {
+ return loggingListAppender.list[0]
+ }
+
+
+}
@Autowired
JsonObjectMapper jsonObjectMapper
- @Autowired
- ObjectMapper objectMapper
-
def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('ncmp-group', CloudEventDeserializer))
def 'Consume and forward valid message'() {
def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json')
def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class)
def testCloudEventSent = CloudEventBuilder.v1()
- .withData(objectMapper.writeValueAsBytes(testEventSent))
+ .withData(jsonObjectMapper.asJsonBytes(testEventSent))
.withId('sample-eventid')
.withType('sample-test-type')
.withSource(URI.create('sample-test-source'))
assert result.topic == null
}
- def 'Map null subscription event to yang model subscription event where #scenario'() {
+ def 'Map empty subscription event to yang model subscription event'() {
given: 'a new Subscription Event with no data'
def testEventToMap = new SubscriptionEvent()
when: 'the event is mapped to a yang model subscription'
and: 'the topic is null'
assert result.topic == null
}
-
}
\ No newline at end of file
import com.fasterxml.jackson.databind.ObjectMapper
import com.hazelcast.map.IMap
import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.mapstruct.factory.Mappers
import org.onap.cps.ncmp.api.impl.events.EventsPublisher
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent.TargetCmHandle
import org.onap.cps.ncmp.api.inventory.InventoryPersistence
import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
import org.onap.cps.ncmp.event.model.SubscriptionEvent
IMap<String, Set<String>> mockForwardedSubscriptionEventCache = Mock(IMap<String, Set<String>>)
@SpringBean
SubscriptionEventResponseOutcome mockSubscriptionEventResponseOutcome = Mock(SubscriptionEventResponseOutcome)
+ @SpringBean
+ SubscriptionPersistence mockSubscriptionPersistence = Mock(SubscriptionPersistence)
+ @SpringBean
+ SubscriptionEventMapper subscriptionEventMapper = Mappers.getMapper(SubscriptionEventMapper)
@Autowired
JsonObjectMapper jsonObjectMapper
def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
+ and: 'the some of the cm handles will be accepted and some of rejected'
+ def cmHandlesToBeSavedInDb = [new TargetCmHandle('CMHandle1', SubscriptionStatus.ACCEPTED),
+ new TargetCmHandle('CMHandle2',SubscriptionStatus.ACCEPTED),
+ new TargetCmHandle('CMHandle3',SubscriptionStatus.REJECTED)]
+ and: 'a yang model subscription event will be saved into the db'
+ def yangModelSubscriptionEventWithAcceptedAndRejectedCmHandles = subscriptionEventMapper.toYangModelSubscriptionEvent(testEventSent)
+ yangModelSubscriptionEventWithAcceptedAndRejectedCmHandles.getPredicates().setTargetCmHandles(cmHandlesToBeSavedInDb)
and: 'the InventoryPersistence returns private properties for the supplied CM Handles'
1 * mockInventoryPersistence.getYangModelCmHandles(["CMHandle1", "CMHandle2", "CMHandle3"]) >> [
createYangModelCmHandleWithDmiProperty(1, 1,"shape","circle"),
- createYangModelCmHandleWithDmiProperty(2, 1,"shape","square"),
- createYangModelCmHandleWithDmiProperty(3, 2,"shape","triangle")
+ createYangModelCmHandleWithDmiProperty(2, 1,"shape","square")
]
and: 'the thread creation delay is reduced to 2 seconds for testing'
objectUnderTest.dmiResponseTimeoutInMs = 2000
then: 'An asynchronous call is made to the blocking variable'
block.get()
then: 'the event is added to the forwarded subscription event cache'
- 1 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set, 600, TimeUnit.SECONDS)
+ 1 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1"] as Set, 600, TimeUnit.SECONDS)
and: 'the event is forwarded twice with the CMHandle private properties and provides a valid listenable future'
1 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName1", "SCO-9989752-cm-subscription-001-DMIName1",
consumerRecord.headers(), subscriptionEvent -> {
targets["CMHandle2"] == ["shape":"square"]
}
)
- 1 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName2", "SCO-9989752-cm-subscription-001-DMIName2",
- consumerRecord.headers(), subscriptionEvent -> {
- Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0)
- targets["CMHandle3"] == ["shape":"triangle"]
- }
- )
+ and: 'the persistence service save the yang model subscription event'
+ 1 * mockSubscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEventWithAcceptedAndRejectedCmHandles)
and: 'a separate thread has been created where the map is polled'
1 * mockForwardedSubscriptionEventCache.containsKey("SCO-9989752cm-subscription-001") >> true
- 1 * mockForwardedSubscriptionEventCache.get(_) >> DMINamesInMap
1 * mockSubscriptionEventResponseOutcome.sendResponse(*_)
and: 'the subscription id is removed from the event cache map returning the asynchronous blocking variable'
1 * mockForwardedSubscriptionEventCache.remove("SCO-9989752cm-subscription-001") >> {block.set(_)}
- where:
- scenario | DMINamesInMap
- 'there are dmis which have not responded' | ["DMIName1", "DMIName2"] as Set
- 'all dmis have responded' | [] as Set
}
def 'Forward CM create subscription where target CM Handles are #scenario'() {
def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json')
def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class)
def consumerRecord = new ConsumerRecord<String, SubscriptionEvent>('topic-name', 0, 0, 'event-key', testEventSent)
+ and: 'the cm handles will be rejected'
+ def rejectedCmHandles = [new TargetCmHandle('CMHandle1', SubscriptionStatus.REJECTED),
+ new TargetCmHandle('CMHandle2',SubscriptionStatus.REJECTED),
+ new TargetCmHandle('CMHandle3',SubscriptionStatus.REJECTED)]
+ and: 'a yang model subscription event will be saved into the db with rejected cm handles'
+ def yangModelSubscriptionEventWithRejectedCmHandles = subscriptionEventMapper.toYangModelSubscriptionEvent(testEventSent)
+ yangModelSubscriptionEventWithRejectedCmHandles.getPredicates().setTargetCmHandles(rejectedCmHandles)
and: 'the InventoryPersistence returns no private properties for the supplied CM Handles'
1 * mockInventoryPersistence.getYangModelCmHandles(["CMHandle1", "CMHandle2", "CMHandle3"]) >> []
and: 'the thread creation delay is reduced to 2 seconds for testing'
objectUnderTest.forwardCreateSubscriptionEvent(testEventSent, consumerRecord.headers())
then: 'the event is not added to the forwarded subscription event cache'
0 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set)
- and: 'the event is forwarded twice with the CMHandle private properties and provides a valid listenable future'
+ and: 'the event is not being forwarded with the CMHandle private properties and does not provides a valid listenable future'
0 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName1", "SCO-9989752-cm-subscription-001-DMIName1",
consumerRecord.headers(),subscriptionEvent -> {
Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0)
0 * mockForwardedSubscriptionEventCache.get(_)
and: 'the subscription id is removed from the event cache map returning the asynchronous blocking variable'
0 * mockForwardedSubscriptionEventCache.remove("SCO-9989752cm-subscription-001") >> {block.set(_)}
+ and: 'the persistence service save target cm handles of the yang model subscription event as rejected '
+ 1 * mockSubscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEventWithRejectedCmHandles)
and: 'subscription outcome has been sent'
- 1 * mockSubscriptionEventResponseOutcome.sendResponse('SCO-9989752', 'cm-subscription-001', true)
+ 1 * mockSubscriptionEventResponseOutcome.sendResponse('SCO-9989752', 'cm-subscription-001')
}
static def createYangModelCmHandleWithDmiProperty(id, dmiId,propertyName, propertyValue) {
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistenceImpl
import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
import org.onap.cps.ncmp.api.models.SubscriptionEventResponse
+import org.onap.cps.spi.model.DataNodeBuilder
import org.onap.cps.utils.JsonObjectMapper
import org.springframework.boot.test.context.SpringBootTest
objectUnderTest.notificationFeatureEnabled = isNotificationFeatureEnabled
and: 'subscription model loader is enabled'
objectUnderTest.subscriptionModelLoaderEnabled = true
+ and: 'a data node exist in db'
+ def leaves1 = [status:'ACCEPTED', cmHandleId:'cmhandle1'] as Map
+ def dataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
+ .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+ .withLeaves(leaves1).build()
+ and: 'subscription persistence service returns data node'
+ mockSubscriptionPersistence.getCmHandlesForSubscriptionEvent(*_) >> [dataNode]
when: 'the valid event is consumed'
objectUnderTest.consumeSubscriptionEventResponse(consumerRecord)
then: 'the forwarded subscription event cache returns only the received dmiName existing for the subscription create event'
and: 'the forwarded subscription event cache returns an empty Map when the dmiName has been removed'
1 * mockForwardedSubscriptionEventCache.get('some-client-idsome-subscription-name') >> ([] as Set)
and: 'the subscription event is removed from the map'
- 1 * mockForwardedSubscriptionEventCache.remove('some-client-idsome-subscription-name')
+ numberOfExpectedCallToRemove * mockForwardedSubscriptionEventCache.remove('some-client-idsome-subscription-name')
and: 'a response outcome has been created'
- numberOfExpectedCallToSendResponse * mockSubscriptionEventResponseOutcome.sendResponse('some-client-id', 'some-subscription-name', isFullOutcomeResponse)
+ numberOfExpectedCallToSendResponse * mockSubscriptionEventResponseOutcome.sendResponse('some-client-id', 'some-subscription-name')
where: 'the following values are used'
- scenario | isNotificationFeatureEnabled | isFullOutcomeResponse || numberOfExpectedCallToSendResponse
- 'Response sent' | true | true || 1
- 'Response not sent' | true | false || 0
- 'Response not sent' | false | true || 0
- 'Response not sent' | false | false || 0
+ scenario | isNotificationFeatureEnabled || numberOfExpectedCallToRemove || numberOfExpectedCallToSendResponse
+ 'Response sent' | true || 1 || 1
+ 'Response not sent' | false || 0 || 0
}
def 'Consume Subscription Event Response where another DMI has not yet responded'() {
package org.onap.cps.ncmp.api.impl.events.avcsubscription
import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.kafka.common.header.internals.RecordHeaders
import org.mapstruct.factory.Mappers
import org.onap.cps.ncmp.api.impl.events.EventsPublisher
import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus
import org.onap.cps.ncmp.api.impl.utils.DataNodeBaseSpec
import org.onap.cps.ncmp.events.avc.subscription.v1.SubscriptionEventOutcome
import org.onap.cps.ncmp.utils.TestUtils
@Autowired
JsonObjectMapper jsonObjectMapper
+ def 'Send response to the client apps successfully'() {
+ given: 'a subscription client id and subscription name'
+ def clientId = 'some-client-id'
+ def subscriptionName = 'some-subscription-name'
+ and: 'the persistence service return a data node'
+ mockSubscriptionPersistence.getCmHandlesForSubscriptionEvent(*_) >> [dataNode4]
+ and: 'the response is being generated from the db'
+ def eventOutcome = objectUnderTest.generateResponse(clientId, subscriptionName)
+ when: 'the response is being sent'
+ objectUnderTest.sendResponse(clientId, subscriptionName)
+ then: 'the publisher publish the response with expected parameters'
+ 1 * mockSubscriptionEventOutcomePublisher.publishEvent('cm-avc-subscription-response', clientId + subscriptionName, new RecordHeaders(), eventOutcome)
+ }
+
+ def 'Check cm handle id to status map to see if it is a full outcome response'() {
+ when: 'is full outcome response evaluated'
+ def response = objectUnderTest.isFullOutcomeResponse(cmHandleIdToStatusMap)
+ then: 'the result will be as expected'
+ response == expectedResult
+ where: 'the following values are used'
+ scenario | cmHandleIdToStatusMap || expectedResult
+ 'The map contains PENDING status' | ['CMHandle1': SubscriptionStatus.PENDING] as Map || false
+ 'The map contains ACCEPTED status' | ['CMHandle1': SubscriptionStatus.ACCEPTED] as Map || true
+ 'The map contains REJECTED status' | ['CMHandle1': SubscriptionStatus.REJECTED] as Map || true
+ 'The map contains PENDING and ACCEPTED statuses' | ['CMHandle1': SubscriptionStatus.PENDING,'CMHandle2': SubscriptionStatus.ACCEPTED] as Map || false
+ 'The map contains REJECTED and ACCEPTED statuses' | ['CMHandle1': SubscriptionStatus.REJECTED,'CMHandle2': SubscriptionStatus.ACCEPTED] as Map || true
+ 'The map contains PENDING and REJECTED statuses' | ['CMHandle1': SubscriptionStatus.PENDING,'CMHandle2': SubscriptionStatus.REJECTED] as Map || false
+ }
+
def 'Generate response via fetching data nodes from database.'() {
given: 'a db call to get data nodes for subscription event'
- 1 * mockSubscriptionPersistence.getDataNodesForSubscriptionEvent() >> [dataNode4]
+ 1 * mockSubscriptionPersistence.getCmHandlesForSubscriptionEvent(*_) >> [dataNode4]
when: 'a response is generated'
- def result = objectUnderTest.generateResponse('some-client-id', 'some-subscription-name', isFullOutcomeResponse)
+ def result = objectUnderTest.generateResponse('some-client-id', 'some-subscription-name')
then: 'the result will have the same values as same as in dataNode4'
- result.eventType == expectedEventType
+ result.eventType == SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME
result.getEvent().getSubscription().getClientID() == 'some-client-id'
result.getEvent().getSubscription().getName() == 'some-subscription-name'
result.getEvent().getPredicates().getPendingTargets() == ['CMHandle3']
result.getEvent().getPredicates().getRejectedTargets() == ['CMHandle1']
result.getEvent().getPredicates().getAcceptedTargets() == ['CMHandle2']
- where: 'the following values are used'
- scenario | isFullOutcomeResponse || expectedEventType
- 'is full outcome' | true || SubscriptionEventOutcome.EventType.COMPLETE_OUTCOME
- 'is partial outcome' | false || SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME
}
def 'Form subscription outcome message with a list of cm handle id to status mapping'() {
'is full outcome' || SubscriptionEventOutcome.EventType.COMPLETE_OUTCOME
'is partial outcome' || SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME
}
+
}
\ No newline at end of file
'ADVISED to ADVISED' | ADVISED | ADVISED || 0 | 0
'READY to READY' | READY | READY || 0 | 0
'LOCKED to LOCKED' | LOCKED | LOCKED || 0 | 0
-
+ 'DELETED to ADVISED' | DELETED | ADVISED || 0 | 1
}
def 'Update and Publish Events on State Change from NO_EXISTING state to ADVISED'() {
1 * mockLcmEventsService.publishLcmEvent(cmHandleId, _, _)
}
+ def 'Update and Publish Events on State Change from DELETING to ADVISED'() {
+ given: 'Cm Handle represented as YangModelCmHandle in DELETING state'
+ yangModelCmHandle = new YangModelCmHandle(id: cmHandleId, dmiProperties: [], publicProperties: [], compositeState: compositeState)
+ when: 'update state is invoked'
+ objectUnderTest.updateCmHandleState(yangModelCmHandle, ADVISED)
+ then: 'the cm handle is saved using inventory persistence'
+ 1 * mockInventoryPersistence.saveCmHandle(yangModelCmHandle)
+ and: 'event service is called to publish event'
+ 1 * mockLcmEventsService.publishLcmEvent(cmHandleId, _, _)
+ }
+
def 'Update and Publish Events on State Change to READY'() {
given: 'Cm Handle represented as YangModelCmHandle'
compositeState = new CompositeState(cmHandleState: ADVISED)
assert (args[0] as Collection<YangModelCmHandle>).id.containsAll('cmhandle1', 'cmhandle2')
}
}
- and: 'event service is called to publish event'
+ and: 'event service is called to publish events'
2 * mockLcmEventsService.publishLcmEvent(_, _, _)
}
assert (args[0] as Map<String, CompositeState>).keySet().containsAll(['cmhandle1','cmhandle2'])
}
}
- and: 'event service is called to publish event'
+ and: 'event service is called to publish events'
2 * mockLcmEventsService.publishLcmEvent(_, _, _)
+ }
+ def 'Batch of existing cm handles is deleted'() {
+ given: 'A batch of deleted cm handles'
+ def cmHandleStateMap = setupBatch('DELETED')
+ when: 'updating a batch of changes'
+ objectUnderTest.updateCmHandleStateBatch(cmHandleStateMap)
+ then : 'existing cm handles composite state is persisted'
+ 1 * mockInventoryPersistence.saveCmHandleStateBatch(_) >> {
+ args -> {
+ assert (args[0] as Map<String, CompositeState>).isEmpty()
+ }
+ }
+ and: 'event service is called to publish events'
+ 2 * mockLcmEventsService.publishLcmEvent(_, _, _)
}
def setupBatch(type) {
return [(yangModelCmHandle1): ADVISED, (yangModelCmHandle2): ADVISED]
}
+ if ('DELETED' == type) {
+ yangModelCmHandle1.compositeState = new CompositeState(cmHandleState: READY)
+ yangModelCmHandle2.compositeState = new CompositeState(cmHandleState: READY)
+ return [(yangModelCmHandle1): DELETED, (yangModelCmHandle2): DELETED]
+ }
+
if ('UPDATE' == type) {
yangModelCmHandle1.compositeState = new CompositeState(cmHandleState: ADVISED)
yangModelCmHandle2.compositeState = new CompositeState(cmHandleState: READY)
return [(yangModelCmHandle1): ADVISED, (yangModelCmHandle2): READY]
}
}
-}
\ No newline at end of file
+}
import com.fasterxml.jackson.databind.ObjectMapper
import org.onap.cps.ncmp.api.impl.config.NcmpConfiguration
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
import org.onap.cps.ncmp.api.impl.utils.DmiServiceUrlBuilder
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext
import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent
import org.onap.cps.ncmp.utils.TestUtils
import org.onap.cps.utils.JsonObjectMapper
import org.spockframework.spring.SpringBean
import static org.onap.cps.ncmp.api.impl.operations.OperationType.UPDATE
@SpringBootTest
-@ContextConfiguration(classes = [NcmpConfiguration.DmiProperties, DmiDataOperations])
+@ContextConfiguration(classes = [EventsPublisher, CpsApplicationContext, NcmpConfiguration.DmiProperties, DmiDataOperations])
class DmiDataOperationsSpec extends DmiOperationsBaseSpec {
@SpringBean
@Autowired
DmiDataOperations objectUnderTest
+ @SpringBean
+ EventsPublisher eventsPublisher = Stub()
+
def 'call get resource data for #expectedDatastoreInUrl from DMI without topic #scenario.'() {
given: 'a cm handle for #cmHandleId'
mockYangModelCmHandleRetrieval(dmiProperties)
result.keySet() == ['CMHandle3', 'CMHandle2', 'CMHandle1'] as Set
result.values() as List == [SubscriptionStatus.PENDING, SubscriptionStatus.ACCEPTED, SubscriptionStatus.REJECTED]
}
+
+
+ def 'Get cm handle id to status map as expected from a nested data node.'() {
+ given: 'a nested data node'
+ def dataNode = new DataNodeBuilder().withDataspace('NCMP-Admin')
+ .withAnchor('AVC-Subscriptions').withXpath('/subscription-registry/subscription')
+ .withLeaves([clientID:'SCO-9989752', isTagged:false, subscriptionName:'cm-subscription-001'])
+ .withChildDataNodes([dataNode4]).build()
+ when:'cm handle id to status is being extracted'
+ def result = DataNodeHelper.getCmHandleIdToStatusMapFromDataNodes([dataNode]);
+ then: 'the keys are retrieved as expected'
+ result.keySet() == ['CMHandle3','CMHandle2','CMHandle1'] as Set
+ }
}
--- /dev/null
+package org.onap.cps.ncmp.api.impl.utils.context
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.onap.cps.utils.JsonObjectMapper
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.test.context.ContextConfiguration
+import spock.lang.Specification;
+
+@SpringBootTest(classes = [ObjectMapper, JsonObjectMapper])
+@ContextConfiguration(classes = [CpsApplicationContext.class])
+class CpsApplicationContextSpec extends Specification {
+
+ def 'Verify if cps application context contains a requested bean.'() {
+ when: 'cps bean is requested from application context'
+ def jsonObjectMapper = CpsApplicationContext.getCpsBean(JsonObjectMapper.class)
+ then: 'requested bean of JsonObjectMapper is not null'
+ assert jsonObjectMapper != null
+ }
+}
* ============LICENSE_END=========================================================
*/
-package org.onap.cps.ncmp.api.impl.utils
+package org.onap.cps.ncmp.api.impl.utils.data.operation
import com.fasterxml.jackson.databind.ObjectMapper
+import io.cloudevents.CloudEvent
+import io.cloudevents.core.CloudEventUtils
+import io.cloudevents.jackson.PojoCloudEventDataMapper
+import io.cloudevents.kafka.CloudEventDeserializer
+import io.cloudevents.kafka.impl.KafkaHeaders
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.onap.cps.ncmp.api.impl.events.EventsPublisher
+import org.onap.cps.ncmp.api.impl.utils.context.CpsApplicationContext
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
import org.onap.cps.ncmp.api.inventory.CmHandleState
import org.onap.cps.ncmp.api.inventory.CompositeStateBuilder
+import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec
import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent
import org.onap.cps.ncmp.utils.TestUtils
import org.onap.cps.utils.JsonObjectMapper
import org.spockframework.spring.SpringBean
-import spock.lang.Specification
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.context.ContextConfiguration
+import java.time.Duration
-class DataOperationRequestUtilsSpec extends Specification {
+@ContextConfiguration(classes = [EventsPublisher, CpsApplicationContext, ObjectMapper])
+class ResourceDataOperationRequestUtilsSpec extends MessagingBaseSpec {
+
+ def cloudEventKafkaConsumer = new KafkaConsumer<>(eventConsumerConfigProperties('test', CloudEventDeserializer))
+ def static clientTopic = 'my-topic-name'
+ def static dataOperationType = 'org.onap.cps.ncmp.events.async1_0_0.DataOperationEvent'
@SpringBean
JsonObjectMapper jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
+ @SpringBean
+ EventsPublisher eventPublisher = new EventsPublisher<CloudEvent>(legacyEventKafkaTemplate, cloudEventKafkaTemplate)
+
+ @Autowired
+ ObjectMapper objectMapper
+
def 'Process per data operation request with #serviceName.'() {
given: 'data operation request with 3 operations'
def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
and: '4 known cm handles: ch1-dmi1, ch2-dmi1, ch3-dmi2, ch4-dmi2'
def yangModelCmHandles = getYangModelCmHandles()
when: 'data operation request is processed'
- def operationsOutPerDmiServiceName = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(dataOperationRequest, yangModelCmHandles)
+ def operationsOutPerDmiServiceName = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(clientTopic,'request-id', dataOperationRequest, yangModelCmHandles)
and: 'converted to a json node'
def dmiDataOperationRequestBody = jsonObjectMapper.asJsonString(operationsOutPerDmiServiceName.get(serviceName))
def dmiDataOperationRequestBodyAsJsonNode = jsonObjectMapper.convertToJsonNode(dmiDataOperationRequestBody).get(operationIndex)
'dmi2' | 2 || 'operational-15' | 'ncmp-datastore:passthrough-operational' | ['ch4-dmi2']
}
+ def 'Process per data operation request with non-ready, non-existing cm handle and publish event to client specified topic'() {
+ given: 'consumer subscribing to client topic'
+ cloudEventKafkaConsumer.subscribe([clientTopic])
+ and: 'data operation request having non-ready and non-existing cm handle ids'
+ def dataOperationRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
+ def dataOperationRequest = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, DataOperationRequest.class)
+ when: 'data operation request is processed'
+ ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(clientTopic, 'request-id', dataOperationRequest, yangModelCmHandles)
+ and: 'subscribed client specified topic is polled and first record is selected'
+ def consumerRecordOut = cloudEventKafkaConsumer.poll(Duration.ofMillis(1500))[0]
+ then: 'verify cloud compliant headers'
+ def consumerRecordOutHeaders = consumerRecordOut.headers()
+ assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_id') != null
+ assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_type') == dataOperationType
+ and: 'verify that extension is included into header'
+ assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_correlationid') == 'request-id'
+ assert KafkaHeaders.getParsedKafkaHeader(consumerRecordOutHeaders, 'ce_destination') == clientTopic
+ and: 'map consumer record to expected event type'
+ def dataOperationResponseEvent = CloudEventUtils.mapData(consumerRecordOut.value(),
+ PojoCloudEventDataMapper.from(objectMapper, DataOperationEvent.class)).getValue()
+ and: 'data operation response event response size is 3'
+ dataOperationResponseEvent.data.responses.size() == 3
+ and: 'verify published response data as json string'
+ jsonObjectMapper.asJsonString(dataOperationResponseEvent.data.responses)
+ == '[{"operationId":"operational-14","ids":["unknown-cm-handle"],"statusCode":"100","statusMessage":"cm handle id(s) not found"},{"operationId":"operational-14","ids":["non-ready-cm handle"],"statusCode":"101","statusMessage":"cm handle(s) not ready"},{"operationId":"running-12","ids":["non-ready-cm handle"],"statusCode":"101","statusMessage":"cm handle(s) not ready"}]'
+ }
+
static def getYangModelCmHandles() {
def dmiProperties = [new YangModelCmHandle.Property('prop', 'some DMI property')]
def readyState = new CompositeStateBuilder().withCmHandleState(CmHandleState.READY).withLastUpdatedTimeNow().build()
+ def advisedState = new CompositeStateBuilder().withCmHandleState(CmHandleState.ADVISED).withLastUpdatedTimeNow().build()
return [new YangModelCmHandle(id: 'ch1-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
new YangModelCmHandle(id: 'ch2-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
new YangModelCmHandle(id: 'ch6-dmi1', dmiServiceName: 'dmi1', dmiProperties: dmiProperties, compositeState: readyState),
new YangModelCmHandle(id: 'ch3-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
new YangModelCmHandle(id: 'ch4-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
new YangModelCmHandle(id: 'ch7-dmi2', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: readyState),
+ new YangModelCmHandle(id: 'non-ready-cm handle', dmiServiceName: 'dmi2', dmiProperties: dmiProperties, compositeState: advisedState)
]
}
}
import ch.qos.logback.classic.Level
import ch.qos.logback.classic.Logger
import ch.qos.logback.core.read.ListAppender
-import org.junit.jupiter.api.AfterEach
-import org.junit.jupiter.api.BeforeEach
import org.onap.cps.api.CpsAdminService
import org.onap.cps.api.CpsDataService
import org.onap.cps.api.CpsModuleService
def applicationReadyEvent = new ApplicationReadyEvent(new SpringApplication(), null, applicationContext, null)
def yangResourceToContentMap
- def logger
- def appender
+ def logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
+ def loggingListAppender
- @BeforeEach
void setup() {
yangResourceToContentMap = objectUnderTest.createYangResourceToContentMap()
- logger = (Logger) LoggerFactory.getLogger(objectUnderTest.getClass())
- appender = new ListAppender()
logger.setLevel(Level.DEBUG)
- appender.start()
- logger.addAppender(appender)
+ loggingListAppender = new ListAppender()
+ logger.addAppender(loggingListAppender)
+ loggingListAppender.start()
applicationContext.refresh()
}
- @AfterEach
- void teardown() {
+ void cleanup() {
((Logger) LoggerFactory.getLogger(SubscriptionModelLoader.class)).detachAndStopAllAppenders()
applicationContext.close()
}
and: 'the data service to create a top level datanode was not called'
0 * mockCpsDataService.saveData(*_)
and: 'the log message contains the correct exception message'
- def logs = appender.list.toString()
+ def logs = loggingListAppender.list.toString()
assert logs.contains("Retrieval of NCMP dataspace fails")
}
when: 'the method to onboard model is called'
objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
then: 'the log message contains the correct exception message'
- def debugMessage = appender.list[0].toString()
+ def debugMessage = loggingListAppender.list[0].toString()
assert debugMessage.contains("Creating schema set failed")
and: 'exception is thrown'
thrown(NcmpStartUpException)
then: 'no exception thrown'
noExceptionThrown()
and: 'the log message contains the correct exception message'
- def infoMessage = appender.list[0].toString()
+ def infoMessage = loggingListAppender.list[0].toString()
assert infoMessage.contains("already exists")
}
when: 'the method to onboard model is called'
objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
then: 'the log message contains the correct exception message'
- def debugMessage = appender.list[0].toString()
+ def debugMessage = loggingListAppender.list[0].toString()
assert debugMessage.contains("Schema Set not found")
and: 'exception is thrown'
thrown(NcmpStartUpException)
then: 'no exception thrown'
noExceptionThrown()
and: 'the log message contains the correct exception message'
- def infoMessage = appender.list[0].toString()
+ def infoMessage = loggingListAppender.list[0].toString()
assert infoMessage.contains("already exists")
}
when: 'the method to onboard model is called'
objectUnderTest.onboardSubscriptionModel(yangResourceToContentMap)
then: 'the log message contains the correct exception message'
- def debugMessage = appender.list[0].toString()
+ def debugMessage = loggingListAppender.list[0].toString()
assert debugMessage.contains("Creating data node for subscription model failed: Invalid JSON")
and: 'exception is thrown'
thrown(NcmpStartUpException)
],
"statusCode": "any-success-status-code",
"statusMessage": "Successfully applied changes",
- "responseContent": {
+ "result": {
"some-key": "some-value"
}
}
"targetIds": [
"ch3-dmi2",
"unknown-cm-handle",
- "ch6-dmi1"
+ "ch6-dmi1",
+ "non-ready-cm handle"
]
},
{
"targetIds": [
"ch1-dmi1",
"ch7-dmi2",
- "ch2-dmi1"
+ "ch2-dmi1",
+ "non-ready-cm handle"
]
},
{
</configOptions>
</configuration>
</execution>
+ <execution>
+ <id>openapi-yaml-gen</id>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ <phase>compile</phase>
+ <configuration>
+ <inputSpec>${project.basedir}/docs/openapi/openapi.yml</inputSpec>
+ <generatorName>openapi-yaml</generatorName>
+ <configOptions>
+ <outputFile>openapi.yaml</outputFile>
+ </configOptions>
+ </configuration>
+ </execution>
</executions>
</plugin>
<plugin>
+ <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<executions>
<execution>
<outputDirectory>${project.basedir}/target/classes/static/api-docs/cps-core</outputDirectory>
<resources>
<resource>
- <directory>${project.basedir}/target/generated-sources/swagger/</directory>
+ <directory>${project.basedir}/target/generated-sources/openapi/</directory>
<includes>
<include>openapi.yaml</include>
</includes>
}
}
+ /**
+ * Serializing generic json object to bytes using Jackson.
+ *
+ * @param jsonObject any json object value
+ * @return the generated JSON as a byte array.
+ */
+ public byte[] asJsonBytes(final Object jsonObject) {
+ try {
+ return objectMapper.writeValueAsBytes(jsonObject);
+ } catch (final JsonProcessingException jsonProcessingException) {
+ log.error("Parsing error occurred while converting JSON object to bytes.");
+ throw new DataValidationException("Parsing error occurred while converting given JSON object to bytes.",
+ jsonProcessingException.getMessage());
+ }
+ }
+
/**
* Deserialize JSON content from given JSON content String to JsonNode.
*
def spiedObjectMapper = Spy(ObjectMapper)
def jsonObjectMapper = new JsonObjectMapper(spiedObjectMapper)
- def 'Map a structured object to json String.'() {
+ def 'Map a structured object to json #type.'() {
given: 'an object model'
def object = spiedObjectMapper.readValue(TestUtils.getResourceFileContent('bookstore.json'), Object)
when: 'the object is mapped to string'
- def content = jsonObjectMapper.asJsonString(object);
+ def content = type == 'String' ? jsonObjectMapper.asJsonString(object) : jsonObjectMapper.asJsonBytes(object)
then: 'the result is a valid json string (can be parsed)'
- def contentMap = new JsonSlurper().parseText(content)
+ def contentMap = new JsonSlurper().parseText(new String(content))
and: 'the parsed content is as expected'
assert contentMap.'test:bookstore'.'bookstore-name' == 'Chapters/Easons'
+ where: 'the following data stores are used'
+ type << ['String', 'bytes']
}
def 'Map a structured object to json String error.'() {
Code Review / cps.git / commitdiff