Refactor and Move NCMP Data Request Handlers

- Simplified NCMP Rest Request Handlers
- Moved responsEntity wrapping to Controller so most handler methods can have clear return types
- Moved NCMP Rest Request Handlers  to Service Layer
- Moved related exceptions and utils to Service Layer
- Used Lombok for constructors
- Improved related testware

Issue-ID: CPS-2266
Change-Id: I0025fab1c92e0d613825093b6e4b43dae044c01a
Signed-off-by: ToineSiebelink <toine.siebelink@est.tech>

9 files changed, 621 insertions, 0 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpCachedResourceRequestHandler.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpCachedResourceRequestHandler.java
new file mode 100644
index 000000000..eb43718f0
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpCachedResourceRequestHandler.java
@@ -0,0 +1,72 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022-2024 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;
+
+import java.util.Collection;
+import lombok.RequiredArgsConstructor;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+import org.onap.cps.ncmp.api.NetworkCmProxyQueryService;
+import org.onap.cps.ncmp.api.models.CmResourceAddress;
+import org.onap.cps.spi.FetchDescendantsOption;
+import org.onap.cps.spi.model.DataNode;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+@Service
+@RequiredArgsConstructor
+public class NcmpCachedResourceRequestHandler extends NcmpDatastoreRequestHandler {
+
+    private final NetworkCmProxyDataService networkCmProxyDataService;
+    private final NetworkCmProxyQueryService networkCmProxyQueryService;
+
+    /**
+     * Executes a synchronous query request for given cm handle.
+     * Note. Currently only ncmp-datastore:operational supports query operations.
+     *
+     * @param cmHandleId         the cm handle
+     * @param resourceIdentifier the resource identifier
+     * @param includeDescendants whether include descendants
+     * @return a collection of data nodes
+     */
+    public Collection<DataNode> executeRequest(final String cmHandleId, final String resourceIdentifier,
+                                                 final boolean includeDescendants) {
+        final FetchDescendantsOption fetchDescendantsOption = getFetchDescendantsOption(includeDescendants);
+        return networkCmProxyQueryService.queryResourceDataOperational(cmHandleId, resourceIdentifier,
+            fetchDescendantsOption);
+    }
+
+    @Override
+    protected Mono<Object> getResourceDataForCmHandle(final CmResourceAddress cmResourceAddress,
+                                                      final String optionsParamInQuery,
+                                                      final String topicParamInQuery,
+                                                      final String requestId,
+                                                      final boolean includeDescendants,
+                                                      final String authorization) {
+        final FetchDescendantsOption fetchDescendantsOption = getFetchDescendantsOption(includeDescendants);
+        return Mono.fromSupplier(
+                () -> networkCmProxyDataService.getResourceDataForCmHandle(cmResourceAddress, fetchDescendantsOption));
+    }
+
+    private static FetchDescendantsOption getFetchDescendantsOption(final boolean includeDescendants) {
+        return includeDescendants ? FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS
+            : FetchDescendantsOption.OMIT_DESCENDANTS;
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandler.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandler.java
new file mode 100644
index 000000000..dbd2bb493
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandler.java
@@ -0,0 +1,99 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022-2024 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;
+
+import java.util.Map;
+import java.util.UUID;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.api.models.CmResourceAddress;
+import org.onap.cps.ncmp.utils.TopicValidator;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+@Slf4j
+@Service
+public abstract class NcmpDatastoreRequestHandler {
+
+    private static final String NO_REQUEST_ID = null;
+    private static final String NO_TOPIC = null;
+
+    @Value("${notification.async.executor.time-out-value-in-ms:60000}")
+    protected int timeOutInMilliSeconds;
+    @Value("${notification.enabled:true}")
+    protected boolean notificationFeatureEnabled;
+
+    /**
+     * Executes synchronous/asynchronous get request for given cm handle.
+     *
+     * @param cmResourceAddress   the name of the datastore, cm handle and resource identifier
+     * @param optionsParamInQuery the options param in query
+     * @param topicParamInQuery   the topic param in query
+     * @param includeDescendants  whether include descendants
+     * @param authorization       contents of Authorization header, or null if not present
+     * @return the result object, depends on use op topic. With topic a map object with request id is returned
+     *         otherwise the result of the request.
+     */
+    public Object executeRequest(final CmResourceAddress cmResourceAddress,
+                                                 final String optionsParamInQuery,
+                                                 final String topicParamInQuery,
+                                                 final boolean includeDescendants,
+                                                 final String authorization) {
+
+        final boolean asyncResponseRequested = topicParamInQuery != null;
+        if (asyncResponseRequested && notificationFeatureEnabled) {
+            return fetchResourceDataAsynchronously(cmResourceAddress, optionsParamInQuery, topicParamInQuery,
+                includeDescendants, authorization);
+        }
+
+        if (asyncResponseRequested) {
+            log.warn("Asynchronous request is unavailable as notification feature is currently disabled, "
+                    + "will use synchronous operation.");
+        }
+        final Mono<Object> resourceDataMono = getResourceDataForCmHandle(cmResourceAddress, optionsParamInQuery,
+                NO_TOPIC, NO_REQUEST_ID, includeDescendants, authorization);
+        return resourceDataMono.block();
+    }
+
+    private Map<String, String> fetchResourceDataAsynchronously(final CmResourceAddress cmResourceAddress,
+                                                                   final String optionsParamInQuery,
+                                                                   final String topicParamInQuery,
+                                                                   final boolean includeDescendants,
+                                                                   final String authorization) {
+        TopicValidator.validateTopicName(topicParamInQuery);
+        final String requestId = UUID.randomUUID().toString();
+        getResourceDataForCmHandle(cmResourceAddress, optionsParamInQuery, topicParamInQuery, requestId,
+                includeDescendants, authorization)
+                .doOnSuccess(result -> log.debug("Async operation succeeded for request id {}: {}", requestId, result))
+                .doOnError(error ->
+                        log.error("Async operation failed for request id {}: {}", requestId, error.getMessage()))
+                .subscribe();
+        log.debug("Received Async request with id {}", requestId);
+        return Map.of("requestId", requestId);
+    }
+
+    protected abstract Mono<Object> getResourceDataForCmHandle(final CmResourceAddress cmResourceAddress,
+                                                               final String optionsParamInQuery,
+                                                               final String topicParamInQuery,
+                                                               final String requestId,
+                                                               final boolean includeDescendant,
+                                                               final String authorization);
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpPassthroughResourceRequestHandler.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpPassthroughResourceRequestHandler.java
new file mode 100644
index 000000000..90d9a23d6
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/NcmpPassthroughResourceRequestHandler.java
@@ -0,0 +1,103 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022-2024 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;
+
+import static org.onap.cps.ncmp.api.impl.operations.DatastoreType.OPERATIONAL;
+import static org.onap.cps.ncmp.api.impl.operations.OperationType.READ;
+
+import java.util.Map;
+import java.util.UUID;
+import lombok.RequiredArgsConstructor;
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService;
+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.CmResourceAddress;
+import org.onap.cps.ncmp.api.models.DataOperationRequest;
+import org.onap.cps.ncmp.exceptions.OperationNotSupportedException;
+import org.onap.cps.ncmp.exceptions.PayloadTooLargeException;
+import org.onap.cps.ncmp.utils.TopicValidator;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+@Service
+@RequiredArgsConstructor
+public class NcmpPassthroughResourceRequestHandler extends NcmpDatastoreRequestHandler {
+
+    private final NetworkCmProxyDataService networkCmProxyDataService;
+    private static final int MAXIMUM_CM_HANDLES_PER_OPERATION = 200;
+    private static final String PAYLOAD_TOO_LARGE_TEMPLATE = "Operation '%s' affects too many (%d) cm handles";
+
+    /**
+     * Executes asynchronous request for group of cm handles to resource data.
+     *
+     * @param topicParamInQuery        the topic param in query
+     * @param dataOperationRequest     data operation request details for resource data
+     * @param authorization            contents of Authorization header, or null if not present
+     * @return a map with one entry of request Id for success or status and error when async feature is disabled
+     */
+    public Map<String, String> executeRequest(final String topicParamInQuery,
+                                              final DataOperationRequest dataOperationRequest,
+                                              final String authorization) {
+        validateDataOperationRequest(topicParamInQuery, dataOperationRequest);
+        if (!notificationFeatureEnabled) {
+            return Map.of("status",
+                "Asynchronous request is unavailable as notification feature is currently disabled.");
+        }
+        final String requestId = UUID.randomUUID().toString();
+        networkCmProxyDataService.executeDataOperationForCmHandles(topicParamInQuery, dataOperationRequest, requestId,
+            authorization);
+        return Map.of("requestId", requestId);
+
+    }
+
+    @Override
+    protected Mono<Object> getResourceDataForCmHandle(final CmResourceAddress cmResourceAddress,
+                                                      final String optionsParamInQuery,
+                                                      final String topicParamInQuery,
+                                                      final String requestId,
+                                                      final boolean includeDescendants,
+                                                      final String authorization) {
+        return networkCmProxyDataService.getResourceDataForCmHandle(cmResourceAddress, optionsParamInQuery,
+                topicParamInQuery, requestId, authorization);
+    }
+
+    private void validateDataOperationRequest(final String topicParamInQuery,
+                                              final DataOperationRequest dataOperationRequest) {
+        TopicValidator.validateTopicName(topicParamInQuery);
+        dataOperationRequest.getDataOperationDefinitions().forEach(dataOperationDetail -> {
+            if (OperationType.fromOperationName(dataOperationDetail.getOperation()) != READ) {
+                throw new OperationNotSupportedException(
+                        dataOperationDetail.getOperation() + " operation not yet supported");
+            }
+            if (DatastoreType.fromDatastoreName(dataOperationDetail.getDatastore()) == OPERATIONAL) {
+                throw new InvalidDatastoreException(dataOperationDetail.getDatastore()
+                        + " datastore is not supported");
+            }
+            if (dataOperationDetail.getCmHandleIds().size() > MAXIMUM_CM_HANDLES_PER_OPERATION) {
+                final String errorMessage = String.format(PAYLOAD_TOO_LARGE_TEMPLATE,
+                        dataOperationDetail.getOperationId(),
+                        dataOperationDetail.getCmHandleIds().size());
+                throw new PayloadTooLargeException(errorMessage);
+            }
+        });
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/InvalidTopicException.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/InvalidTopicException.java
new file mode 100644
index 000000000..fcf2a28cc
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/InvalidTopicException.java
@@ -0,0 +1,40 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.exceptions;
+
+import lombok.Getter;
+
+public class InvalidTopicException extends RuntimeException {
+
+    @Getter
+    final String details;
+
+    /**
+     * Constructor.
+     *
+     * @param message the error message
+     * @param details the error details
+     */
+    public InvalidTopicException(final String message, final String details) {
+        super(message);
+        this.details = details;
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/OperationNotSupportedException.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/OperationNotSupportedException.java
new file mode 100644
index 000000000..d75c0bd47
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/OperationNotSupportedException.java
@@ -0,0 +1,32 @@
+/*
+ *  ============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.exceptions;
+
+public class OperationNotSupportedException extends RuntimeException {
+    /**
+     * Instantiates a new not implemented operation exception.
+     *
+     * @param message the message
+     */
+    public OperationNotSupportedException(final String message) {
+        super(message);
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/PayloadTooLargeException.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/PayloadTooLargeException.java
new file mode 100644
index 000000000..dc7057af7
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/exceptions/PayloadTooLargeException.java
@@ -0,0 +1,31 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 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.exceptions;
+
+public class PayloadTooLargeException extends RuntimeException {
+
+    /**
+     * Instantiates a new payload too large exception.
+     */
+    public PayloadTooLargeException(final String message) {
+        super(message);
+    }
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/TopicValidator.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/TopicValidator.java
new file mode 100644
index 000000000..f9fed8d43
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/utils/TopicValidator.java
@@ -0,0 +1,47 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.utils;
+
+import java.util.regex.Pattern;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.onap.cps.ncmp.exceptions.InvalidTopicException;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class TopicValidator {
+
+    private static final Pattern TOPIC_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9]([._-](?![._-])|"
+        + "[a-zA-Z0-9]){0,120}[a-zA-Z0-9]$");
+
+    /**
+     * Validate kafka topic name pattern.
+     *
+     * @param topicName name of the topic to be validated
+     *
+     * @throws InvalidTopicException if the topic is not valid
+     */
+    public static void validateTopicName(final String topicName) {
+        if (!TOPIC_NAME_PATTERN.matcher(topicName).matches()) {
+            throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
+        }
+    }
+
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandlerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandlerSpec.groovy
new file mode 100644
index 000000000..b73f9a46d
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/NcmpDatastoreRequestHandlerSpec.groovy
@@ -0,0 +1,150 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023-2024 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
+
+import org.onap.cps.ncmp.api.NetworkCmProxyDataService
+import org.onap.cps.ncmp.api.impl.exception.InvalidDatastoreException
+import org.onap.cps.ncmp.api.impl.exception.InvalidOperationException
+import org.onap.cps.ncmp.api.models.CmResourceAddress
+import org.onap.cps.ncmp.api.models.DataOperationDefinition
+import org.onap.cps.ncmp.api.models.DataOperationRequest
+import org.onap.cps.ncmp.exceptions.OperationNotSupportedException
+import org.onap.cps.ncmp.exceptions.PayloadTooLargeException
+import org.springframework.http.HttpStatus
+import org.springframework.http.ResponseEntity
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+
+class NcmpDatastoreRequestHandlerSpec extends Specification {
+
+    def mockNetworkCmProxyDataService = Mock(NetworkCmProxyDataService)
+
+    def objectUnderTest = new NcmpPassthroughResourceRequestHandler(mockNetworkCmProxyDataService)
+
+    def NO_AUTH_HEADER = null
+
+    def setup() {
+        objectUnderTest.timeOutInMilliSeconds = 100
+    }
+
+    def 'Attempt to execute async get request with #scenario.'() {
+        given: 'notification feature is turned on/off'
+            objectUnderTest.notificationFeatureEnabled = notificationFeatureEnabled
+        and: 'a CM resource address'
+            def cmResourceAddress = new CmResourceAddress('ds', 'ch1', 'resource1')
+        and: 'the (mocked) service when called with the correct parameters returns a response from dmi'
+            def resultFromDmi = new ResponseEntity('response from dmi',HttpStatus.I_AM_A_TEAPOT)
+            def synchronousResult = Mono.justOrEmpty(resultFromDmi)
+            mockNetworkCmProxyDataService.getResourceDataForCmHandle(cmResourceAddress, 'options', _, _, NO_AUTH_HEADER) >> synchronousResult
+        when: 'get request is executed with topic = #topic'
+            def response = objectUnderTest.executeRequest(cmResourceAddress, 'options', topic, false, NO_AUTH_HEADER)
+        then: 'a successful result with/without request id is returned'
+            if (expectSynchronousResponse) {
+                assert response.toString().contains('response from dmi')
+                assert response.toString().contains("I'm a teapot")
+            } else {
+                // expect request id in a map
+                assert response.keySet()[0] == 'requestId'
+            }
+        where: 'the following parameters are used'
+            scenario                   | notificationFeatureEnabled | topic   || expectSynchronousResponse
+            'feature on, valid topic'  | true                       | 'valid' || false
+            'feature on, no topic'     | true                       | null    || true
+            'feature off, valid topic' | false                      | 'valid' || true
+            'feature off, no topic'    | false                      | null    || true
+    }
+
+    def 'Attempt to execute async data operation request with feature #scenario.'() {
+        given: 'a extended request handler that supports bulk requests'
+           def objectUnderTest = new NcmpPassthroughResourceRequestHandler(mockNetworkCmProxyDataService)
+        and: 'notification feature is turned on/off'
+            objectUnderTest.notificationFeatureEnabled = notificationFeatureEnabled
+        when: 'data operation request is executed'
+            objectUnderTest.executeRequest('someTopic', new DataOperationRequest(), NO_AUTH_HEADER)
+        then: 'the task is executed in an async fashion or not'
+            expectedCalls * mockNetworkCmProxyDataService.executeDataOperationForCmHandles('someTopic', _, _, null)
+        where: 'the following parameters are used'
+            scenario | notificationFeatureEnabled || expectedCalls
+            'on'     | true                       || 1
+            'off'    | false                      || 0
+    }
+
+    def 'Execute async data operation request with datastore #datastore.'() {
+        given: 'notification feature is turned on'
+            objectUnderTest.notificationFeatureEnabled = true
+        and: 'a data operation request with datastore: #datastore'
+            def dataOperationDefinition = new DataOperationDefinition(operation: 'read', datastore: datastore)
+            def dataOperationRequest = new DataOperationRequest(dataOperationDefinitions: [dataOperationDefinition])
+        when: 'data operation request is executed'
+            def response = objectUnderTest.executeRequest('myTopic', dataOperationRequest, NO_AUTH_HEADER)
+        and: 'a map with request id is returned'
+            assert response.keySet()[0] == 'requestId'
+        then: 'the network service is invoked'
+            1 * mockNetworkCmProxyDataService.executeDataOperationForCmHandles('myTopic', dataOperationRequest, _, NO_AUTH_HEADER)
+        where: 'the following datastores are used'
+            datastore << ['ncmp-datastore:passthrough-running', 'ncmp-datastore:passthrough-operational']
+    }
+
+    def 'Attempt to execute async data operation request with error #scenario'() {
+        given: 'a data operation definition with datastore: #datastore'
+            def dataOperationDefinition = new DataOperationDefinition(operation: 'read', datastore: datastore)
+        when: 'data operation request is executed'
+            def dataOperationRequest = new DataOperationRequest(dataOperationDefinitions: [dataOperationDefinition])
+            objectUnderTest.executeRequest('myTopic', dataOperationRequest, NO_AUTH_HEADER)
+        then: 'the correct error is thrown'
+            def thrown = thrown(InvalidDatastoreException)
+            assert thrown.message.contains(expectedErrorMessage)
+        where: 'the following datastore names are used'
+            scenario                | datastore                    || expectedErrorMessage
+            'unsupported datastore' | 'ncmp-datastore:operational' || 'not supported'
+            'invalid datastore'     | 'invalid'                    || 'invalid datastore name'
+    }
+
+    def 'Attempt to execute async data operation request with #scenario operation: #operation.'() {
+        given: 'a data operation definition with operation: #operation'
+            def dataOperationDefinition = new DataOperationDefinition(operation: operation, datastore: 'ncmp-datastore:passthrough-running')
+        when: 'data operation request is executed'
+            objectUnderTest.executeRequest('someTopic', new DataOperationRequest(dataOperationDefinitions:[dataOperationDefinition]), NO_AUTH_HEADER)
+        then: 'the expected type of exception is thrown'
+            thrown(expectedException)
+        where: 'the following operations are used'
+            scenario      | operation || expectedException
+            'invalid'     | 'invalid' || InvalidOperationException
+            'unsupported' | 'create'  || OperationNotSupportedException
+            'unsupported' | 'update'  || OperationNotSupportedException
+            'unsupported' | 'patch'   || OperationNotSupportedException
+            'unsupported' | 'delete'  || OperationNotSupportedException
+    }
+
+    def 'Attempt to execute async data operation request with too many cm handles.'() {
+        given: 'a data operation definition with too many cm handles'
+            def tooMany = objectUnderTest.MAXIMUM_CM_HANDLES_PER_OPERATION+1
+            def cmHandleIds = new String[tooMany]
+            def dataOperationDefinition = new DataOperationDefinition(operationId: 'abc', operation: 'read', datastore: 'ncmp-datastore:passthrough-running', cmHandleIds: cmHandleIds)
+        when: 'data operation request is executed'
+            objectUnderTest.executeRequest('someTopic', new DataOperationRequest(dataOperationDefinitions:[dataOperationDefinition]), NO_AUTH_HEADER)
+        then: 'a payload too large exception is thrown'
+            def exceptionThrown = thrown(PayloadTooLargeException)
+        and: 'the error message contains the offending number of cm handles'
+            assert exceptionThrown.message == "Operation 'abc' affects too many (${tooMany}) cm handles"
+    }
+
+}
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/TopicValidatorSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/TopicValidatorSpec.groovy
new file mode 100644
index 000000000..f96502984
--- /dev/null
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/utils/TopicValidatorSpec.groovy
@@ -0,0 +1,47 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 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.utils
+
+import org.onap.cps.ncmp.exceptions.InvalidTopicException
+import org.onap.cps.ncmp.utils.TopicValidator
+import spock.lang.Specification
+
+class TopicValidatorSpec extends Specification {
+
+    def 'Valid topic name validation.'() {
+        when: 'a valid topic name is validated'
+            TopicValidator.validateTopicName('my-valid-topic')
+        then: 'no exception is thrown'
+            noExceptionThrown()
+    }
+
+    def 'Validating invalid topic names.'() {
+        when: 'the invalid topic name is validated'
+            TopicValidator.validateTopicName(topicName)
+        then: 'boolean response will be returned for #scenario'
+            thrown(InvalidTopicException)
+        where: 'the following names are used'
+            scenario                  | topicName
+            'empty topic'             | ''
+            'blank topic'             | ' '
+            'invalid non empty topic' | '1_5_*_#'
+    }
+}