<description>This artifact contains dependencyManagement declarations of upstream versions.</description>
<properties>
- <groovy.version>3.0.8</groovy.version>
+ <groovy.version>3.0.9</groovy.version>
<nexusproxy>https://nexus.onap.org</nexusproxy>
<releaseNexusPath>/content/repositories/releases/</releaseNexusPath>
<snapshotNexusPath>/content/repositories/snapshots/</snapshotNexusPath>
<dependency>
<groupId>io.swagger.core.v3</groupId>
<artifactId>swagger-annotations</artifactId>
- <version>2.1.4</version>
+ <version>2.2.10</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<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",
<groupId>org.onap.cps</groupId>
<artifactId>cps-ncmp-rest</artifactId>
</dependency>
+ <!-- T E S T - D E P E N D E N C I E S -->
+ <dependency>
+ <groupId>org.spockframework</groupId>
+ <artifactId>spock-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.UUID;
import javax.validation.Valid;
import javax.validation.constraints.NotNull;
import org.onap.cps.ncmp.api.impl.operations.DatastoreType;
import org.onap.cps.ncmp.rest.api.NetworkCmProxyApi;
import org.onap.cps.ncmp.rest.model.CmHandleQueryParameters;
-import org.onap.cps.ncmp.rest.model.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.rest.model.DataOperationRequest;
import org.onap.cps.ncmp.rest.model.RestModuleDefinition;
import org.onap.cps.ncmp.rest.model.RestModuleReference;
import org.onap.cps.ncmp.rest.model.RestOutputCmHandle;
import org.onap.cps.ncmp.rest.model.RestOutputCmHandleCompositeState;
import org.onap.cps.ncmp.rest.model.RestOutputCmHandlePublicProperties;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.core.io.ClassPathResource;
+import org.onap.cps.ncmp.rest.stub.providers.ResourceProvider;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
-
@Slf4j
@RestController
@RequestMapping("${rest.api.ncmp-stub-base-path}")
public class NetworkCmProxyStubController implements NetworkCmProxyApi {
- @Value("${stub.path}")
- private String pathToResponseFiles;
+ @Autowired
+ private ResourceProvider resourceProvider;
+
+ @Autowired
+ private ObjectMapper objectMapper;
+
private static final String ASYNC_REQUEST_ID = "requestId";
@Override
final Map<String, Object> asyncResponseData = asyncResponse.getBody();
Object responseObject = null;
// read JSON file and map/convert to java POJO
- final ClassPathResource resource = new ClassPathResource(
- pathToResponseFiles + "passthrough-operational-example.json");
- try (InputStream inputStream = resource.getInputStream()) {
- final String string = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
- final ObjectMapper mapper = new ObjectMapper();
- responseObject = mapper.readValue(string, Object.class);
- } catch (final IOException exception) {
- log.error("Error reading the file.", exception);
+ try {
+ final Optional<Object> optionalResponseObject = getResponseObject(
+ "passthrough-operational-example.json", Object.class);
+ if (optionalResponseObject.isPresent()) {
+ responseObject = optionalResponseObject.get();
+ }
+
+ } catch (final IOException ioException) {
+ log.error("Error reading the file.", ioException);
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
}
+
if (asyncResponseData == null) {
return ResponseEntity.ok(responseObject);
}
@Override
public ResponseEntity<List<RestOutputCmHandle>> searchCmHandles(
final CmHandleQueryParameters cmHandleQueryParameters) {
- List<RestOutputCmHandle> restOutputCmHandles = null;
// read JSON file and map/convert to java POJO
- final ClassPathResource resource = new ClassPathResource(pathToResponseFiles + "cmHandlesSearch.json");
- try (InputStream inputStream = resource.getInputStream()) {
- final String string = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
- final ObjectMapper mapper = new ObjectMapper();
- restOutputCmHandles = Arrays.asList(mapper.readValue(string, RestOutputCmHandle[].class));
- } catch (final IOException exception) {
- log.error("Error reading the file.", exception);
+ try {
+ final Optional<RestOutputCmHandle[]> optionalResponseObject = getResponseObject("cmHandlesSearch.json",
+ RestOutputCmHandle[].class);
+ if (optionalResponseObject.isPresent()) {
+ final List<RestOutputCmHandle> restOutputCmHandles = Arrays.asList(optionalResponseObject.get());
+ return ResponseEntity.ok(restOutputCmHandles);
+ }
+ } catch (final IOException ioException) {
+ log.error("Error reading the file.", ioException);
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
}
- return ResponseEntity.ok(restOutputCmHandles);
+
+ return ResponseEntity.ok(Collections.<RestOutputCmHandle>emptyList());
}
private ResponseEntity<Map<String, Object>> populateAsyncResponse(final String topicParamInQuery) {
return asyncResponseData;
}
+ private <T> Optional<T> getResponseObject(final String filename, final Class<T> type) throws IOException {
+ final Optional<InputStream> optionalInputStream = resourceProvider.getResourceInputStream(filename);
+ if (optionalInputStream.isPresent()) {
+ final String content = new String(optionalInputStream.get().readAllBytes(), StandardCharsets.UTF_8);
+ return Optional.of(objectMapper.readValue(content, type));
+ }
+ return Optional.empty();
+ }
+
@Override
public ResponseEntity<Void> createResourceDataRunningForCmHandle(@NotNull @Valid final String resourceIdentifier,
final String datastoreName, final String cmHandle,
}
@Override
- public ResponseEntity<Object> getResourceDataForCmHandleBatch(@NotNull @Valid final String topic,
- @Valid final ResourceDataBatchRequest body) {
+ public ResponseEntity<Object> executeDataOperationForCmHandles(final String topicParamInQuery,
+ final DataOperationRequest dataOperationRequest) {
return new ResponseEntity<>(HttpStatus.NOT_IMPLEMENTED);
}
--- /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.rest.stub.providers;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Optional;
+
+public interface ResourceProvider {
+
+ Optional<InputStream> getResourceInputStream(final String filename) throws IOException;
+
+}
--- /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.rest.stub.providers;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class ResourceProviderImpl implements ResourceProvider {
+
+ private String pathToResponseFiles;
+
+ @Autowired
+ public ResourceProviderImpl(@Value("${stub.path}") final String pathToResponseFiles) {
+ this.pathToResponseFiles = pathToResponseFiles;
+ }
+
+ @Override
+ public Optional<InputStream> getResourceInputStream(final String filename) throws IOException {
+ final Path path = Paths.get(pathToResponseFiles).resolve(filename);
+
+ if (Files.exists(path)) {
+ log.info("Found resource file on file system using path: {}", path);
+ return Optional.of(Files.newInputStream(path));
+ }
+
+ log.warn("Couldn't find file on file system '{}', will search it in classpath", path);
+
+ final ClassPathResource resource = new ClassPathResource(path.toString());
+ if (resource.exists()) {
+ log.info("Found resource in classpath using path: {}", path);
+ return Optional.of(resource.getInputStream());
+ }
+
+ log.error("{} file not found on classpath or on file system", path);
+ return Optional.empty();
+ }
+
+}
--- /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.rest.stub.providers
+
+import java.nio.file.Files
+import java.nio.file.Path
+import org.springframework.util.FileSystemUtils
+import spock.lang.Shared
+import spock.lang.Specification
+import spock.lang.TempDir
+
+class ResourceProviderSpec extends Specification {
+
+ @TempDir
+ @Shared
+ def tempDirectory
+
+ def setupSpec() {
+ tempDirectory = Files.createTempDirectory('spock-test')
+ Files.write(tempDirectory.resolve('file.txt'), 'Dummy file content'.getBytes())
+ }
+
+ def cleanupSpec() {
+ if(Files.exists(tempDirectory)) {
+ FileSystemUtils.deleteRecursively(tempDirectory)
+ }
+ }
+
+ def 'Resource Provider with existing file on #scenario'() {
+
+ given: 'a resource provider with base stub folder defined on #scenario'
+ def resourceProvider = new ResourceProviderImpl(dir)
+ when: 'attempting to access that file #filename'
+ def optional= resourceProvider.getResourceInputStream(filename)
+ then: 'it is present'
+ assert optional.isPresent()
+ where:
+ scenario | dir | filename
+ 'classpath' | '/stubs/' | 'passthrough-operational-example.json'
+ 'file system' | tempDirectory.toString() | 'file.txt'
+ }
+
+ def 'Resource Provider without required resource file on #scenario'() {
+
+ given: 'a resource provider with base stub folder defined on #scenario'
+ def resourceProvider = new ResourceProviderImpl(dir)
+ when: 'attempting to access unknown-file.txt'
+ def optional= resourceProvider.getResourceInputStream('unknown-file.txt')
+ then: 'it is empty'
+ assert optional.isEmpty()
+ where:
+ scenario | dir
+ 'classpath' | '/stubs/'
+ 'file system' | tempDirectory.toString()
+ }
+
+}
<properties>
<parent.directory>${project.parent.basedir}/..</parent.directory>
<sonar.skip>true</sonar.skip>
+ <jacoco.skip>true</jacoco.skip>
</properties>
<modules>
properties:
message:
type: string
- example: "Bad Gateway Error Message NCMP"
+ example: 'Bad Gateway Error Message NCMP'
dmi-response:
type: object
properties:
example: |
module stores {
yang-version 1.1;
- namespace "org:onap:ccsdk:sample";
+ namespace 'org:onap:ccsdk:sample';
prefix book-store;
- revision "2020-09-15" {
+ revision '2020-09-15' {
description
- "Sample Model";
+ 'Sample Model';
}
}
cmHandleQueryParameters:
type: array
items:
- type: object
$ref: '#/components/schemas/ConditionProperties'
conditions:
deprecated: true
type: array
items:
- type: object
$ref: '#/components/schemas/OldConditionProperties'
description: not necessary, it is just for backward compatibility
conditionParameters:
type: array
items:
- type: object
$ref: '#/components/schemas/ModuleNameAsJsonObject'
ModuleNameAsJsonObject:
properties:
properties:
state:
$ref: '#/components/schemas/CmHandleCompositeState'
- # Batch Request Schemas
- ResourceDataBatchRequest:
+ # Data Operation Request Schemas
+ DataOperationRequest:
type: object
- title: get resource data for given array of operations
+ title: execute data operation for given array of operations
properties:
operations:
type: array
items:
- type: object
- $ref: '#/components/schemas/BatchOperationDefinition'
- description: contains batch request details
- BatchOperationDefinition:
+ $ref: '#/components/schemas/DataOperationDefinition'
+ description: contains group of data operation requests
+ DataOperationDefinition:
required:
- operation
- datastore
type: array
items:
type: string
- example: [ "da310eecdb8d44c2acc0ddaae01174b1","c748c58f8e0b438f9fd1f28370b17d47" ]
+ example: [ "da310eecdb8d44c2acc0ddaae01174b1","c748c58f8e0b438f9fd1f28370b17d47" ]
examples:
dataSampleRequest:
502:
$ref: 'components.yaml#/components/responses/BadGateway'
-getResourceDataForCmHandleBatch:
+dataOperationForCmHandle:
post:
tags:
- network-cm-proxy
- summary: Get resource data for batch of cm handle ids
+ summary: Execute a data operation for group of cm handle ids
description: This request will be handled asynchronously using messaging to the supplied topic. The rest response will be an acknowledge with a requestId to identify the relevant messages.
- operationId: getResourceDataForCmHandleBatch
+ operationId: executeDataOperationForCmHandles
parameters:
- $ref: 'components.yaml#/components/parameters/requiredTopicParamInQuery'
requestBody:
content:
application/json:
schema:
- $ref: 'components.yaml#/components/schemas/ResourceDataBatchRequest'
+ $ref: 'components.yaml#/components/schemas/DataOperationRequest'
responses:
200:
description: OK
# ============LICENSE_START=======================================================
# Copyright (C) 2021 Bell Canada
-# Modifications Copyright (C) 2022 Nordix Foundation
+# Modifications 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.
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=========================================================
-openapi: 3.0.1
+openapi: 3.0.3
info:
title: NCMP Inventory API
description: NCMP Inventory API
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=========================================================
-openapi: 3.0.1
+openapi: 3.0.3
info:
title: NCMP to CPS Proxy API
description: NCMP to CPS Proxy API
$ref: 'ncmp.yml#/resourceDataForCmHandle'
/v1/data:
- $ref: 'ncmp.yml#/getResourceDataForCmHandleBatch'
+ $ref: 'ncmp.yml#/dataOperationForCmHandle'
/v1/ch/{cm-handle}/data/ds/{datastore-name}/query:
$ref: 'ncmp.yml#/queryResourceDataForCmHandle'
<plugins>
<!-- Swagger code generation. -->
<plugin>
- <groupId>io.swagger.codegen.v3</groupId>
- <artifactId>swagger-codegen-maven-plugin</artifactId>
+ <groupId>org.openapitools</groupId>
+ <artifactId>openapi-generator-maven-plugin</artifactId>
+ <version>6.6.0</version>
<executions>
<execution>
<id>ncmp-code-gen</id>
<invokerPackage>org.onap.cps.ncmp.rest.controller</invokerPackage>
<modelPackage>org.onap.cps.ncmp.rest.model</modelPackage>
<apiPackage>org.onap.cps.ncmp.rest.api</apiPackage>
- <language>spring</language>
+ <generatorName>spring</generatorName>
<generateSupportingFiles>false</generateSupportingFiles>
+ <generateAliasAsModel>true</generateAliasAsModel>
<configOptions>
<sourceFolder>src/gen/java</sourceFolder>
<dateLibrary>java11</dateLibrary>
<interfaceOnly>true</interfaceOnly>
<useTags>true</useTags>
+ <openApiNullable>false</openApiNullable>
+ <skipDefaultInterface>true</skipDefaultInterface>
</configOptions>
</configuration>
</execution>
<invokerPackage>org.onap.cps.ncmp.rest.controller</invokerPackage>
<modelPackage>org.onap.cps.ncmp.rest.model</modelPackage>
<apiPackage>org.onap.cps.ncmp.rest.api</apiPackage>
- <language>spring</language>
+ <generatorName>spring</generatorName>
<generateSupportingFiles>false</generateSupportingFiles>
<configOptions>
<sourceFolder>src/gen/java</sourceFolder>
<dateLibrary>java11</dateLibrary>
<interfaceOnly>true</interfaceOnly>
<useTags>true</useTags>
+ <openApiNullable>false</openApiNullable>
+ <skipDefaultInterface>true</skipDefaultInterface>
</configOptions>
</configuration>
</execution>
<phase>compile</phase>
<configuration>
<inputSpec>${project.basedir}/docs/openapi/openapi-inventory.yml</inputSpec>
- <language>openapi-yaml</language>
+ <generatorName>openapi-yaml</generatorName>
<configOptions>
<outputFile>openapi-inventory.yaml</outputFile>
</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>
import org.onap.cps.ncmp.rest.controller.handlers.NcmpDatastoreRequestHandler;
import org.onap.cps.ncmp.rest.controller.handlers.NcmpPassthroughResourceRequestHandler;
import org.onap.cps.ncmp.rest.mapper.CmHandleStateMapper;
-import org.onap.cps.ncmp.rest.mapper.ResourceDataBatchRequestMapper;
+import org.onap.cps.ncmp.rest.mapper.DataOperationRequestMapper;
import org.onap.cps.ncmp.rest.model.CmHandlePublicProperties;
import org.onap.cps.ncmp.rest.model.CmHandleQueryParameters;
-import org.onap.cps.ncmp.rest.model.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.rest.model.DataOperationRequest;
import org.onap.cps.ncmp.rest.model.RestModuleDefinition;
import org.onap.cps.ncmp.rest.model.RestModuleReference;
import org.onap.cps.ncmp.rest.model.RestOutputCmHandle;
private final CmHandleStateMapper cmHandleStateMapper;
private final NcmpCachedResourceRequestHandler ncmpCachedResourceRequestHandler;
private final NcmpPassthroughResourceRequestHandler ncmpPassthroughResourceRequestHandler;
- private final ResourceDataBatchRequestMapper resourceDataBatchRequestMapper;
+ private final DataOperationRequestMapper dataOperationRequestMapper;
/**
* Get resource data from datastore.
}
@Override
- public ResponseEntity<Object> getResourceDataForCmHandleBatch(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest) {
+ public ResponseEntity<Object> executeDataOperationForCmHandles(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest) {
return ncmpPassthroughResourceRequestHandler.executeRequest(topicParamInQuery,
- resourceDataBatchRequestMapper.toResourceDataBatchRequest(resourceDataBatchRequest));
+ dataOperationRequestMapper.toDataOperationRequest(dataOperationRequest));
}
/**
/**
* Patch resource data from passthrough-running.
*
- * @param resourceIdentifier resource identifier
* @param datastoreName name of the datastore
* @param cmHandle cm handle identifier
+ * @param resourceIdentifier resource identifier
* @param requestBody the request body
* @param contentType content type of body
* @return {@code ResponseEntity} response from dmi plugin
*/
@Override
- public ResponseEntity<Object> patchResourceDataRunningForCmHandle(final String resourceIdentifier,
- final String datastoreName,
+ public ResponseEntity<Object> patchResourceDataRunningForCmHandle(final String datastoreName,
final String cmHandle,
+ final String resourceIdentifier,
final Object requestBody,
final String contentType) {
/**
* Create resource data in datastore pass-through running for given cm-handle.
*
- * @param resourceIdentifier resource identifier
* @param datastoreName name of the datastore
* @param cmHandle cm handle identifier
+ * @param resourceIdentifier resource identifier
* @param requestBody the request body
* @param contentType content type of body
* @return {@code ResponseEntity} response from dmi plugin
*/
@Override
- public ResponseEntity<Void> createResourceDataRunningForCmHandle(final String resourceIdentifier,
- final String datastoreName,
+ public ResponseEntity<Void> createResourceDataRunningForCmHandle(final String datastoreName,
final String cmHandle,
+ final String resourceIdentifier,
final Object requestBody,
final String contentType) {
/**
* Update resource data in datastore pass-through running for given cm-handle.
*
- * @param resourceIdentifier resource identifier
* @param datastoreName name of the datastore
* @param cmHandle cm handle identifier
+ * @param resourceIdentifier resource identifier
* @param requestBody the request body
* @param contentType content type of the body
* @return response entity
*/
@Override
- public ResponseEntity<Object> updateResourceDataRunningForCmHandle(final String resourceIdentifier,
- final String datastoreName,
+ public ResponseEntity<Object> updateResourceDataRunningForCmHandle(final String datastoreName,
final String cmHandle,
+ final String resourceIdentifier,
final Object requestBody,
final String contentType) {
validateDataStore(PASSTHROUGH_RUNNING, datastoreName);
import org.onap.cps.ncmp.api.impl.exception.InvalidDatastoreException;
import org.onap.cps.ncmp.api.impl.operations.DatastoreType;
import org.onap.cps.ncmp.api.impl.operations.OperationType;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.ncmp.rest.exceptions.OperationNotSupportedException;
import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
import org.onap.cps.ncmp.rest.util.TopicValidator;
}
/**
- * Executes asynchronous request for batch of cm handles to resource data.
+ * Executes asynchronous request for group of cm handles to resource data.
*
* @param topicParamInQuery the topic param in query
- * @param resourceDataBatchRequest batch request details for resource data
+ * @param dataOperationRequest data operation request details for resource data
* @return the response entity
*/
public ResponseEntity<Object> executeRequest(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest) {
- validateBatchRequest(topicParamInQuery, resourceDataBatchRequest);
+ final DataOperationRequest
+ dataOperationRequest) {
+ validateDataOperationRequest(topicParamInQuery, dataOperationRequest);
if (!notificationFeatureEnabled) {
return ResponseEntity.ok(Map.of("status",
"Asynchronous request is unavailable as notification feature is currently disabled."));
}
- return getRequestIdAndSendBatchRequestToDmiService(topicParamInQuery, resourceDataBatchRequest);
+ return getRequestIdAndSendDataOperationRequestToDmiService(topicParamInQuery, dataOperationRequest);
}
protected ResponseEntity<Object> executeTaskAsync(final String topicParamInQuery,
return executeTaskAsync(topicParamInQuery, requestId, taskSupplier);
}
- private ResponseEntity<Object> getRequestIdAndSendBatchRequestToDmiService(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest) {
+ private ResponseEntity<Object> getRequestIdAndSendDataOperationRequestToDmiService(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest) {
final String requestId = UUID.randomUUID().toString();
- sendResourceDataBatchRequestAsynchronously(topicParamInQuery, resourceDataBatchRequest, requestId);
+ sendDataOperationRequestAsynchronously(topicParamInQuery, dataOperationRequest, requestId);
return ResponseEntity.ok(Map.of("requestId", requestId));
}
- private void validateBatchRequest(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest) {
+ private void validateDataOperationRequest(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest) {
TopicValidator.validateTopicName(topicParamInQuery);
- resourceDataBatchRequest.getBatchOperationDefinitions().forEach(batchOperationDetail -> {
- if (OperationType.fromOperationName(batchOperationDetail.getOperation()) != READ) {
+ dataOperationRequest.getDataOperationDefinitions().forEach(dataOperationDetail -> {
+ if (OperationType.fromOperationName(dataOperationDetail.getOperation()) != READ) {
throw new OperationNotSupportedException(
- batchOperationDetail.getOperation() + " operation not yet supported for target ids :"
- + batchOperationDetail.getCmHandleIds());
- } else if (DatastoreType.fromDatastoreName(batchOperationDetail.getDatastore()) == OPERATIONAL) {
- throw new InvalidDatastoreException(batchOperationDetail.getDatastore()
+ dataOperationDetail.getOperation() + " operation not yet supported for target ids :"
+ + dataOperationDetail.getCmHandleIds());
+ } else if (DatastoreType.fromDatastoreName(dataOperationDetail.getDatastore()) == OPERATIONAL) {
+ throw new InvalidDatastoreException(dataOperationDetail.getDatastore()
+ " datastore is not supported for target ids : "
- + batchOperationDetail.getCmHandleIds());
+ + dataOperationDetail.getCmHandleIds());
}
});
}
import java.util.function.Supplier;
import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
@Async
@Override
- public void sendResourceDataBatchRequestAsynchronously(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest,
- final String requestId) {
- networkCmProxyDataService.requestResourceDataForCmHandleBatch(topicParamInQuery, resourceDataBatchRequest,
+ public void sendDataOperationRequestAsynchronously(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest,
+ final String requestId) {
+ networkCmProxyDataService.executeDataOperationForCmHandles(topicParamInQuery, dataOperationRequest,
requestId);
}
package org.onap.cps.ncmp.rest.controller.handlers;
import java.util.function.Supplier;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.spi.FetchDescendantsOption;
public interface TaskManagementDefaultHandler {
return NO_OBJECT_SUPPLIER;
}
- default void sendResourceDataBatchRequestAsynchronously(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest,
- final String requestId) {
+ default void sendDataOperationRequestAsynchronously(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest,
+ final String requestId) {
}
static FetchDescendantsOption getFetchDescendantsOption(final boolean includeDescendants) {
/*
* ============LICENSE_START=======================================================
* Copyright (C) 2021 Pantheon.tech
- * Modifications Copyright (C) 2021-2022 Nordix Foundation
+ * Modifications Copyright (C) 2021-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.
import org.onap.cps.ncmp.rest.controller.NetworkCmProxyController;
import org.onap.cps.ncmp.rest.controller.NetworkCmProxyInventoryController;
import org.onap.cps.ncmp.rest.model.DmiErrorMessage;
-import org.onap.cps.ncmp.rest.model.DmiErrorMessageDmiresponse;
+import org.onap.cps.ncmp.rest.model.DmiErrorMessageDmiResponse;
import org.onap.cps.ncmp.rest.model.ErrorMessage;
import org.onap.cps.spi.exceptions.AlreadyDefinedException;
import org.onap.cps.spi.exceptions.AlreadyDefinedExceptionBatch;
final HttpStatus httpStatus,
final HttpClientRequestException httpClientRequestException) {
final var dmiErrorMessage = new DmiErrorMessage();
- final var dmiErrorResponse = new DmiErrorMessageDmiresponse();
+ final var dmiErrorResponse = new DmiErrorMessageDmiResponse();
dmiErrorResponse.setHttpCode(httpClientRequestException.getHttpStatus());
dmiErrorResponse.setBody(httpClientRequestException.getDetails());
dmiErrorMessage.setMessage(httpClientRequestException.getMessage());
import org.mapstruct.Mapping;
import org.mapstruct.NullValueCheckStrategy;
import org.mapstruct.NullValuePropertyMappingStrategy;
-import org.onap.cps.ncmp.api.models.BatchOperationDefinition;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.api.models.DataOperationDefinition;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
@Mapper(componentModel = "spring", nullValueCheckStrategy = NullValueCheckStrategy.ALWAYS,
nullValuePropertyMappingStrategy = NullValuePropertyMappingStrategy.SET_TO_DEFAULT)
-public interface ResourceDataBatchRequestMapper {
+public interface DataOperationRequestMapper {
- @Mapping(source = "operations", target = "batchOperationDefinitions")
- ResourceDataBatchRequest toResourceDataBatchRequest(
- org.onap.cps.ncmp.rest.model.ResourceDataBatchRequest resourceDataBatchRequest);
+ @Mapping(source = "operations", target = "dataOperationDefinitions")
+ DataOperationRequest toDataOperationRequest(
+ org.onap.cps.ncmp.rest.model.DataOperationRequest dataOperationRequest);
@Mapping(source = "targetIds", target = "cmHandleIds")
- BatchOperationDefinition toBatchOperationDefinition(
- org.onap.cps.ncmp.rest.model.BatchOperationDefinition batchOperationDefinition);
+ DataOperationDefinition toDataOperationDefinition(
+ org.onap.cps.ncmp.rest.model.DataOperationDefinition dataOperationDefinition);
}
import org.onap.cps.ncmp.api.inventory.CompositeState
import org.onap.cps.ncmp.api.inventory.DataStoreSyncState
import org.onap.cps.ncmp.api.inventory.LockReasonCategory
-import org.onap.cps.ncmp.rest.model.BatchOperationDefinition
+import org.onap.cps.ncmp.rest.model.DataOperationRequest
+import org.onap.cps.ncmp.rest.model.DataOperationDefinition
import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle
import org.onap.cps.ncmp.rest.controller.handlers.NcmpCachedResourceRequestHandler
import org.onap.cps.ncmp.rest.controller.handlers.NcmpPassthroughResourceRequestHandler
import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor
import org.onap.cps.ncmp.rest.mapper.CmHandleStateMapper
-import org.onap.cps.ncmp.rest.mapper.ResourceDataBatchRequestMapper
-import org.onap.cps.ncmp.rest.model.ResourceDataBatchRequest
+import org.onap.cps.ncmp.rest.mapper.DataOperationRequestMapper
import org.onap.cps.ncmp.rest.util.DeprecationHelper
import org.onap.cps.spi.FetchDescendantsOption
import org.onap.cps.spi.model.ModuleDefinition
CmHandleStateMapper cmHandleStateMapper = Mappers.getMapper(CmHandleStateMapper)
@SpringBean
- ResourceDataBatchRequestMapper resourceDataBatchRequestMapper = Mappers.getMapper(ResourceDataBatchRequestMapper)
+ DataOperationRequestMapper dataOperationRequestMapper = Mappers.getMapper(DataOperationRequestMapper)
@SpringBean
CpsNcmpTaskExecutor spiedCpsTaskExecutor = Spy()
'invalid non-empty topic value in url' | 'passthrough-operational' | '&topic=1_5_*_#'
}
- def 'Get (async) batch resource data from dmi service.'() {
- given: 'batch resource data url'
+ def 'Execute (async) data operation to read data from dmi service.'() {
+ given: 'data operation url'
def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
- def resourceDataBatchRequestJsonData = jsonObjectMapper.asJsonString(
- getResourceDataBatchRequest("read", datastore.datastoreName))
- def expectedDmiResourceDataBatchRequest
- = jsonObjectMapper.convertJsonString(resourceDataBatchRequestJsonData, org.onap.cps.ncmp.api.models.ResourceDataBatchRequest.class)
- when: 'post data resource request is performed'
+ def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
+ getDataOperationRequest("read", datastore.datastoreName))
+ def expectedDmiDataOperationRequest
+ = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, org.onap.cps.ncmp.api.models.DataOperationRequest.class)
+ when: 'post data operation request is performed'
def response = mvc.perform(
post(getUrl)
.contentType(MediaType.APPLICATION_JSON)
- .content(resourceDataBatchRequestJsonData)
+ .content(dataOperationRequestJsonData)
).andReturn().response
then: 'response status is Ok'
response.status == HttpStatus.OK.value()
then: 'wait a little to allow execution of service method by task executor (on separate thread)'
Thread.sleep(100)
then: 'the service has been invoked with the correct parameters '
- 1 * mockNetworkCmProxyDataService.requestResourceDataForCmHandleBatch('my-topic-name', expectedDmiResourceDataBatchRequest, _)
+ 1 * mockNetworkCmProxyDataService.executeDataOperationForCmHandles('my-topic-name', expectedDmiDataOperationRequest, _)
where: 'the following data stores are used'
datastore << [PASSTHROUGH_RUNNING, PASSTHROUGH_OPERATIONAL]
}
- def 'Get batch resource data for #scenario from dmi service.'() {
- given: 'batch resource data url'
+ def 'Execute (async) data operation for #scenario from dmi service.'() {
+ given: 'data operation url'
def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
- def resourceDataBatchRequestJsonData = jsonObjectMapper.asJsonString(
- getResourceDataBatchRequest(operation, datastore))
+ def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
+ getDataOperationRequest(operation, datastore))
when: 'post data resource request is performed'
def response = mvc.perform(
post(getUrl)
.contentType(MediaType.APPLICATION_JSON)
- .content(resourceDataBatchRequestJsonData)
+ .content(dataOperationRequestJsonData)
).andReturn().response
then: 'response status is BAD_REQUEST'
response.status == HttpStatus.BAD_REQUEST.value()
'non-supported operation (passthrough-operational)' | PASSTHROUGH_OPERATIONAL.datastoreName | 'create'
}
- def 'Get batch resource data when notification feature is disabled for datastore: #datastore.'() {
- given: 'batch resource data url'
+ def 'Get data operation resource data when notification feature is disabled for datastore: #datastore.'() {
+ given: 'data operation url'
def getUrl = "$ncmpBasePathV1/data?topic=my-topic-name"
- def resourceDataBatchRequestJsonData = jsonObjectMapper.asJsonString(
- getResourceDataBatchRequest("read", datastore.datastoreName))
+ def dataOperationRequestJsonData = jsonObjectMapper.asJsonString(
+ getDataOperationRequest("read", datastore.datastoreName))
ncmpPassthroughResourceRequestHandler.notificationFeatureEnabled = false
when: 'post data resource request is performed'
def response = mvc.perform(
post(getUrl)
.contentType(MediaType.APPLICATION_JSON)
- .content(resourceDataBatchRequestJsonData)
+ .content(dataOperationRequestJsonData)
).andReturn().response
then: 'response status is Ok'
response.status == HttpStatus.OK.value()
return assertContainsAll(response, expectedContent)
}
- def getResourceDataBatchRequest(operation, datastore) {
- def resourceDataBatchRequest = new ResourceDataBatchRequest()
- def batchOperationDefinitions = new ArrayList()
- batchOperationDefinitions.add(getBatchOperationDefinition(operation, datastore))
- resourceDataBatchRequest.addOperationsItem(batchOperationDefinitions)
+ def getDataOperationRequest(operation, datastore) {
+ def dataOperationRequest = new DataOperationRequest()
+ def dataOperationDefinitions = new ArrayList()
+ dataOperationDefinitions.add(getDataOperationDefinition(operation, datastore))
+ dataOperationRequest.addOperationsItem(dataOperationDefinitions)
+ return dataOperationRequest
}
- def getBatchOperationDefinition(operation, datastore) {
- def batchOperationDefinition = new BatchOperationDefinition()
- batchOperationDefinition.setOperation(operation)
- batchOperationDefinition.setOperationId("operational-12")
- batchOperationDefinition.setDatastore(datastore)
- batchOperationDefinition.setOptions("some option")
- batchOperationDefinition.setResourceIdentifier("some resource identifier")
- batchOperationDefinition.addTargetIdsItem("some-cm-handle")
- return batchOperationDefinition
+ def getDataOperationDefinition(operation, datastore) {
+ def dataOperationDefinition = new DataOperationDefinition()
+ dataOperationDefinition.setOperation(operation)
+ dataOperationDefinition.setOperationId("operational-12")
+ dataOperationDefinition.setDatastore(datastore)
+ dataOperationDefinition.setOptions("some option")
+ dataOperationDefinition.setResourceIdentifier("some resource identifier")
+ dataOperationDefinition.addTargetIdsItem("some-cm-handle")
+ return dataOperationDefinition
}
}
/*
* ============LICENSE_START=======================================================
* Copyright (C) 2021 highstreet technologies GmbH
- * Modifications Copyright (C) 2021-2022 Nordix Foundation
+ * Modifications Copyright (C) 2021-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.
import org.onap.cps.ncmp.rest.controller.handlers.NcmpPassthroughResourceRequestHandler
import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor
import org.onap.cps.ncmp.rest.mapper.CmHandleStateMapper
-import org.onap.cps.ncmp.rest.mapper.ResourceDataBatchRequestMapper
+import org.onap.cps.ncmp.rest.mapper.DataOperationRequestMapper
import org.onap.cps.ncmp.rest.util.DeprecationHelper
import org.onap.cps.spi.exceptions.AlreadyDefinedException
import org.onap.cps.spi.exceptions.AlreadyDefinedExceptionBatch
CmHandleStateMapper cmHandleStateMapper = Mappers.getMapper(CmHandleStateMapper)
@SpringBean
- ResourceDataBatchRequestMapper resourceDataBatchRequestMapper = Mappers.getMapper(ResourceDataBatchRequestMapper)
+ DataOperationRequestMapper dataOperationRequestMapper = Mappers.getMapper(DataOperationRequestMapper)
@SpringBean
CpsNcmpTaskExecutor stubbedCpsTaskExecutor = Stub()
--- /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;
+ }
+}
import org.onap.cps.ncmp.api.inventory.CompositeState;
import org.onap.cps.ncmp.api.models.CmHandleQueryApiParameters;
import org.onap.cps.ncmp.api.models.CmHandleQueryServiceParameters;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.ncmp.api.models.DmiPluginRegistration;
import org.onap.cps.ncmp.api.models.DmiPluginRegistrationResponse;
import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
import org.onap.cps.spi.FetchDescendantsOption;
import org.onap.cps.spi.model.ModuleDefinition;
import org.onap.cps.spi.model.ModuleReference;
FetchDescendantsOption fetchDescendantsOption);
/**
- * Get resource data for batch of cm handles using dmi.
+ * Execute (async) data operation for group of cm handles using dmi.
*
* @param topicParamInQuery topic name for (triggering) async responses
- * @param resourceDataBatchRequest cm handle identifiers
+ * @param dataOperationRequest contains a list of operation definitions(multiple operations)
*/
- void requestResourceDataForCmHandleBatch(String topicParamInQuery,
- ResourceDataBatchRequest
- resourceDataBatchRequest,
- String requestId);
+ void executeDataOperationForCmHandles(String topicParamInQuery,
+ DataOperationRequest dataOperationRequest,
+ String requestId);
/**
import org.onap.cps.ncmp.api.models.CmHandleQueryServiceParameters;
import org.onap.cps.ncmp.api.models.CmHandleRegistrationResponse;
import org.onap.cps.ncmp.api.models.CmHandleRegistrationResponse.RegistrationError;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.ncmp.api.models.DmiPluginRegistration;
import org.onap.cps.ncmp.api.models.DmiPluginRegistrationResponse;
import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
import org.onap.cps.spi.FetchDescendantsOption;
import org.onap.cps.spi.exceptions.AlreadyDefinedExceptionBatch;
import org.onap.cps.spi.exceptions.CpsException;
}
@Override
- public void requestResourceDataForCmHandleBatch(final String topicParamInQuery,
- final ResourceDataBatchRequest
- resourceDataBatchRequest,
- final String requestId) {
- dmiDataOperations.requestResourceDataFromDmi(topicParamInQuery, resourceDataBatchRequest, requestId);
+ public void executeDataOperationForCmHandles(final String topicParamInQuery,
+ final DataOperationRequest
+ dataOperationRequest,
+ final String requestId) {
+ dmiDataOperations.requestResourceDataFromDmi(topicParamInQuery, dataOperationRequest, requestId);
}
@Override
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@Configuration
@EnableKafka
@RequiredArgsConstructor
-public class KafkaTemplateConfig<T> {
+public class KafkaConfig<T> {
private final KafkaProperties kafkaProperties;
return new DefaultKafkaConsumerFactory<>(consumerConfigProperties);
}
+ /**
+ * A legacy Kafka event template for executing high-level operations. The legacy producer factory ensure this.
+ *
+ * @return an instance of legacy Kafka template.
+ */
+ @Bean
+ @Primary
+ public KafkaTemplate<String, T> legacyEventKafkaTemplate() {
+ final KafkaTemplate<String, T> kafkaTemplate = new KafkaTemplate<>(legacyEventProducerFactory());
+ kafkaTemplate.setConsumerFactory(legacyEventConsumerFactory());
+ return kafkaTemplate;
+ }
+
+ /**
+ * A legacy concurrent kafka listener container factory.
+ *
+ * @return instance of Concurrent kafka listener factory
+ */
+ @Bean
+ public ConcurrentKafkaListenerContainerFactory<String, T> legacyEventConcurrentKafkaListenerContainerFactory() {
+ final ConcurrentKafkaListenerContainerFactory<String, T> containerFactory =
+ new ConcurrentKafkaListenerContainerFactory<>();
+ containerFactory.setConsumerFactory(legacyEventConsumerFactory());
+ return containerFactory;
+ }
+
/**
* This sets the strategy for creating cloud Kafka producer instance from kafka properties defined into
* application.yml with CloudEventSerializer.
return new DefaultKafkaConsumerFactory<>(consumerConfigProperties);
}
- /**
- * A legacy Kafka event template for executing high-level operations. The legacy producer factory ensure this.
- *
- * @return an instance of legacy Kafka template.
- */
- @Bean
- @Primary
- public KafkaTemplate<String, T> legacyEventKafkaTemplate() {
- final KafkaTemplate<String, T> kafkaTemplate = new KafkaTemplate<>(legacyEventProducerFactory());
- kafkaTemplate.setConsumerFactory(legacyEventConsumerFactory());
- return kafkaTemplate;
- }
/**
* A cloud Kafka event template for executing high-level operations. The cloud producer factory ensure this.
return kafkaTemplate;
}
+ /**
+ * A Concurrent CloudEvent kafka listener container factory.
+ *
+ * @return instance of Concurrent kafka listener factory
+ */
+ @Bean
+ public ConcurrentKafkaListenerContainerFactory<String, CloudEvent>
+ cloudEventConcurrentKafkaListenerContainerFactory() {
+ final ConcurrentKafkaListenerContainerFactory<String, CloudEvent> containerFactory =
+ new ConcurrentKafkaListenerContainerFactory<>();
+ containerFactory.setConsumerFactory(cloudEventConsumerFactory());
+ return containerFactory;
+ }
+
}
--- /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();
+ }
+}
*
* @param avcEventConsumerRecord Incoming raw consumer record
*/
- @KafkaListener(topics = "${app.dmi.cm-events.topic}")
+ @KafkaListener(topics = "${app.dmi.cm-events.topic}",
+ containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory")
public void consumeAndForward(final ConsumerRecord<String, CloudEvent> avcEventConsumerRecord) {
log.debug("Consuming AVC event {} ...", avcEventConsumerRecord.value());
final String newEventId = UUID.randomUUID().toString();
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,
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 java.util.List;
import lombok.Builder;
import lombok.Getter;
-import org.onap.cps.ncmp.api.models.BatchOperationDefinition;
+import org.onap.cps.ncmp.api.models.DataOperationDefinition;
@JsonInclude(JsonInclude.Include.NON_NULL)
@Getter
@Builder
@JsonPropertyOrder({"operation", "operationId", "datastore", "options", "resourceIdentifier", "cmHandles"})
-public class DmiBatchOperation {
+public class DmiDataOperation {
@JsonProperty("operation")
private OperationType operationType;
private final List<CmHandle> cmHandles = new ArrayList<>();
/**
- * Create and initialise a (outgoing) DMI batch operation.
+ * Create and initialise a (outgoing) DMI data operation.
*
- * @param batchOperationDefinition batchOperationDefinition definition of incoming of batch request
+ * @param dataOperationDefinition definition of incoming of dataOperation request
* @return mapped dmi operation details
*/
- public static DmiBatchOperation buildDmiBatchRequestBodyWithoutCmHandles(
- final BatchOperationDefinition batchOperationDefinition) {
+ public static DmiDataOperation buildDmiDataOperationRequestBodyWithoutCmHandles(
+ final DataOperationDefinition dataOperationDefinition) {
- return DmiBatchOperation.builder()
- .operationType(OperationType.fromOperationName(batchOperationDefinition.getOperation()))
- .operationId(batchOperationDefinition.getOperationId())
- .datastore(DatastoreType.fromDatastoreName(batchOperationDefinition.getDatastore()).getDatastoreName())
- .options(batchOperationDefinition.getOptions())
- .resourceIdentifier(batchOperationDefinition.getResourceIdentifier())
+ return DmiDataOperation.builder()
+ .operationType(OperationType.fromOperationName(dataOperationDefinition.getOperation()))
+ .operationId(dataOperationDefinition.getOperationId())
+ .datastore(DatastoreType.fromDatastoreName(dataOperationDefinition.getDatastore()).getDatastoreName())
+ .options(dataOperationDefinition.getOptions())
+ .resourceIdentifier(dataOperationDefinition.getResourceIdentifier())
.build();
}
}
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.ResourceDataBatchRequestUtils;
+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;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
import org.onap.cps.spi.exceptions.CpsException;
import org.onap.cps.utils.JsonObjectMapper;
import org.springframework.http.ResponseEntity;
* The data wil be returned as message on the topic specified.
*
* @param topicParamInQuery topic name for (triggering) async responses
- * @param resourceDataBatchRequest batch request for resource data
+ * @param dataOperationRequest data operation request to execute operations
* @param requestId requestId for as a response
*/
public void requestResourceDataFromDmi(final String topicParamInQuery,
- final ResourceDataBatchRequest resourceDataBatchRequest,
+ final DataOperationRequest dataOperationRequest,
final String requestId) {
final Set<String> cmHandlesIds
- = getDistinctCmHandleIdsFromBatchRequest(resourceDataBatchRequest);
+ = getDistinctCmHandleIdsFromDataOperationRequest(dataOperationRequest);
final Collection<YangModelCmHandle> yangModelCmHandles
- = getYangModelCmHandlesInReadyState(cmHandlesIds);
+ = inventoryPersistence.getYangModelCmHandles(cmHandlesIds);
- final Map<String, List<DmiBatchOperation>> operationsOutPerDmiServiceName
- = ResourceDataBatchRequestUtils.processPerOperationInBatchRequest(resourceDataBatchRequest,
- yangModelCmHandles);
+ final Map<String, List<DmiDataOperation>> operationsOutPerDmiServiceName
+ = ResourceDataOperationRequestUtils.processPerDefinitionInDataOperationsRequest(topicParamInQuery,
+ requestId, dataOperationRequest, yangModelCmHandles);
- buildBatchRequestUrlAndSendToDmiService(topicParamInQuery, requestId, operationsOutPerDmiServiceName);
+ buildDataOperationRequestUrlAndSendToDmiService(topicParamInQuery, requestId, operationsOutPerDmiServiceName);
}
/**
cmHandleId));
}
- private String getDmiServiceBatchRequestUrl(final String dmiServiceName,
- final String topicParamInQuery,
- final String requestId) {
- final MultiValueMap<String, String> batchRequestQueryParams = dmiServiceUrlBuilder
- .getBatchRequestQueryParams(topicParamInQuery, requestId);
- return dmiServiceUrlBuilder.getBatchRequestUrl(batchRequestQueryParams,
- dmiServiceUrlBuilder.populateBatchUriVariables(dmiServiceName));
+ private String getDmiServiceDataOperationRequestUrl(final String dmiServiceName,
+ final String topicParamInQuery,
+ final String requestId) {
+ final MultiValueMap<String, String> dataOperationRequestQueryParams = dmiServiceUrlBuilder
+ .getDataOperationRequestQueryParams(topicParamInQuery, requestId);
+ return dmiServiceUrlBuilder.getDataOperationRequestUrl(dataOperationRequestQueryParams,
+ dmiServiceUrlBuilder.populateDataOperationRequestUriVariables(dmiServiceName));
}
private void validateIfCmHandleStateReady(final YangModelCmHandle yangModelCmHandle,
}
}
- private static Set<String> getDistinctCmHandleIdsFromBatchRequest(final ResourceDataBatchRequest
- resourceDataBatchRequest) {
- return resourceDataBatchRequest.getBatchOperationDefinitions().stream()
- .flatMap(batchOperationDefinition ->
- batchOperationDefinition.getCmHandleIds().stream()).collect(Collectors.toSet());
+ private static Set<String> getDistinctCmHandleIdsFromDataOperationRequest(final DataOperationRequest
+ dataOperationRequest) {
+ return dataOperationRequest.getDataOperationDefinitions().stream()
+ .flatMap(dataOperationDefinition ->
+ 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 buildBatchRequestUrlAndSendToDmiService(final String topicParamInQuery,
- final String requestId,
- final Map<String, List<DmiBatchOperation>>
+ private void buildDataOperationRequestUrlAndSendToDmiService(final String topicParamInQuery,
+ final String requestId,
+ final Map<String, List<DmiDataOperation>>
groupsOutPerDmiServiceName) {
groupsOutPerDmiServiceName.entrySet().forEach(groupsOutPerDmiServiceNameEntry -> {
final String dmiServiceName = groupsOutPerDmiServiceNameEntry.getKey();
- final List<DmiBatchOperation> dmiBatchRequestBodies = groupsOutPerDmiServiceNameEntry.getValue();
- final String dmiBatchResourceDataUrl = getDmiServiceBatchRequestUrl(dmiServiceName, topicParamInQuery,
- requestId);
- sendBatchRequestToDmiService(dmiBatchResourceDataUrl, dmiBatchRequestBodies);
+ final List<DmiDataOperation> dmiDataOperationRequestBodies = groupsOutPerDmiServiceNameEntry.getValue();
+ final String dmiDataOperationResourceUrl =
+ getDmiServiceDataOperationRequestUrl(dmiServiceName, topicParamInQuery, requestId);
+ sendDataOperationRequestToDmiService(dmiDataOperationResourceUrl, dmiDataOperationRequestBodies);
});
}
- private void sendBatchRequestToDmiService(final String batchResourceDataUrl,
- final List<DmiBatchOperation> dmiBatchRequestBodies) {
- final String batchRequestBodiesAsJsonString = jsonObjectMapper.asJsonString(dmiBatchRequestBodies);
- TaskExecutor.executeTask(() -> dmiRestClient.postOperationWithJsonData(batchResourceDataUrl,
- batchRequestBodiesAsJsonString, READ), DEFAULT_ASYNC_TASK_EXECUTOR_TIMEOUT_IN_MILLISECONDS)
+ private void sendDataOperationRequestToDmiService(final String dataOperationResourceUrl,
+ final List<DmiDataOperation> dmiDataOperationRequestBodies) {
+ final String dataOperationRequestBodiesAsJsonString =
+ jsonObjectMapper.asJsonString(dmiDataOperationRequestBodies);
+ TaskExecutor.executeTask(() -> dmiRestClient.postOperationWithJsonData(dataOperationResourceUrl,
+ dataOperationRequestBodiesAsJsonString, READ),
+ DEFAULT_ASYNC_TASK_EXECUTOR_TIMEOUT_IN_MILLISECONDS)
.whenCompleteAsync(this::handleTaskCompletion);
}
}
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)));
+ }
}
}
/**
- * This method builds batch request url.
+ * This method builds data operation request url.
*
- * @param batchRequestQueryParams query param map as key, value pair
- * @param batchRequestUriVariables uri param map as key (placeholder), value pair
- * @return {@code String} batch request url as string
+ * @param dataoperationRequestQueryParams query param map as key, value pair
+ * @param dataoperationRequestUriVariables uri param map as key (placeholder), value pair
+ * @return {@code String} data operation request url as string
*/
- public String getBatchRequestUrl(final MultiValueMap<String, String> batchRequestQueryParams,
- final Map<String, Object> batchRequestUriVariables) {
- return getBatchResourceDataBasePathUriBuilder()
- .queryParams(batchRequestQueryParams)
- .uriVariables(batchRequestUriVariables)
+ public String getDataOperationRequestUrl(final MultiValueMap<String, String> dataoperationRequestQueryParams,
+ final Map<String, Object> dataoperationRequestUriVariables) {
+ return getDataOperationResourceDataBasePathUriBuilder()
+ .queryParams(dataoperationRequestQueryParams)
+ .uriVariables(dataoperationRequestUriVariables)
.buildAndExpand().toUriString();
}
}
/**
- * This method creates the dmi service url builder object with path variables for batch of cm handles.
+ * This method creates the dmi service url builder object with path variables for data operation request.
*
* @return {@code UriComponentsBuilder} dmi service url builder object
*/
- public UriComponentsBuilder getBatchResourceDataBasePathUriBuilder() {
+ public UriComponentsBuilder getDataOperationResourceDataBasePathUriBuilder() {
return UriComponentsBuilder.newInstance()
.path("{dmiServiceName}")
.pathSegment("{dmiBasePath}")
}
/**
- * This method populates uri variables for batch request.
+ * This method populates uri variables for data operation request.
*
* @param dmiServiceName dmi service name
* @return {@code Map<String, Object>} uri variables as map
*/
- public Map<String, Object> populateBatchUriVariables(final String dmiServiceName) {
+ public Map<String, Object> populateDataOperationRequestUriVariables(final String dmiServiceName) {
final Map<String, Object> uriVariables = new HashMap<>();
final String dmiBasePath = dmiProperties.getDmiBasePath();
uriVariables.put("dmiServiceName", dmiServiceName);
}
/**
- * This method is used to populate map from query params for batch request.
+ * This method is used to populate map from query params for data operation request.
*
* @param topicParamInQuery topic into url param
* @param requestId unique id of response for valid topic
* @return all valid query params as map
*/
- public MultiValueMap<String, String> getBatchRequestQueryParams(final String topicParamInQuery,
- final String requestId) {
+ public MultiValueMap<String, String> getDataOperationRequestQueryParams(final String topicParamInQuery,
+ final String requestId) {
final MultiValueMap<String, String> queryParams = new LinkedMultiValueMap<>();
getQueryParamConsumer().accept("topic", topicParamInQuery, queryParams);
getQueryParamConsumer().accept("requestId", requestId, queryParams);
/*
* ============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.DmiBatchOperation;
-import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
-import org.onap.cps.ncmp.api.models.BatchOperationDefinition;
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest;
-
-@Slf4j
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class ResourceDataBatchRequestUtils {
-
- private static final String UNKNOWN_SERVICE_NAME = null;
-
- /**
- * Create a list of DMI batch operation per DMI service (name).
- *
- * @param resourceDataBatchRequestIn incoming batch request details for resource data
- * @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<DmiBatchOperation>> processPerOperationInBatchRequest(
- final ResourceDataBatchRequest resourceDataBatchRequestIn,
- 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<DmiBatchOperation>> dmiBatchOperationsOutPerDmiServiceName = new HashMap<>();
-
- for (final BatchOperationDefinition batchOperationDefinitionIn :
- resourceDataBatchRequestIn.getBatchOperationDefinitions()) {
- for (final String cmHandleId : batchOperationDefinitionIn.getCmHandleIds()) {
- final String dmiServiceName = dmiServiceNamesPerCmHandleId.get(cmHandleId);
- final Map<String, String> cmHandleIdProperties
- = dmiPropertiesPerCmHandleIdPerServiceName.get(dmiServiceName).get(cmHandleId);
- if (cmHandleIdProperties == null) {
- publishErrorMessageToClientTopic(cmHandleId);
- } else {
- final DmiBatchOperation dmiBatchOperationOut = getOrAddDmiBatchOperation(dmiServiceName,
- batchOperationDefinitionIn, dmiBatchOperationsOutPerDmiServiceName);
- final CmHandle cmHandle = CmHandle.buildCmHandleWithProperties(cmHandleId, cmHandleIdProperties);
- dmiBatchOperationOut.getCmHandles().add(cmHandle);
- }
- }
- }
- return dmiBatchOperationsOutPerDmiServiceName;
- }
-
- 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 DmiBatchOperation getOrAddDmiBatchOperation(final String dmiServiceName,
- final BatchOperationDefinition
- batchOperationDefinitionIn,
- final Map<String, List<DmiBatchOperation>>
- dmiBatchOperationsOutPerDmiServiceName) {
- dmiBatchOperationsOutPerDmiServiceName
- .computeIfAbsent(dmiServiceName, dmiServiceNameAsKey -> new ArrayList<>());
- final List<DmiBatchOperation> dmiBatchOperationsOut
- = dmiBatchOperationsOutPerDmiServiceName.get(dmiServiceName);
- final boolean isNewOperation = dmiBatchOperationsOut.isEmpty()
- || !dmiBatchOperationsOut.get(dmiBatchOperationsOut.size() - 1).getOperationId()
- .equals(batchOperationDefinitionIn.getOperationId());
- if (isNewOperation) {
- final DmiBatchOperation newDmiBatchOperationOut =
- DmiBatchOperation.buildDmiBatchRequestBodyWithoutCmHandles(batchOperationDefinitionIn);
- dmiBatchOperationsOut.add(newDmiBatchOperationOut);
- return newDmiBatchOperationOut;
- }
- return dmiBatchOperationsOut.get(dmiBatchOperationsOut.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));
+ }
+ }
+}
@EqualsAndHashCode
@JsonInclude(JsonInclude.Include.NON_EMPTY)
@JsonIgnoreProperties(ignoreUnknown = true)
-public class BatchOperationDefinition {
+public class DataOperationDefinition {
private String operation;
private String operationId;
@JsonProperty("targetIds")
@Valid
- private List<String> cmHandleIds = new ArrayList();
+ private List<String> cmHandleIds = new ArrayList<>();
}
@EqualsAndHashCode
@JsonInclude(JsonInclude.Include.NON_EMPTY)
@JsonIgnoreProperties(ignoreUnknown = true)
-public class ResourceDataBatchRequest {
+public class DataOperationRequest {
@JsonProperty("operations")
@Valid
- private List<BatchOperationDefinition> batchOperationDefinitions = Collections.emptyList();
+ private List<DataOperationDefinition> dataOperationDefinitions = Collections.emptyList();
}
import org.onap.cps.ncmp.api.inventory.InventoryPersistence
import org.onap.cps.ncmp.api.inventory.LockReasonCategory
import org.onap.cps.ncmp.api.inventory.DataStoreSyncState
-import org.onap.cps.ncmp.api.models.BatchOperationDefinition
+import org.onap.cps.ncmp.api.models.DataOperationDefinition
import org.onap.cps.ncmp.api.models.CmHandleQueryApiParameters
import org.onap.cps.ncmp.api.models.CmHandleQueryServiceParameters
import org.onap.cps.ncmp.api.models.ConditionApiProperties
import org.onap.cps.ncmp.api.models.DmiPluginRegistration
import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle
-import org.onap.cps.ncmp.api.models.ResourceDataBatchRequest
+import org.onap.cps.ncmp.api.models.DataOperationRequest
import org.onap.cps.spi.exceptions.CpsException
import org.onap.cps.spi.model.ConditionProperties
import spock.lang.Shared
response == '{dmi-response}'
}
- def 'Get batch resource data for #datastoreName from DMI.'() {
+ def 'Execute (async) data operation for #datastoreName from DMI.'() {
given: 'cpsDataService returns valid data node'
- def resourceDataBatchRequest = getResourceDataBatchRequest(datastoreName)
- when: 'get batch resource data is called'
- objectUnderTest.requestResourceDataForCmHandleBatch('some topic', resourceDataBatchRequest, 'requestId')
- then: 'get batch resource data returns expected response'
- 1 * mockDmiDataOperations.requestResourceDataFromDmi('some topic', resourceDataBatchRequest, 'requestId')
+ def dataOperationRequest = getDataOperationRequest(datastoreName)
+ when: 'request resource data for data operation is called'
+ objectUnderTest.executeDataOperationForCmHandles('some topic', dataOperationRequest, 'requestId')
+ then: 'request resource data for data operation returns expected response'
+ 1 * mockDmiDataOperations.requestResourceDataFromDmi('some topic', dataOperationRequest, 'requestId')
where: 'the following data stores are used'
datastoreName << [PASSTHROUGH_RUNNING.datastoreName, PASSTHROUGH_OPERATIONAL.datastoreName]
}
cmHandleXPath, FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >> dataNode
}
- def getResourceDataBatchRequest(datastore) {
- def resourceDataBatchRequest = new ResourceDataBatchRequest()
- def batchOperationDefinitions = new ArrayList()
- batchOperationDefinitions.add(getBatchOperationDefinition(datastore))
- resourceDataBatchRequest.setBatchOperationDefinitions(batchOperationDefinitions)
+ def getDataOperationRequest(datastore) {
+ def dataOperationRequest = new DataOperationRequest()
+ def dataOperationDefinitions = new ArrayList()
+ dataOperationDefinitions.add(getDataOperationDefinition(datastore))
+ dataOperationRequest.setDataOperationDefinitions(dataOperationDefinitions)
+ return dataOperationRequest
}
- def getBatchOperationDefinition(datastore) {
- def batchOperationDefinition = new BatchOperationDefinition()
- batchOperationDefinition.setOperation("read")
- batchOperationDefinition.setOperationId("operational-12")
- batchOperationDefinition.setDatastore(datastore)
+ def getDataOperationDefinition(datastore) {
+ def dataOperationDefinition = new DataOperationDefinition()
+ dataOperationDefinition.setOperation("read")
+ dataOperationDefinition.setOperationId("operational-12")
+ dataOperationDefinition.setDatastore(datastore)
def targetIds = new ArrayList()
targetIds.add("some-cm-handle")
- batchOperationDefinition.setCmHandleIds(targetIds)
- return batchOperationDefinition
+ dataOperationDefinition.setCmHandleIds(targetIds)
+ return dataOperationDefinition
}
}
KafkaProducer<String, CloudEvent> producer = new KafkaProducer<>(eventProducerConfigProperties(CloudEventSerializer))
producer.send(record)
and: 'wait a little for async processing of message'
- TimeUnit.MILLISECONDS.sleep(100)
+ TimeUnit.MILLISECONDS.sleep(300)
then: 'the event has only been forwarded for the correct type'
expectedNUmberOfCallsToPublishForwardedEvent * mockEventsPublisher.publishCloudEvent(*_)
where: 'the following event types are used'
KafkaProducer<String, String> producer = new KafkaProducer<>(eventProducerConfigProperties(StringSerializer))
producer.send(record)
and: 'wait a little for async processing of message'
- TimeUnit.MILLISECONDS.sleep(100)
+ TimeUnit.MILLISECONDS.sleep(300)
then: 'the event is not processed by this consumer'
0 * mockEventsPublisher.publishCloudEvent(*_)
}
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)
import spock.lang.Shared
import spock.lang.Specification
-@SpringBootTest(classes = [KafkaProperties, KafkaTemplateConfig])
+@SpringBootTest(classes = [KafkaProperties, KafkaConfig])
@EnableSharedInjection
@EnableConfigurationProperties
-class KafkaTemplateConfigSpec extends Specification {
+class KafkaConfigSpec extends Specification {
@Shared
@Autowired
--- /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
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.models.ResourceDataBatchRequest
+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)
'datastore running with properties' | [yangModelCmHandleProperty] | PASSTHROUGH_RUNNING | OPTIONS_PARAM || '{"operation":"read","cmHandleProperties":{"prop1":"val1"}}' | 'passthrough-running' | '&options=(a=1,b=2)'
}
- def 'call get batch resource data from DMI service #scenario.'() {
- given: 'collection of yang model cm Handles and resource data batch request'
+ def 'Execute (async) data operation from DMI service.'() {
+ given: 'collection of yang model cm Handles and data operation request'
mockYangModelCmHandleCollectionRetrieval([yangModelCmHandleProperty])
- def resourceDataBatchRequestJsonData = TestUtils.getResourceFileContent('resourceDataBatchRequest.json')
- def resourceDataBatchRequest = spiedJsonObjectMapper.convertJsonString(resourceDataBatchRequestJsonData, ResourceDataBatchRequest.class)
- resourceDataBatchRequest.batchOperationDefinitions[0].cmHandleIds = [cmHandleId]
+ def dataOperationBatchRequestJsonData = TestUtils.getResourceFileContent('dataOperationRequest.json')
+ def dataOperationRequest = spiedJsonObjectMapper.convertJsonString(dataOperationBatchRequestJsonData, DataOperationRequest.class)
+ dataOperationRequest.dataOperationDefinitions[0].cmHandleIds = [cmHandleId]
def requestBodyAsJsonStringArg = null
and: 'a positive response from DMI service when it is called with valid request parameters'
def responseFromDmi = new ResponseEntity<Object>(HttpStatus.ACCEPTED)
def expectedDmiBatchResourceDataUrl = "ncmp/v1/data/topic=my-topic-name"
def expectedBatchRequestAsJson = '[{"operation":"read","operationId":"operational-14","datastore":"ncmp-datastore:passthrough-operational","options":"some option","resourceIdentifier":"some resource identifier","cmHandles":[{"id":"some-cm-handle","cmHandleProperties":{"prop1":"val1"}}]}]'
mockDmiRestClient.postOperationWithJsonData(expectedDmiBatchResourceDataUrl, _, READ.operationName) >> responseFromDmi
- dmiServiceUrlBuilder.getBatchRequestUrl(_, _) >> expectedDmiBatchResourceDataUrl
- when: 'get resource data for batch of cm handles are invoked'
- objectUnderTest.requestResourceDataFromDmi('my-topic-name', resourceDataBatchRequest, 'requestId')
+ dmiServiceUrlBuilder.getDataOperationRequestUrl(_, _) >> expectedDmiBatchResourceDataUrl
+ when: 'get resource data for group of cm handles are invoked'
+ objectUnderTest.requestResourceDataFromDmi('my-topic-name', dataOperationRequest, 'requestId')
then: 'wait a little to allow execution of service method by task executor (on separate thread)'
Thread.sleep(100)
then: 'validate ncmp generated dmi request body json args'
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
+ }
}
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
import org.onap.cps.ncmp.api.impl.config.NcmpConfiguration
import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle
-import spock.lang.Shared
import spock.lang.Specification
class DmiServiceUrlBuilderSpec extends Specification {
- @Shared
- YangModelCmHandle yangModelCmHandle = YangModelCmHandle.toYangModelCmHandle('dmiServiceName',
+ static YangModelCmHandle yangModelCmHandle = YangModelCmHandle.toYangModelCmHandle('dmiServiceName',
'dmiDataServiceName', 'dmiModuleServiceName', new NcmpServiceCmHandle(cmHandleId: 'some-cm-handle-id'))
NcmpConfiguration.DmiProperties dmiProperties = new NcmpConfiguration.DmiProperties()
def objectUnderTest = new DmiServiceUrlBuilder(dmiProperties, mockCpsValidator)
+ def setup() {
+ dmiProperties.dmiBasePath = 'dmi'
+ }
+
def 'Create the dmi service url with #scenario.'() {
given: 'uri variables'
- dmiProperties.dmiBasePath = 'dmi'
- def uriVars = objectUnderTest.populateUriVariables(PASSTHROUGH_RUNNING.datastoreName, yangModelCmHandle.resolveDmiServiceName(RequiredDmiService.DATA),
- "cmHandle")
+ def uriVars = objectUnderTest.populateUriVariables(PASSTHROUGH_RUNNING.datastoreName, yangModelCmHandle.resolveDmiServiceName(RequiredDmiService.DATA), 'cmHandle')
and: 'query params'
- def uriQueries = objectUnderTest.populateQueryParams(resourceId,
- 'optionsParamInQuery', topic)
+ def uriQueries = objectUnderTest.populateQueryParams(resourceId, 'optionsParamInQuery', topic)
when: 'a dmi datastore service url is generated'
def dmiServiceUrl = objectUnderTest.getDmiDatastoreUrl(uriQueries, uriVars)
then: 'service url is generated as expected'
def 'Populate dmi data store url #scenario.'() {
given: 'uri variables are created'
dmiProperties.dmiBasePath = dmiBasePath
- def uriVars = objectUnderTest.populateUriVariables(PASSTHROUGH_RUNNING.datastoreName, yangModelCmHandle.resolveDmiServiceName(RequiredDmiService.DATA),
- "cmHandle")
+ def uriVars = objectUnderTest.populateUriVariables(PASSTHROUGH_RUNNING.datastoreName, yangModelCmHandle.resolveDmiServiceName(RequiredDmiService.DATA), 'cmHandle')
and: 'null query params'
- def uriQueries = objectUnderTest.populateQueryParams(null,
- null, null)
+ def uriQueries = objectUnderTest.populateQueryParams(null, null, null)
when: 'a dmi datastore service url is generated'
def dmiServiceUrl = objectUnderTest.getDmiDatastoreUrl(uriQueries, uriVars)
then: 'the created dmi service url matches the expected'
'with base path / ' | 'Invalid base path as it starts with /' | '/dmi' || 'dmiServiceName//dmi/v1/ch/cmHandle/data/ds/ncmp-datastore:passthrough-running'
'without base path / ' | 'Valid path as it does not starts with /' | 'dmi' || 'dmiServiceName/dmi/v1/ch/cmHandle/data/ds/ncmp-datastore:passthrough-running'
}
+
+ def 'Bath request Url creation.'() {
+ given: 'the required path parameters'
+ def batchRequestUriVariables = [dmiServiceName: 'some-service', dmiBasePath: 'testBase', cmHandleId: '123']
+ and: 'the relevant query parameters'
+ def batchRequestQueryParams = objectUnderTest.getDataOperationRequestQueryParams('some topic', 'some id')
+ when: 'a URL is created'
+ def result = objectUnderTest.getDataOperationRequestUrl(batchRequestQueryParams, batchRequestUriVariables)
+ then: 'it is formed correctly'
+ assert result.toString() == 'some-service/testBase/v1/data?topic=some topic&requestId=some id'
+ }
+
+ def 'Populate batch uri variables.'() {
+ expect: 'Populate batch uri variables returns a map with given service name and base path from setup'
+ assert objectUnderTest.populateDataOperationRequestUriVariables('some service') == [dmiServiceName: 'some service', dmiBasePath: 'dmi' ]
+ }
}
+++ /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 com.fasterxml.jackson.databind.ObjectMapper
-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.models.ResourceDataBatchRequest
-import org.onap.cps.ncmp.utils.TestUtils
-import org.onap.cps.utils.JsonObjectMapper
-import org.spockframework.spring.SpringBean
-import spock.lang.Specification
-
-class ResourceDataBatchRequestUtilsSpec extends Specification {
-
- @SpringBean
- JsonObjectMapper jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
-
- def 'Process per operation in batch request with #serviceName.'() {
- given: 'batch request with 3 operations'
- def resourceDataBatchRequestJsonData = TestUtils.getResourceFileContent('resourceDataBatchRequest.json')
- def resourceDataBatchRequest = jsonObjectMapper.convertJsonString(resourceDataBatchRequestJsonData, ResourceDataBatchRequest.class)
- and: '4 known cm handles: ch1-dmi1, ch2-dmi1, ch3-dmi2, ch4-dmi2'
- def yangModelCmHandles = getYangModelCmHandles()
- when: 'Operation in batch request is processed'
- def operationsOutPerDmiServiceName = ResourceDataBatchRequestUtils.processPerOperationInBatchRequest(resourceDataBatchRequest, yangModelCmHandles)
- and: 'converted to a json node'
- def dmiBatchRequestBody = jsonObjectMapper.asJsonString(operationsOutPerDmiServiceName.get(serviceName))
- def dmiBatchRequestBodyAsJsonNode = jsonObjectMapper.convertToJsonNode(dmiBatchRequestBody).get(operationIndex)
- then: 'it contains the correct operation details'
- assert dmiBatchRequestBodyAsJsonNode.get('operation').asText() == 'read'
- assert dmiBatchRequestBodyAsJsonNode.get('operationId').asText() == expectedOperationId
- assert dmiBatchRequestBodyAsJsonNode.get('datastore').asText() == expectedDatastore
- and: 'the correct cm handles (just for #serviceName)'
- assert dmiBatchRequestBodyAsJsonNode.get('cmHandles').size() == expectedCmHandleIds.size()
- expectedCmHandleIds.each {
- dmiBatchRequestBodyAsJsonNode.get('cmHandles').toString().contains(it)
- }
- where: 'the following dmi service and operations are checked'
- serviceName | operationIndex || expectedOperationId | expectedDatastore | expectedCmHandleIds
- 'dmi1' | 0 || 'operational-14' | 'ncmp-datastore:passthrough-operational' | ['ch6-dmi1']
- 'dmi1' | 1 || 'running-12' | 'ncmp-datastore:passthrough-running' | ['ch1-dmi1', 'ch2-dmi1']
- 'dmi1' | 2 || 'operational-15' | 'ncmp-datastore:passthrough-operational' | ['ch6-dmi1']
- 'dmi2' | 0 || 'operational-14' | 'ncmp-datastore:passthrough-operational' | ['ch3-dmi2']
- 'dmi2' | 1 || 'running-12' | 'ncmp-datastore:passthrough-running' | ['ch7-dmi2']
- 'dmi2' | 2 || 'operational-15' | 'ncmp-datastore:passthrough-operational' | ['ch4-dmi2']
- }
-
- static def getYangModelCmHandles() {
- def dmiProperties = [new YangModelCmHandle.Property('prop', 'some DMI property')]
- def readyState = new CompositeStateBuilder().withCmHandleState(CmHandleState.READY).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: 'ch8-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),
- ]
- }
-}
--- /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
+ }
+}
--- /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 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 org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.context.ContextConfiguration
+import java.time.Duration
+
+@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')
+ def dataOperationRequest = jsonObjectMapper.convertJsonString(dataOperationRequestJsonData, DataOperationRequest.class)
+ 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(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)
+ then: 'it contains the correct operation details'
+ assert dmiDataOperationRequestBodyAsJsonNode.get('operation').asText() == 'read'
+ assert dmiDataOperationRequestBodyAsJsonNode.get('operationId').asText() == expectedOperationId
+ assert dmiDataOperationRequestBodyAsJsonNode.get('datastore').asText() == expectedDatastore
+ and: 'the correct cm handles (just for #serviceName)'
+ assert dmiDataOperationRequestBodyAsJsonNode.get('cmHandles').size() == expectedCmHandleIds.size()
+ expectedCmHandleIds.each {
+ dmiDataOperationRequestBodyAsJsonNode.get('cmHandles').toString().contains(it)
+ }
+ where: 'the following dmi service and operations are checked'
+ serviceName | operationIndex || expectedOperationId | expectedDatastore | expectedCmHandleIds
+ 'dmi1' | 0 || 'operational-14' | 'ncmp-datastore:passthrough-operational' | ['ch6-dmi1']
+ 'dmi1' | 1 || 'running-12' | 'ncmp-datastore:passthrough-running' | ['ch1-dmi1', 'ch2-dmi1']
+ 'dmi1' | 2 || 'operational-15' | 'ncmp-datastore:passthrough-operational' | ['ch6-dmi1']
+ 'dmi2' | 0 || 'operational-14' | 'ncmp-datastore:passthrough-operational' | ['ch3-dmi2']
+ 'dmi2' | 1 || 'running-12' | 'ncmp-datastore:passthrough-running' | ['ch7-dmi2']
+ '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: 'ch8-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"
]
},
{
# ============LICENSE_START=======================================================
-# Copyright (C) 2021 Nordix Foundation
+# Copyright (C) 2021-2023 Nordix Foundation
# Modifications Copyright (C) 2021 Pantheon.tech
# Modifications Copyright (C) 2021 Bell Canada.
# Modifications Copyright (C) 2022-2023 TechMahindra Ltd.
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=========================================================
-openapi: 3.0.1
+openapi: 3.0.3
info:
title: ONAP Open API v3 Configuration Persistence Service
description: Configuration Persistence Service is a Model Driven Generic Database
<plugins>
<!-- Swagger code generation. -->
<plugin>
- <groupId>io.swagger.codegen.v3</groupId>
- <artifactId>swagger-codegen-maven-plugin</artifactId>
+ <groupId>org.openapitools</groupId>
+ <artifactId>openapi-generator-maven-plugin</artifactId>
+ <version>6.6.0</version>
<executions>
<execution>
<id>code-gen</id>
<invokerPackage>org.onap.cps.rest.controller</invokerPackage>
<modelPackage>org.onap.cps.rest.model</modelPackage>
<apiPackage>org.onap.cps.rest.api</apiPackage>
- <language>spring</language>
+ <generatorName>spring</generatorName>
<generateSupportingFiles>false</generateSupportingFiles>
<configOptions>
<sourceFolder>src/gen/java</sourceFolder>
<dateLibrary>java11</dateLibrary>
<interfaceOnly>true</interfaceOnly>
<useTags>true</useTags>
+ <openApiNullable>false</openApiNullable>
+ <skipDefaultInterface>true</skipDefaultInterface>
+ </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>
/**
* Create a {@link SchemaSet}.
*
- * @param multipartFile multipart file
- * @param schemaSetName schemaset name
* @param dataspaceName dataspace name
+ * @param schemaSetName schemaset name
+ * @param multipartFile multipart file
* @return a {@Link ResponseEntity} of created schemaset name & {@link HttpStatus} CREATED
*/
@Override
- public ResponseEntity<String> createSchemaSet(@NotNull @Valid final String schemaSetName,
- final String dataspaceName, @Valid final MultipartFile multipartFile) {
+ public ResponseEntity<String> createSchemaSet(final String dataspaceName,
+ @NotNull @Valid final String schemaSetName,
+ final MultipartFile multipartFile) {
cpsModuleService.createSchemaSet(dataspaceName, schemaSetName, extractYangResourcesMap(multipartFile));
return new ResponseEntity<>(schemaSetName, HttpStatus.CREATED);
}
/**
* Create a {@link SchemaSet}.
*
- * @param multipartFile multipart file
- * @param schemaSetName schemaset name
* @param dataspaceName dataspace name
+ * @param schemaSetName schemaset name
+ * @param multipartFile multipart file
* @return a {@Link ResponseEntity} of created schema set without any response body & {@link HttpStatus} CREATED
*/
@Override
@Timed(value = "cps.rest.admin.controller.schemaset.create",
description = "Time taken to create schemaset from controller")
- public ResponseEntity<Void> createSchemaSetV2(@NotNull @Valid final String schemaSetName,
- final String dataspaceName, @Valid final MultipartFile multipartFile) {
+ public ResponseEntity<Void> createSchemaSetV2(final String dataspaceName,
+ @NotNull @Valid final String schemaSetName,
+ final MultipartFile multipartFile) {
cpsModuleService.createSchemaSet(dataspaceName, schemaSetName, extractYangResourcesMap(multipartFile));
return new ResponseEntity<>(HttpStatus.CREATED);
}
* ============LICENSE_START=======================================================
* Copyright (C) 2020-2022 Bell Canada.
* Modifications Copyright (C) 2021 Pantheon.tech
- * Modifications Copyright (C) 2021-2022 Nordix Foundation
+ * Modifications Copyright (C) 2021-2023 Nordix Foundation
* Modifications Copyright (C) 2022-2023 TechMahindra Ltd.
* Modifications Copyright (C) 2022 Deutsche Telekom AG
* ================================================================================
private final PrefixResolver prefixResolver;
@Override
- public ResponseEntity<String> createNode(@RequestHeader(value = "Content-Type") final String contentTypeHeader,
- final String apiVersion,
+ public ResponseEntity<String> createNode(final String apiVersion,
final String dataspaceName, final String anchorName,
+ @RequestHeader(value = "Content-Type") final String contentTypeHeader,
final String nodeData, final String parentNodeXpath,
final String observedTimestamp) {
final ContentType contentType = contentTypeHeader.contains(MediaType.APPLICATION_XML_VALUE) ? ContentType.XML
}
@Override
- public ResponseEntity<String> addListElements(final String parentNodeXpath, final String apiVersion,
- final String dataspaceName, final String anchorName, final Object jsonData, final String observedTimestamp) {
+ public ResponseEntity<String> addListElements(final String apiVersion, final String dataspaceName,
+ final String anchorName, final String parentNodeXpath,
+ final Object jsonData, final String observedTimestamp) {
cpsDataService.saveListElements(dataspaceName, anchorName, parentNodeXpath,
jsonObjectMapper.asJsonString(jsonData), toOffsetDateTime(observedTimestamp));
return new ResponseEntity<>(HttpStatus.CREATED);
}
@Override
- public ResponseEntity<Object> replaceListContent(final String parentNodeXpath,
- final String apiVersion, final String dataspaceName, final String anchorName, final Object jsonData,
+ public ResponseEntity<Object> replaceListContent(final String apiVersion,
+ final String dataspaceName, final String anchorName,
+ final String parentNodeXpath, final Object jsonData,
final String observedTimestamp) {
cpsDataService.replaceListContent(dataspaceName, anchorName, parentNodeXpath,
jsonObjectMapper.asJsonString(jsonData), toOffsetDateTime(observedTimestamp));
}
}
+ /**
+ * 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.'() {
<exclude>org/onap/cps/rest/model/*</exclude>
<exclude>org/onap/cps/cpspath/parser/antlr4/*</exclude>
<exclude>org/onap/cps/ncmp/rest/model/*</exclude>
- <exclude>org/onap/cps/ncmp/rest/controller/*MapperImpl.class</exclude>
- <exclude>org/onap/cps/rest/controller/*MapperImpl.class</exclude>
- <exclude>org/onap/cps/ncmp/api/impl/async/*MapperImpl.class</exclude>
+ <exclude>org/onap/cps/**/*MapperImpl.class</exclude>
<exclude>org/onap/cps/ncmp/rest/stub/*</exclude>
</excludes>
</configuration>
Code Review / cps.git / commitdiff