Add bi-directional python executor tests

Issue-ID: CCSDK-1747
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: I2e2b974d35d6878418eafe8ece9fcb1d69622a61

4 files changed, 395 insertions, 0 deletions

diff --git a/ms/blueprintsprocessor/modules/services/execution-service/pom.xml b/ms/blueprintsprocessor/modules/services/execution-service/pom.xml
index 8bee7c91c..f3044c8bb 100644
--- a/ms/blueprintsprocessor/modules/services/execution-service/pom.xml
+++ b/ms/blueprintsprocessor/modules/services/execution-service/pom.xml
@@ -64,5 +64,10 @@
             <groupId>org.onap.ccsdk.sli.core</groupId>
             <artifactId>sli-provider</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>io.grpc</groupId>
+            <artifactId>grpc-testing</artifactId>
+        </dependency>
     </dependencies>
 </project>
diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionService.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionService.kt
new file mode 100644
index 000000000..07848164e
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionService.kt
@@ -0,0 +1,174 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * 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.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.services.execution
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.github.marcoferrer.krotoplus.coroutines.client.ClientBidiCallChannel
+import com.github.marcoferrer.krotoplus.coroutines.client.clientCallBidiStreaming
+import io.grpc.ManagedChannel
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.launch
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.GrpcClientProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.BluePrintGrpcClientService
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.BluePrintGrpcLibPropertyService
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintException
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.stereotype.Service
+
+interface StreamingRemoteExecutionService<ReqT, ResT> {
+
+    suspend fun openSubscription(selector: Any, requestId: String): Flow<ResT>
+
+    suspend fun send(input: ReqT)
+
+    suspend fun cancelSubscription(requestId: String)
+
+    suspend fun closeChannel(selector: Any)
+}
+
+@Service
+@ConditionalOnProperty(prefix = "blueprintsprocessor.streamingRemoteExecution", name = ["enabled"],
+        havingValue = "true", matchIfMissing = false)
+class StreamingRemoteExecutionServiceImpl(private val bluePrintGrpcLibPropertyService: BluePrintGrpcLibPropertyService)
+    : StreamingRemoteExecutionService<ExecutionServiceInput, ExecutionServiceOutput> {
+
+    private val log = logger(StreamingRemoteExecutionServiceImpl::class)
+
+    private val grpcChannels: MutableMap<String, ManagedChannel> = hashMapOf()
+
+    private val commChannels: MutableMap<String,
+            ClientBidiCallChannel<ExecutionServiceInput, ExecutionServiceOutput>> = hashMapOf()
+
+
+    /**
+     * Open new channel to send and receive for grpc properties [selector] for [requestId],
+     * Create the only one GRPC channel per host port and reuse for further communication.
+     * Create request communication channel to send and receive requests and responses.
+     * We can send multiple request with same requestId with unique subRequestId.
+     * Consume the flow for responses,
+     * Client should cancel the subscription for the request Id once no longer response is needed.
+     * */
+    @FlowPreview
+    override suspend fun openSubscription(selector: Any, requestId: String): Flow<ExecutionServiceOutput> {
+
+        if (!commChannels.containsKey(requestId)) {
+            /** Get GRPC Channel*/
+            val grpcChannel = grpcChannel(selector)
+
+            /** Get Send and Receive Channel for bidirectional process method*/
+            val channels = clientCallBidiStreaming(BluePrintProcessingServiceGrpc.getProcessMethod(), grpcChannel)
+            commChannels[requestId] = channels
+        }
+
+        val commChannel = commChannels[requestId]
+                ?: throw BluePrintException("failed to create response subscription for request($requestId) channel")
+
+        log.info("created subscription for request($requestId)")
+
+        return commChannel.responseChannel.consumeAsFlow()
+    }
+
+    /**
+     * Send the [input]request, by reusing same GRPC channel and Communication channel
+     * for the request Id.
+     */
+    override suspend fun send(input: ExecutionServiceInput) {
+        val requestId = input.commonHeader.requestId
+        val subRequestId = input.commonHeader.subRequestId
+        val sendChannel = commChannels[requestId]?.requestChannel
+                ?: throw BluePrintException("failed to get request($requestId) send channel")
+        coroutineScope {
+            launch {
+                sendChannel.send(input)
+                log.trace("Message sent for request($requestId) : subRequest($subRequestId)")
+            }
+        }
+    }
+
+    /** Cancel the Subscription for the [requestId], This closes communication channel **/
+    @ExperimentalCoroutinesApi
+    override suspend fun cancelSubscription(requestId: String) {
+        commChannels[requestId]?.let {
+            if (!it.requestChannel.isClosedForSend)
+                it.requestChannel.close()
+            /** If receive channel has to close immediately, once the subscription has cancelled, then enable this */
+            //it.responseChannel.cancel(CancellationException("subscription cancelled"))
+            commChannels.remove(requestId)
+            log.info("closed subscription for request($requestId)")
+        }
+    }
+
+    /** Close the GRPC channel for the host port poperties [selector]*/
+    override suspend fun closeChannel(selector: Any) {
+        val grpcProperties = grpcProperties(selector)
+        val selectorName = "${grpcProperties.host}:${grpcProperties.port}"
+        if (grpcChannels.containsKey(selectorName)) {
+            grpcChannels[selectorName]!!.shutdownNow()
+            grpcChannels.remove(selectorName)
+            log.info("grpc channel($selectorName) shutdown completed")
+        }
+    }
+
+    /** Check GRPC channel has been cached and not shutdown, If not re create channel and chache it. */
+    private suspend fun grpcChannel(selector: Any): ManagedChannel {
+        val grpcProperties = grpcProperties(selector)
+        val selectorName = "${grpcProperties.host}:${grpcProperties.port}"
+        val isGrpcChannelCached = grpcChannels.containsKey(selectorName)
+        val grpcChannel = if (isGrpcChannelCached) {
+            if (grpcChannels[selectorName]!!.isShutdown) {
+                createGrpcChannel(grpcProperties)
+            } else {
+                grpcChannels[selectorName]!!
+            }
+        } else {
+            createGrpcChannel(grpcProperties)
+        }
+        grpcChannels[selectorName] = grpcChannel
+        return grpcChannel
+    }
+
+    suspend fun createGrpcChannel(grpcProperties: GrpcClientProperties): ManagedChannel {
+        val grpcClientService: BluePrintGrpcClientService = bluePrintGrpcLibPropertyService
+                .blueprintGrpcClientService(grpcProperties)
+        return grpcClientService.channel()
+    }
+
+    private fun grpcProperties(selector: Any): GrpcClientProperties {
+        return when (selector) {
+            is String -> {
+                bluePrintGrpcLibPropertyService.grpcClientProperties(selector.toString())
+            }
+            is JsonNode -> {
+                bluePrintGrpcLibPropertyService.grpcClientProperties(selector)
+            }
+            is GrpcClientProperties -> {
+                selector
+            }
+            else -> {
+                throw BluePrintException("couldn't process selector($selector)")
+            }
+        }
+    }
+}
diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/MockBluePrintProcessingServer.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/MockBluePrintProcessingServer.kt
new file mode 100644
index 000000000..e291aa78e
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/MockBluePrintProcessingServer.kt
@@ -0,0 +1,97 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * 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.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.services.execution.scripts
+
+import io.grpc.ServerBuilder
+import io.grpc.stub.StreamObserver
+import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
+import org.onap.ccsdk.cds.controllerblueprints.common.api.Status
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput
+
+private val log = logger(MockBluePrintProcessingServer::class)
+
+
+class MockBluePrintProcessingServer : BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase() {
+
+    override fun process(responseObserver: StreamObserver<ExecutionServiceOutput>): StreamObserver<ExecutionServiceInput> {
+
+        return object : StreamObserver<ExecutionServiceInput> {
+            override fun onNext(executionServiceInput: ExecutionServiceInput) {
+                log.info("Received requestId(${executionServiceInput.commonHeader.requestId})  " +
+                        "subRequestId(${executionServiceInput.commonHeader.subRequestId})")
+                responseObserver.onNext(buildNotification(executionServiceInput))
+                responseObserver.onNext(buildResponse(executionServiceInput))
+                responseObserver.onCompleted()
+            }
+
+            override fun onError(error: Throwable) {
+                log.debug("Fail to process message", error)
+                responseObserver.onError(io.grpc.Status.INTERNAL
+                        .withDescription(error.message)
+                        .asException())
+            }
+
+            override fun onCompleted() {
+                log.info("Completed")
+            }
+        }
+    }
+
+
+    private fun buildNotification(input: ExecutionServiceInput): ExecutionServiceOutput {
+        val status = Status.newBuilder()
+                .setEventType(EventType.EVENT_COMPONENT_NOTIFICATION)
+                .build()
+        return ExecutionServiceOutput.newBuilder()
+                .setCommonHeader(input.commonHeader)
+                .setActionIdentifiers(input.actionIdentifiers)
+                .setStatus(status)
+                .build()
+    }
+
+    private fun buildResponse(input: ExecutionServiceInput): ExecutionServiceOutput {
+
+        val status = Status.newBuilder().setCode(200)
+                .setEventType(EventType.EVENT_COMPONENT_EXECUTED)
+                .build()
+        return ExecutionServiceOutput.newBuilder()
+                .setCommonHeader(input.commonHeader)
+                .setActionIdentifiers(input.actionIdentifiers)
+                .setStatus(status)
+                .build()
+
+    }
+}
+
+/** For Integration testing stat this server */
+fun main() {
+    try {
+        val server = ServerBuilder
+                .forPort(50052)
+                .addService(MockBluePrintProcessingServer())
+                .build()
+        server.start()
+        log.info("GRPC Serve started(${server.isShutdown}) on port(${server.port})...")
+        server.awaitTermination()
+    } catch (e: Exception) {
+        e.printStackTrace()
+    }
+
+}
\ No newline at end of file
diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt
new file mode 100644
index 000000000..c9ff23573
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt
@@ -0,0 +1,119 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * 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.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.services.execution
+
+import io.grpc.inprocess.InProcessChannelBuilder
+import io.grpc.inprocess.InProcessServerBuilder
+import io.grpc.testing.GrpcCleanupRule
+import io.mockk.coEvery
+import io.mockk.mockk
+import io.mockk.spyk
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.collect
+import org.junit.Rule
+import org.junit.Test
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.GRPCLibConstants
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.TokenAuthGrpcClientProperties
+import org.onap.ccsdk.cds.blueprintsprocessor.grpc.service.BluePrintGrpcLibPropertyService
+import org.onap.ccsdk.cds.blueprintsprocessor.services.execution.scripts.MockBluePrintProcessingServer
+import org.onap.ccsdk.cds.controllerblueprints.common.api.ActionIdentifiers
+import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader
+import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
+import java.util.*
+
+
+class StreamingRemoteExecutionServiceTest {
+
+    val log = logger(StreamingRemoteExecutionServiceTest::class)
+
+    @get:Rule
+    val grpcCleanup = GrpcCleanupRule()
+    private val serverName = InProcessServerBuilder.generateName()
+    private val serverBuilder = InProcessServerBuilder.forName(serverName).directExecutor()
+    private val channelBuilder = InProcessChannelBuilder.forName(serverName).directExecutor()
+
+    private val tokenAuthGrpcClientProperties = TokenAuthGrpcClientProperties().apply {
+        host = "127.0.0.1"
+        port = 50052
+        type = GRPCLibConstants.TYPE_TOKEN_AUTH
+        token = "Basic Y2NzZGthcHBzOmNjc2RrYXBwcw=="
+    }
+
+    @Test
+    @ExperimentalCoroutinesApi
+    @FlowPreview
+    fun testStreamingChannel() {
+        grpcCleanup.register(serverBuilder.addService(MockBluePrintProcessingServer()).build().start())
+        val channel = grpcCleanup.register(channelBuilder.maxInboundMessageSize(1024).build())
+
+        runBlocking {
+            val bluePrintGrpcLibPropertyService = BluePrintGrpcLibPropertyService(mockk())
+
+            val streamingRemoteExecutionService = StreamingRemoteExecutionServiceImpl(bluePrintGrpcLibPropertyService)
+
+            val spyStreamingRemoteExecutionService = spyk(streamingRemoteExecutionService)
+            /** To test with real server, uncomment below line */
+            coEvery { spyStreamingRemoteExecutionService.createGrpcChannel(any()) } returns channel
+
+            val responseFlowsDeferred = arrayListOf<Deferred<*>>()
+
+            repeat(1) { count ->
+                val requestId = "12345-$count"
+                val responseFlow = spyStreamingRemoteExecutionService.openSubscription(tokenAuthGrpcClientProperties, requestId)
+
+                val deferred = async {
+                    responseFlow.collect {
+                        log.info("Received $count-response (${it.commonHeader.subRequestId}) : ${it.status.eventType}")
+                        if (it.status.eventType == EventType.EVENT_COMPONENT_EXECUTED) {
+                            spyStreamingRemoteExecutionService.cancelSubscription(it.commonHeader.requestId)
+                        }
+                    }
+                }
+                responseFlowsDeferred.add(deferred)
+                /** Sending Multiple messages with same requestId  and different subRequestId */
+                spyStreamingRemoteExecutionService.send(getRequest(requestId))
+            }
+            responseFlowsDeferred.awaitAll()
+            streamingRemoteExecutionService.closeChannel(tokenAuthGrpcClientProperties)
+        }
+
+    }
+
+    private fun getRequest(requestId: String): ExecutionServiceInput {
+        val commonHeader = CommonHeader.newBuilder()
+                .setTimestamp("2012-04-23T18:25:43.511Z")
+                .setOriginatorId("System")
+                .setRequestId(requestId)
+                .setSubRequestId("$requestId-" + UUID.randomUUID().toString()).build()
+
+
+        val actionIdentifier = ActionIdentifiers.newBuilder()
+                .setActionName("SampleScript")
+                .setBlueprintName("sample-cba")
+                .setBlueprintVersion("1.0.0")
+                .build()
+
+        return ExecutionServiceInput.newBuilder()
+                .setCommonHeader(commonHeader)
+                .setActionIdentifiers(actionIdentifier)
+                //.setPayload(payloadBuilder.build())
+                .build()
+
+    }
+}
\ No newline at end of file