otf-camunda/src/main/java/org/oran/otf/camunda/delegate/otf/common/CallTestHeadDelegate.java

   1 /*  Copyright (c) 2019 AT&T Intellectual Property.                             #\r
   2 #                                                                              #\r
   3 #   Licensed under the Apache License, Version 2.0 (the "License");            #\r
   4 #   you may not use this file except in compliance with the License.           #\r
   5 #   You may obtain a copy of the License at                                    #\r
   6 #                                                                              #\r
   7 #       http://www.apache.org/licenses/LICENSE-2.0                             #\r
   8 #                                                                              #\r
   9 #   Unless required by applicable law or agreed to in writing, software        #\r
  10 #   distributed under the License is distributed on an "AS IS" BASIS,          #\r
  11 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   #\r
  12 #   See the License for the specific language governing permissions and        #\r
  13 #   limitations under the License.                                             #\r
  14 ##############################################################################*/\r
  15 \r
  16 \r
  17 package org.oran.otf.camunda.delegate.otf.common;\r
  18 \r
  19 import org.oran.otf.camunda.delegate.otf.common.runnable.TestHeadCallable;\r
  20 import org.oran.otf.camunda.exception.TestExecutionException;\r
  21 import org.oran.otf.camunda.workflow.utility.WorkflowTask;\r
  22 import org.oran.otf.camunda.workflow.utility.WorkflowUtility;\r
  23 import org.oran.otf.common.model.*;\r
  24 import org.oran.otf.common.model.local.BpmnInstance;\r
  25 import org.oran.otf.common.model.local.TestHeadNode;\r
  26 import org.oran.otf.common.model.local.TestHeadResult;\r
  27 import org.oran.otf.common.repository.*;\r
  28 import org.oran.otf.common.utility.Utility;\r
  29 import org.oran.otf.common.utility.database.Generic;\r
  30 import org.oran.otf.common.utility.permissions.PermissionChecker;\r
  31 import org.oran.otf.common.utility.permissions.UserPermission;\r
  32 import com.mongodb.client.result.UpdateResult;\r
  33 import java.util.ArrayList;\r
  34 import java.util.Collections;\r
  35 import java.util.HashMap;\r
  36 import java.util.List;\r
  37 import java.util.Map;\r
  38 import java.util.concurrent.ExecutorService;\r
  39 import java.util.concurrent.TimeUnit;\r
  40 \r
  41 import org.camunda.bpm.engine.delegate.DelegateExecution;\r
  42 import org.camunda.bpm.engine.delegate.JavaDelegate;\r
  43 import org.oran.otf.common.model.*;\r
  44 import org.oran.otf.common.repository.*;\r
  45 import org.slf4j.Logger;\r
  46 import org.slf4j.LoggerFactory;\r
  47 import org.springframework.beans.factory.annotation.Autowired;\r
  48 import org.springframework.data.mongodb.core.MongoTemplate;\r
  49 import org.springframework.data.mongodb.core.query.Criteria;\r
  50 import org.springframework.data.mongodb.core.query.Query;\r
  51 import org.springframework.data.mongodb.core.query.Update;\r
  52 import org.springframework.stereotype.Component;\r
  53 \r
  54 @Component\r
  55 public class CallTestHeadDelegate implements JavaDelegate {\r
  56   private static final Logger logger = LoggerFactory.getLogger(CallTestHeadDelegate.class);\r
  57 \r
  58   @Autowired\r
  59   private UserRepository userRepository;\r
  60   @Autowired\r
  61   private GroupRepository groupRepository;\r
  62   @Autowired\r
  63   private WorkflowUtility utility;\r
  64   @Autowired\r
  65   private TestDefinitionRepository testDefinitionRepository;\r
  66   @Autowired\r
  67   private TestHeadRepository testHeadRepository;\r
  68   @Autowired\r
  69   private TestInstanceRepository testInstanceRepository;\r
  70   @Autowired\r
  71   private MongoTemplate mongoOperation;\r
  72 \r
  73   // Used to retrieve the results from test head runnables.\r
  74   List<TestHeadResult> testHeadResults = Collections.synchronizedList(new ArrayList<>());\r
  75 \r
  76   @Override\r
  77   public void execute(DelegateExecution execution) throws Exception {\r
  78     callTestHead(\r
  79         execution.getCurrentActivityId(),\r
  80         execution.getProcessDefinitionId(),\r
  81         execution.getProcessInstanceId(),\r
  82         execution.getProcessBusinessKey(),\r
  83         execution.getVariables());\r
  84   }\r
  85 \r
  86   public void callTestHead(\r
  87       String currentActivityId,\r
  88       String processDefinitionId,\r
  89       String processInstanceId,\r
  90       String processBusinessKey,\r
  91       Map<String, Object> variables)\r
  92       throws Exception {\r
  93     final String logPrefix = Utility.getLoggerPrefix();\r
  94     logger.info(logPrefix + "::execute()");\r
  95 \r
  96     // Get vthInput from the Camunda execution variable map.\r
  97     List<Map<String, Object>> activityParameters = utility.getVthInput(variables, currentActivityId, logPrefix);\r
  98 \r
  99     // Get the current test execution object.\r
 100     TestExecution testExecution = utility.getTestExecution(variables, logPrefix);\r
 101 \r
 102     // Lookup the test head before making computations in the loop, and before calling the runnable.\r
 103     // If the lookup is made inside the runnable, concurrent test head calls would bombard the db.\r
 104     TestHead testHead = getTestHead(testExecution, currentActivityId, processDefinitionId);\r
 105 \r
 106     WorkflowTask workflowTask = new WorkflowTask(processInstanceId, activityParameters.size(), false);\r
 107     ExecutorService pool = workflowTask.getPool();\r
 108 \r
 109     // Try to cast each parameter to a Map, and create runnable tasks.\r
 110     for (int i = 0; i < activityParameters.size(); i++) {\r
 111       Object oTestHeadParameter = activityParameters.get(i);\r
 112       Map<?, ?> mTestHeadParameter;\r
 113       try {\r
 114         mTestHeadParameter = Utility.toMap(oTestHeadParameter);\r
 115         verifyOtfTestHead(mTestHeadParameter, testHead, testExecution, currentActivityId);\r
 116       } catch (Exception e) {\r
 117         // TODO: Make a design decision to either stop the execution, or attempt to convert the\r
 118         // other parameters.\r
 119         logger.error(\r
 120             String.format(\r
 121                 "Unable to convert test head parameter at vthInput[%s][%d] to a Map.",\r
 122                 currentActivityId, i));\r
 123         continue;\r
 124       }\r
 125 \r
 126       // Get all the arguments for the runnable.\r
 127       Object oHeaders = mTestHeadParameter.get("headers"); // optional\r
 128       Object oMethod = mTestHeadParameter.get("method"); // required\r
 129       Object oPayload = mTestHeadParameter.get("payload"); // optional\r
 130       Object oTimeoutInMillis = mTestHeadParameter.get("timeoutInMillis"); // optional\r
 131 \r
 132       // Target typed parameters. Convert all objects to their expected types. Throw exceptions for\r
 133       // required parameters, or for parameters that are provided but not of the expected type.\r
 134       Map<String, String> headers = new HashMap<>();\r
 135       String method = "";\r
 136       Map<String, Object> payload = new HashMap<>();\r
 137       int timeoutInMillis = 0;\r
 138 \r
 139       if (oHeaders != null) {\r
 140         try {\r
 141           headers = (Map<String, String>) Utility.toMap(oHeaders);\r
 142         } catch (Exception e) {\r
 143           logger.error(\r
 144               String.format(\r
 145                   "Unable to convert test head parameter at vthInput[%s][%d][headers] to a Map.",\r
 146                   currentActivityId, i));\r
 147         }\r
 148       }\r
 149 \r
 150       if (oMethod == null) {\r
 151         throw new TestExecutionException(\r
 152             String.format(\r
 153                 "vthInput[%s][%d][method] is a required parameter.", currentActivityId, i));\r
 154       } else {\r
 155         try {\r
 156           method = (String) oMethod;\r
 157         } catch (ClassCastException cce) {\r
 158           throw new TestExecutionException(\r
 159               String.format(\r
 160                   "Unable to read vthInput[%s][%d][method] as primitive type String.",\r
 161                   processInstanceId, i));\r
 162         }\r
 163       }\r
 164 \r
 165       if (oPayload != null) {\r
 166         try {\r
 167           payload = (Map<String, Object>) Utility.toMap(oPayload);\r
 168         } catch (Exception e) {\r
 169           logger.error(\r
 170               String.format(\r
 171                   "Unable to convert test head parameter at vthInput[%s][%d][payload] to a Map.",\r
 172                   currentActivityId, i));\r
 173         }\r
 174       }\r
 175 \r
 176       if (oTimeoutInMillis != null) {\r
 177         try {\r
 178           timeoutInMillis = (int) oTimeoutInMillis;\r
 179         } catch (ClassCastException cce) {\r
 180           throw new TestExecutionException(\r
 181               String.format(\r
 182                   "Unable to read vthInput[%s][%d][timeoutInMillis] as primitive type int.",\r
 183                   currentActivityId, i));\r
 184         }\r
 185       }\r
 186 \r
 187 //      logger.info("{}(BEFORE) PRINTING THREAD INFORMATION", logPrefix);\r
 188 //      WorkflowTask.printThreadInformation();\r
 189 //      logger.info("{}(BEFORE) PRINTING WORKFLOW TASKS", logPrefix);\r
 190 //      WorkflowTask.printWorkflowTaskResources();\r
 191       TestHeadCallable callable =\r
 192           new TestHeadCallable(\r
 193               timeoutInMillis,\r
 194               method,\r
 195               headers,\r
 196               payload,\r
 197               testHead,\r
 198               currentActivityId,\r
 199               testExecution,\r
 200               mongoOperation);\r
 201 \r
 202       // Submit the test head call to the executor service.\r
 203       workflowTask.getFutures().add(pool.submit(callable));\r
 204     }\r
 205 \r
 206     // Prevent new tasks from being submitted, and allow running tasks to finish.\r
 207     pool.shutdown();\r
 208 \r
 209     int numResults = 0;\r
 210     while (!pool.isTerminated()) {\r
 211       try {\r
 212         pool.awaitTermination(1, TimeUnit.SECONDS);\r
 213       } catch (InterruptedException e) {\r
 214         workflowTask.shutdown(true);\r
 215         throw e;\r
 216       }\r
 217     }\r
 218 \r
 219     workflowTask.shutdown(false);\r
 220 \r
 221 //    logger.info("{}(AFTER) PRINTING THREAD INFORMATION", logPrefix);\r
 222 //    WorkflowTask.printThreadInformation();\r
 223 //    logger.info("{}(AFTER) PRINTING WORKFLOW TASKS", logPrefix);\r
 224 //    WorkflowTask.printWorkflowTaskResources();\r
 225   }\r
 226 \r
 227   private void saveTestHeadResults(String businessKey) {\r
 228     Query query = new Query();\r
 229     query.addCriteria(Criteria.where("businessKey").is(businessKey));\r
 230     Update update = new Update();\r
 231     update.set("testHeadResults", testHeadResults);\r
 232     UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
 233     // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
 234     if (result.getMatchedCount() == 0) {\r
 235       throw new TestExecutionException(\r
 236           String.format(\r
 237               "Unable to log the test result because a testExecution associated with businessKey, %s, was not found.",\r
 238               businessKey));\r
 239     } else if (result.getModifiedCount() == 0) {\r
 240       throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
 241     }\r
 242   }\r
 243 \r
 244   private TestHead getTestHead(\r
 245       TestExecution testExecution, String currentActivityId, String processDefinitionId) {\r
 246     List<BpmnInstance> bpmnInstances = testExecution.getHistoricTestDefinition().getBpmnInstances();\r
 247     BpmnInstance bpmnInstance =\r
 248         bpmnInstances.stream()\r
 249             .filter(\r
 250                 _bpmnInstance ->\r
 251                     _bpmnInstance.getProcessDefinitionId().equalsIgnoreCase(processDefinitionId))\r
 252             .findFirst()\r
 253             .orElse(null);\r
 254 \r
 255     if (bpmnInstance == null) {\r
 256       throw new TestExecutionException(\r
 257           String.format(\r
 258               "Error looking BpmnInstance with processDefinitionId %s.", processDefinitionId));\r
 259     }\r
 260 \r
 261     List<TestHeadNode> testHeads = bpmnInstance.getTestHeads();\r
 262     TestHeadNode testHeadNode =\r
 263         testHeads.stream()\r
 264             .filter(testHead -> testHead.getBpmnVthTaskId().equals(currentActivityId))\r
 265             .findAny()\r
 266             .orElse(null);\r
 267 \r
 268     if (testHeadNode == null) {\r
 269       throw new TestExecutionException(\r
 270           String.format(\r
 271               "No test head associated with the currentActivityId %s.", currentActivityId));\r
 272     }\r
 273 \r
 274     TestHead testHead = Generic.findByIdGeneric(testHeadRepository, testHeadNode.getTestHeadId());\r
 275     if (testHead == null) {\r
 276       throw new TestExecutionException(\r
 277           String.format(\r
 278               "The test head with id, %s, was not found in the database.",\r
 279               testHeadNode.getTestHeadId()));\r
 280     }\r
 281     User testExecUser = userRepository.findById(testExecution.getExecutorId().toString()).orElse(null);\r
 282     Group testheadGroup =  groupRepository.findById(testHead.getGroupId().toString()).orElse(null);\r
 283     if(testExecUser == null){\r
 284       throw new TestExecutionException(\r
 285               String.format("Can not find user, user id: %s",testExecution.getExecutorId().toString()));\r
 286     }\r
 287     if(testheadGroup == null){\r
 288       throw new TestExecutionException(\r
 289               String.format("Can not find test head group, group id: %s",testHead.getGroupId().toString())\r
 290       );\r
 291     }\r
 292 \r
 293     if( (testHead.isPublic() != null && !testHead.isPublic()) &&\r
 294             !PermissionChecker.hasPermissionTo(testExecUser,testheadGroup,UserPermission.Permission.EXECUTE,groupRepository)){\r
 295       throw new TestExecutionException(\r
 296               String.format(\r
 297                       "User(%s) does not have permission to in testHead Group(%s)",\r
 298                       testExecUser.get_id().toString(),testheadGroup.get_id().toString()\r
 299               ));\r
 300     }\r
 301     return testHead;\r
 302   }\r
 303 \r
 304   private void verifyOtfTestHead(Map activityParams, TestHead testHead, TestExecution execution, String currentActivityId){\r
 305     String testHeadName = testHead.getTestHeadName().toLowerCase();\r
 306     switch(testHeadName) {\r
 307       case "robot":\r
 308         try {\r
 309           TestInstance testInstance = Generic.findByIdGeneric(testInstanceRepository, execution.getHistoricTestInstance().get_id());\r
 310           Map<String, Object> internalTestDataByActivity = (Map<String, Object>) testInstance.getInternalTestData().get(currentActivityId);\r
 311           String robotFileId = (String) internalTestDataByActivity.get("robotFileId");\r
 312           Map<String, Object> testData = new HashMap<>();\r
 313           Map<String, Object> vthInput = new HashMap<>();\r
 314           testData.put("robotFileId", robotFileId);\r
 315           vthInput.put("testData", testData);\r
 316           Map<String, Object> payload = (Map<String, Object>) activityParams.get("payload");\r
 317           payload.put("vthInput", vthInput);\r
 318         }\r
 319         catch (Exception e){\r
 320           throw new TestExecutionException(\r
 321                   String.format(\r
 322                           "Robot test head needs a robot file id: %s.", e.getMessage()));\r
 323         }\r
 324         break;\r
 325       default:\r
 326         break;\r
 327     }\r
 328   }\r
 329 }\r