ms/blueprintsprocessor/modules/blueprints/blueprint-core/src/test/kotlin/org/onap/ccsdk/cds/controllerblueprints/core/service/BluePrintWorkflowServiceTest.kt

   1 /*
   2  *  Copyright © 2019 IBM.
   3  *
   4  *  Licensed under the Apache License, Version 2.0 (the "License");
   5  *  you may not use this file except in compliance with the License.
   6  *  You may obtain a copy of the License at
   7  *
   8  *      http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  *  Unless required by applicable law or agreed to in writing, software
  11  *  distributed under the License is distributed on an "AS IS" BASIS,
  12  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  *  See the License for the specific language governing permissions and
  14  *  limitations under the License.
  15  */
  16
  17 package org.onap.ccsdk.cds.controllerblueprints.core.service
  18
  19 import io.mockk.every
  20 import io.mockk.mockk
  21 import kotlinx.coroutines.CompletableDeferred
  22 import kotlinx.coroutines.async
  23 import kotlinx.coroutines.awaitAll
  24 import kotlinx.coroutines.coroutineScope
  25 import kotlinx.coroutines.runBlocking
  26 import kotlinx.coroutines.withTimeout
  27 import org.junit.Test
  28 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintException
  29 import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
  30 import org.onap.ccsdk.cds.controllerblueprints.core.data.EdgeLabel
  31 import org.onap.ccsdk.cds.controllerblueprints.core.data.Graph
  32 import org.onap.ccsdk.cds.controllerblueprints.core.logger
  33 import org.onap.ccsdk.cds.controllerblueprints.core.toGraph
  34 import kotlin.test.assertNotNull
  35
  36 class BluePrintWorkflowServiceTest {
  37     @Test
  38     fun testSimpleFlow() {
  39         runBlocking {
  40             val graph = "[START>A/SUCCESS, A>B/SUCCESS, B>C/SUCCESS, C>D/SUCCESS, D>E/SUCCESS, E>END/SUCCESS]"
  41                 .toGraph()
  42             val simpleWorkflow = TestBluePrintWorkFlowService()
  43             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "B", "C", "D", "E"), null)
  44             val input = "123456"
  45             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
  46             assertNotNull(response, "failed to get response")
  47         }
  48     }
  49
  50     @Test
  51     fun testMultipleFlows() {
  52         runBlocking {
  53             coroutineScope {
  54                 val wfs = listOf("12345", "12346").map {
  55                     async {
  56                         val graph = "[START>A/SUCCESS, A>B/SUCCESS, B>C/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
  57                             .toGraph()
  58                         val simpleWorkflow = TestBluePrintWorkFlowService()
  59                         simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "B", "C", "D"), null)
  60                         val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(it), it)
  61                         assertNotNull(response, "failed to get response")
  62                     }
  63                 }
  64                 wfs.awaitAll()
  65             }
  66         }
  67     }
  68
  69     @Test
  70     fun testMissingEdgeForBFailureState() {
  71         runBlocking {
  72             val graph = "[START>A/SUCCESS, A>B/SUCCESS, B>C/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
  73                 .toGraph()
  74             val simpleWorkflow = TestBluePrintWorkFlowService()
  75             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "C", "D", "E"), arrayListOf("B"))
  76             val input = "123456"
  77             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
  78             assertNotNull(response, "failed to get response")
  79         }
  80     }
  81
  82     @Test
  83     fun testBExceptionFlow() {
  84         runBlocking {
  85             val graph = "[START>A/SUCCESS, A>B/SUCCESS, B>C/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
  86                 .toGraph()
  87             val simpleWorkflow = TestBluePrintWorkFlowService()
  88             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "C", "D", "E"), null)
  89             val input = "123456"
  90             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
  91             assertNotNull(response, "failed to get response")
  92         }
  93     }
  94
  95     @Test
  96     fun testTimeoutExceptionFlow() {
  97         runBlocking {
  98             val graph = "[START>A/SUCCESS, A>TO/SUCCESS, TO>C/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
  99                 .toGraph()
 100             val simpleWorkflow = TestBluePrintWorkFlowService()
 101             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "TO", "C", "D", "E"), null)
 102             val input = "123456"
 103             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
 104             assertNotNull(response, "failed to get response")
 105         }
 106     }
 107
 108     @Test
 109     fun testConditionalFlow() {
 110         runBlocking {
 111             val graph = "[START>A/SUCCESS, A>B/SUCCESS, A>C/FAILURE, B>D/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
 112                 .toGraph()
 113             val simpleWorkflow = TestBluePrintWorkFlowService()
 114             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "B", "C", "D", "E"), null)
 115             val input = "123456"
 116             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
 117             assertNotNull(response, "failed to get response")
 118         }
 119     }
 120
 121     @Test
 122     fun testBothConditionalFlow() {
 123         runBlocking {
 124             // Failure Flow
 125             val failurePatGraph = "[START>A/SUCCESS, A>B/SUCCESS, A>C/FAILURE, B>D/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
 126                 .toGraph()
 127             val failurePathWorkflow = TestBluePrintWorkFlowService()
 128             failurePathWorkflow.simulatedState = prepareSimulation(
 129                 arrayListOf("B", "C", "D", "E"),
 130                 arrayListOf("A")
 131             )
 132             val failurePathWorkflowInput = "123456"
 133             val failurePathResponse = failurePathWorkflow.executeWorkflow(failurePatGraph, mockBluePrintRuntimeService(), failurePathWorkflowInput)
 134             assertNotNull(failurePathResponse, "failed to get response")
 135         }
 136     }
 137
 138     @Test
 139     fun testMultipleSkipFlow() {
 140         runBlocking {
 141             val graph = "[START>A/SUCCESS, A>B/SUCCESS, A>C/FAILURE, C>D/SUCCESS, D>E/SUCCESS, B>E/SUCCESS, E>END/SUCCESS]"
 142                 .toGraph()
 143             val simpleWorkflow = TestBluePrintWorkFlowService()
 144             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "B", "C", "D", "E"), null)
 145             val input = "123456"
 146             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
 147             assertNotNull(response, "failed to get response")
 148         }
 149     }
 150
 151     @Test
 152     fun testParallelFlow() {
 153         runBlocking {
 154             val graph = "[START>A/SUCCESS, A>B/SUCCESS, A>C/SUCCESS, B>D/SUCCESS, C>D/SUCCESS, D>END/SUCCESS]"
 155                 .toGraph()
 156             val simpleWorkflow = TestBluePrintWorkFlowService()
 157             simpleWorkflow.simulatedState = prepareSimulation(arrayListOf("A", "B", "C", "D"), null)
 158             val input = "123456"
 159             val response = simpleWorkflow.executeWorkflow(graph, mockBluePrintRuntimeService(), input)
 160             assertNotNull(response, "failed to get response")
 161         }
 162     }
 163
 164     private fun mockBluePrintRuntimeService(): BluePrintRuntimeService<*> {
 165         return mockBluePrintRuntimeService("123456")
 166     }
 167
 168     private fun mockBluePrintRuntimeService(id: String): BluePrintRuntimeService<*> {
 169         val bluePrintRuntimeService = mockk<BluePrintRuntimeService<*>>()
 170         every { bluePrintRuntimeService.id() } returns id
 171         return bluePrintRuntimeService
 172     }
 173
 174     private fun prepareSimulation(successes: List<String>?, failures: List<String>?): MutableMap<String, EdgeLabel> {
 175         val simulatedState: MutableMap<String, EdgeLabel> = hashMapOf()
 176         successes?.forEach {
 177             simulatedState[it] = EdgeLabel.SUCCESS
 178         }
 179         failures?.forEach {
 180             simulatedState[it] = EdgeLabel.FAILURE
 181         }
 182         return simulatedState
 183     }
 184 }
 185
 186 class TestBluePrintWorkFlowService :
 187     AbstractBluePrintWorkFlowService<String, String>() {
 188
 189     val log = logger(TestBluePrintWorkFlowService::class)
 190
 191     lateinit var simulatedState: MutableMap<String, EdgeLabel>
 192
 193     override suspend fun initializeWorkflow(input: String): EdgeLabel {
 194         return EdgeLabel.SUCCESS
 195     }
 196
 197     override suspend fun executeWorkflow(graph: Graph, bluePrintRuntimeService: BluePrintRuntimeService<*>, input: String): String {
 198         log.info("Executing Graph : $graph")
 199         this.graph = graph
 200         this.workflowId = bluePrintRuntimeService.id()
 201         val output = CompletableDeferred<String>()
 202         val startMessage = WorkflowExecuteMessage(input, output)
 203         val workflowActor = workflowActor()
 204         if (!workflowActor.isClosedForSend) {
 205             workflowActor().send(startMessage)
 206         } else {
 207             throw BluePrintProcessorException("workflow actor is closed for send $workflowActor")
 208         }
 209         return startMessage.output.await()
 210     }
 211
 212     override suspend fun prepareNodeExecutionMessage(node: Graph.Node):
 213             NodeExecuteMessage<String, String> {
 214         return NodeExecuteMessage(node, "$node Input", "")
 215     }
 216
 217     override suspend fun executeNode(
 218         node: Graph.Node,
 219         nodeInput: String,
 220         nodeOutput: String
 221     ): EdgeLabel {
 222         //        val random = (1..10).random() * 100
 223         //        log.info("workflow($workflowId) node(${node.id}) will reply in $random ms")
 224         //        kotlinx.coroutines.delay(random.toLong())
 225         //        //Simulation for timeout
 226         if (node.id == "TO") {
 227             withTimeout(1) {
 228                 kotlinx.coroutines.delay(2)
 229             }
 230         }
 231         return simulatedState[node.id] ?: throw BluePrintException("failed to get status for the node($node)")
 232     }
 233
 234     override suspend fun prepareNodeSkipMessage(node: Graph.Node): NodeSkipMessage<String, String> {
 235         val nodeOutput = ""
 236         return NodeSkipMessage(node, "$node Skip Input", nodeOutput)
 237     }
 238
 239     override suspend fun skipNode(
 240         node: Graph.Node,
 241         nodeInput: String,
 242         nodeOutput: String
 243     ): EdgeLabel {
 244         return simulatedState[node.id] ?: throw BluePrintException("failed to get status for the node($node)")
 245     }
 246
 247     override suspend fun cancelNode(
 248         node: Graph.Node,
 249         nodeInput: String,
 250         nodeOutput: String
 251     ): EdgeLabel {
 252         TODO("not implemented") // To change body of created functions use File | Settings | File Templates.
 253     }
 254
 255     override suspend fun restartNode(
 256         node: Graph.Node,
 257         nodeInput: String,
 258         nodeOutput: String
 259     ): EdgeLabel {
 260         TODO("not implemented") // To change body of created functions use File | Settings | File Templates.
 261     }
 262
 263     override suspend fun prepareWorkflowOutput(): String {
 264         if (exceptions.isNotEmpty()) {
 265             exceptions.forEach {
 266                 log.error("workflow($workflowId) exceptions :", it)
 267             }
 268         }
 269         return "Final Response"
 270     }
 271 }