--- /dev/null
+/* Copyright (c) 2019 AT&T Intellectual Property. #\r
+# #\r
+# Licensed under the Apache License, Version 2.0 (the "License"); #\r
+# you may not use this file except in compliance with the License. #\r
+# You may obtain a copy of the License at #\r
+# #\r
+# http://www.apache.org/licenses/LICENSE-2.0 #\r
+# #\r
+# Unless required by applicable law or agreed to in writing, software #\r
+# distributed under the License is distributed on an "AS IS" BASIS, #\r
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #\r
+# See the License for the specific language governing permissions and #\r
+# limitations under the License. #\r
+##############################################################################*/\r
+\r
+\r
+package org.oran.otf.camunda.delegate.otf.common;\r
+\r
+import org.oran.otf.camunda.delegate.otf.common.runnable.TestHeadCallable;\r
+import org.oran.otf.camunda.exception.TestExecutionException;\r
+import org.oran.otf.camunda.workflow.utility.WorkflowTask;\r
+import org.oran.otf.camunda.workflow.utility.WorkflowUtility;\r
+import org.oran.otf.common.model.*;\r
+import org.oran.otf.common.model.local.BpmnInstance;\r
+import org.oran.otf.common.model.local.TestHeadNode;\r
+import org.oran.otf.common.model.local.TestHeadResult;\r
+import org.oran.otf.common.repository.*;\r
+import org.oran.otf.common.utility.Utility;\r
+import org.oran.otf.common.utility.database.Generic;\r
+import org.oran.otf.common.utility.permissions.PermissionChecker;\r
+import org.oran.otf.common.utility.permissions.UserPermission;\r
+import com.mongodb.client.result.UpdateResult;\r
+import java.util.ArrayList;\r
+import java.util.Collections;\r
+import java.util.HashMap;\r
+import java.util.List;\r
+import java.util.Map;\r
+import java.util.concurrent.ExecutorService;\r
+import java.util.concurrent.TimeUnit;\r
+\r
+import org.camunda.bpm.engine.delegate.DelegateExecution;\r
+import org.camunda.bpm.engine.delegate.JavaDelegate;\r
+import org.oran.otf.common.model.*;\r
+import org.oran.otf.common.repository.*;\r
+import org.slf4j.Logger;\r
+import org.slf4j.LoggerFactory;\r
+import org.springframework.beans.factory.annotation.Autowired;\r
+import org.springframework.data.mongodb.core.MongoTemplate;\r
+import org.springframework.data.mongodb.core.query.Criteria;\r
+import org.springframework.data.mongodb.core.query.Query;\r
+import org.springframework.data.mongodb.core.query.Update;\r
+import org.springframework.stereotype.Component;\r
+\r
+@Component\r
+public class CallTestHeadDelegate implements JavaDelegate {\r
+ private static final Logger logger = LoggerFactory.getLogger(CallTestHeadDelegate.class);\r
+\r
+ @Autowired\r
+ private UserRepository userRepository;\r
+ @Autowired\r
+ private GroupRepository groupRepository;\r
+ @Autowired\r
+ private WorkflowUtility utility;\r
+ @Autowired\r
+ private TestDefinitionRepository testDefinitionRepository;\r
+ @Autowired\r
+ private TestHeadRepository testHeadRepository;\r
+ @Autowired\r
+ private TestInstanceRepository testInstanceRepository;\r
+ @Autowired\r
+ private MongoTemplate mongoOperation;\r
+\r
+ // Used to retrieve the results from test head runnables.\r
+ List<TestHeadResult> testHeadResults = Collections.synchronizedList(new ArrayList<>());\r
+\r
+ @Override\r
+ public void execute(DelegateExecution execution) throws Exception {\r
+ callTestHead(\r
+ execution.getCurrentActivityId(),\r
+ execution.getProcessDefinitionId(),\r
+ execution.getProcessInstanceId(),\r
+ execution.getProcessBusinessKey(),\r
+ execution.getVariables());\r
+ }\r
+\r
+ public void callTestHead(\r
+ String currentActivityId,\r
+ String processDefinitionId,\r
+ String processInstanceId,\r
+ String processBusinessKey,\r
+ Map<String, Object> variables)\r
+ throws Exception {\r
+ final String logPrefix = Utility.getLoggerPrefix();\r
+ logger.info(logPrefix + "::execute()");\r
+\r
+ // Get vthInput from the Camunda execution variable map.\r
+ List<Map<String, Object>> activityParameters = utility.getVthInput(variables, currentActivityId, logPrefix);\r
+\r
+ // Get the current test execution object.\r
+ TestExecution testExecution = utility.getTestExecution(variables, logPrefix);\r
+\r
+ // Lookup the test head before making computations in the loop, and before calling the runnable.\r
+ // If the lookup is made inside the runnable, concurrent test head calls would bombard the db.\r
+ TestHead testHead = getTestHead(testExecution, currentActivityId, processDefinitionId);\r
+\r
+ WorkflowTask workflowTask = new WorkflowTask(processInstanceId, activityParameters.size(), false);\r
+ ExecutorService pool = workflowTask.getPool();\r
+\r
+ // Try to cast each parameter to a Map, and create runnable tasks.\r
+ for (int i = 0; i < activityParameters.size(); i++) {\r
+ Object oTestHeadParameter = activityParameters.get(i);\r
+ Map<?, ?> mTestHeadParameter;\r
+ try {\r
+ mTestHeadParameter = Utility.toMap(oTestHeadParameter);\r
+ verifyOtfTestHead(mTestHeadParameter, testHead, testExecution, currentActivityId);\r
+ } catch (Exception e) {\r
+ // TODO: Make a design decision to either stop the execution, or attempt to convert the\r
+ // other parameters.\r
+ logger.error(\r
+ String.format(\r
+ "Unable to convert test head parameter at vthInput[%s][%d] to a Map.",\r
+ currentActivityId, i));\r
+ continue;\r
+ }\r
+\r
+ // Get all the arguments for the runnable.\r
+ Object oHeaders = mTestHeadParameter.get("headers"); // optional\r
+ Object oMethod = mTestHeadParameter.get("method"); // required\r
+ Object oPayload = mTestHeadParameter.get("payload"); // optional\r
+ Object oTimeoutInMillis = mTestHeadParameter.get("timeoutInMillis"); // optional\r
+\r
+ // Target typed parameters. Convert all objects to their expected types. Throw exceptions for\r
+ // required parameters, or for parameters that are provided but not of the expected type.\r
+ Map<String, String> headers = new HashMap<>();\r
+ String method = "";\r
+ Map<String, Object> payload = new HashMap<>();\r
+ int timeoutInMillis = 0;\r
+\r
+ if (oHeaders != null) {\r
+ try {\r
+ headers = (Map<String, String>) Utility.toMap(oHeaders);\r
+ } catch (Exception e) {\r
+ logger.error(\r
+ String.format(\r
+ "Unable to convert test head parameter at vthInput[%s][%d][headers] to a Map.",\r
+ currentActivityId, i));\r
+ }\r
+ }\r
+\r
+ if (oMethod == null) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "vthInput[%s][%d][method] is a required parameter.", currentActivityId, i));\r
+ } else {\r
+ try {\r
+ method = (String) oMethod;\r
+ } catch (ClassCastException cce) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Unable to read vthInput[%s][%d][method] as primitive type String.",\r
+ processInstanceId, i));\r
+ }\r
+ }\r
+\r
+ if (oPayload != null) {\r
+ try {\r
+ payload = (Map<String, Object>) Utility.toMap(oPayload);\r
+ } catch (Exception e) {\r
+ logger.error(\r
+ String.format(\r
+ "Unable to convert test head parameter at vthInput[%s][%d][payload] to a Map.",\r
+ currentActivityId, i));\r
+ }\r
+ }\r
+\r
+ if (oTimeoutInMillis != null) {\r
+ try {\r
+ timeoutInMillis = (int) oTimeoutInMillis;\r
+ } catch (ClassCastException cce) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Unable to read vthInput[%s][%d][timeoutInMillis] as primitive type int.",\r
+ currentActivityId, i));\r
+ }\r
+ }\r
+\r
+// logger.info("{}(BEFORE) PRINTING THREAD INFORMATION", logPrefix);\r
+// WorkflowTask.printThreadInformation();\r
+// logger.info("{}(BEFORE) PRINTING WORKFLOW TASKS", logPrefix);\r
+// WorkflowTask.printWorkflowTaskResources();\r
+ TestHeadCallable callable =\r
+ new TestHeadCallable(\r
+ timeoutInMillis,\r
+ method,\r
+ headers,\r
+ payload,\r
+ testHead,\r
+ currentActivityId,\r
+ testExecution,\r
+ mongoOperation);\r
+\r
+ // Submit the test head call to the executor service.\r
+ workflowTask.getFutures().add(pool.submit(callable));\r
+ }\r
+\r
+ // Prevent new tasks from being submitted, and allow running tasks to finish.\r
+ pool.shutdown();\r
+\r
+ int numResults = 0;\r
+ while (!pool.isTerminated()) {\r
+ try {\r
+ pool.awaitTermination(1, TimeUnit.SECONDS);\r
+ } catch (InterruptedException e) {\r
+ workflowTask.shutdown(true);\r
+ throw e;\r
+ }\r
+ }\r
+\r
+ workflowTask.shutdown(false);\r
+\r
+// logger.info("{}(AFTER) PRINTING THREAD INFORMATION", logPrefix);\r
+// WorkflowTask.printThreadInformation();\r
+// logger.info("{}(AFTER) PRINTING WORKFLOW TASKS", logPrefix);\r
+// WorkflowTask.printWorkflowTaskResources();\r
+ }\r
+\r
+ private void saveTestHeadResults(String businessKey) {\r
+ Query query = new Query();\r
+ query.addCriteria(Criteria.where("businessKey").is(businessKey));\r
+ Update update = new Update();\r
+ update.set("testHeadResults", testHeadResults);\r
+ UpdateResult result = mongoOperation.updateFirst(query, update, TestExecution.class);\r
+ // Check the status of the findAndUpdate database, and appropriately handle the errors.\r
+ if (result.getMatchedCount() == 0) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Unable to log the test result because a testExecution associated with businessKey, %s, was not found.",\r
+ businessKey));\r
+ } else if (result.getModifiedCount() == 0) {\r
+ throw new TestExecutionException("Unable to persist the testExecution to the database.");\r
+ }\r
+ }\r
+\r
+ private TestHead getTestHead(\r
+ TestExecution testExecution, String currentActivityId, String processDefinitionId) {\r
+ List<BpmnInstance> bpmnInstances = testExecution.getHistoricTestDefinition().getBpmnInstances();\r
+ BpmnInstance bpmnInstance =\r
+ bpmnInstances.stream()\r
+ .filter(\r
+ _bpmnInstance ->\r
+ _bpmnInstance.getProcessDefinitionId().equalsIgnoreCase(processDefinitionId))\r
+ .findFirst()\r
+ .orElse(null);\r
+\r
+ if (bpmnInstance == null) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Error looking BpmnInstance with processDefinitionId %s.", processDefinitionId));\r
+ }\r
+\r
+ List<TestHeadNode> testHeads = bpmnInstance.getTestHeads();\r
+ TestHeadNode testHeadNode =\r
+ testHeads.stream()\r
+ .filter(testHead -> testHead.getBpmnVthTaskId().equals(currentActivityId))\r
+ .findAny()\r
+ .orElse(null);\r
+\r
+ if (testHeadNode == null) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "No test head associated with the currentActivityId %s.", currentActivityId));\r
+ }\r
+\r
+ TestHead testHead = Generic.findByIdGeneric(testHeadRepository, testHeadNode.getTestHeadId());\r
+ if (testHead == null) {\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "The test head with id, %s, was not found in the database.",\r
+ testHeadNode.getTestHeadId()));\r
+ }\r
+ User testExecUser = userRepository.findById(testExecution.getExecutorId().toString()).orElse(null);\r
+ Group testheadGroup = groupRepository.findById(testHead.getGroupId().toString()).orElse(null);\r
+ if(testExecUser == null){\r
+ throw new TestExecutionException(\r
+ String.format("Can not find user, user id: %s",testExecution.getExecutorId().toString()));\r
+ }\r
+ if(testheadGroup == null){\r
+ throw new TestExecutionException(\r
+ String.format("Can not find test head group, group id: %s",testHead.getGroupId().toString())\r
+ );\r
+ }\r
+\r
+ if( (testHead.isPublic() != null && !testHead.isPublic()) &&\r
+ !PermissionChecker.hasPermissionTo(testExecUser,testheadGroup,UserPermission.Permission.EXECUTE,groupRepository)){\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "User(%s) does not have permission to in testHead Group(%s)",\r
+ testExecUser.get_id().toString(),testheadGroup.get_id().toString()\r
+ ));\r
+ }\r
+ return testHead;\r
+ }\r
+\r
+ private void verifyOtfTestHead(Map activityParams, TestHead testHead, TestExecution execution, String currentActivityId){\r
+ String testHeadName = testHead.getTestHeadName().toLowerCase();\r
+ switch(testHeadName) {\r
+ case "robot":\r
+ try {\r
+ TestInstance testInstance = Generic.findByIdGeneric(testInstanceRepository, execution.getHistoricTestInstance().get_id());\r
+ Map<String, Object> internalTestDataByActivity = (Map<String, Object>) testInstance.getInternalTestData().get(currentActivityId);\r
+ String robotFileId = (String) internalTestDataByActivity.get("robotFileId");\r
+ Map<String, Object> testData = new HashMap<>();\r
+ Map<String, Object> vthInput = new HashMap<>();\r
+ testData.put("robotFileId", robotFileId);\r
+ vthInput.put("testData", testData);\r
+ Map<String, Object> payload = (Map<String, Object>) activityParams.get("payload");\r
+ payload.put("vthInput", vthInput);\r
+ }\r
+ catch (Exception e){\r
+ throw new TestExecutionException(\r
+ String.format(\r
+ "Robot test head needs a robot file id: %s.", e.getMessage()));\r
+ }\r
+ break;\r
+ default:\r
+ break;\r
+ }\r
+ }\r
+}\r
Code Review / it / otf.git / blobdiff