This is an automated email from the ASF dual-hosted git repository.
leonbao pushed a commit to branch json_split
in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/json_split by this push:
new 3c9b214 [Feature][jsonsplit]add ProcessServiceTest (#5097)
3c9b214 is described below
commit 3c9b214da7b776487b6f1939132763977c11c7cf
Author: Simon <[email protected]>
AuthorDate: Wed Mar 24 15:14:22 2021 +0800
[Feature][jsonsplit]add ProcessServiceTest (#5097)
* Modify Project and ProjectUser Mapper
* Modify Project and ProjectUser Mapper
* project_code is bigint(20)
* modify ERROR name
* modify saveProcessDefine, remove the duplicate code with
createTaskAndRelation
* modify import/export processdefinition, add genProcessData
* fix ut and bug
* code style
* repalce project_id with code
* conflicts solve
* conflicts solve
* conflicts solve
* bugfix
* modify listResources mothod and remove getResourceIds mothod
* 1
* conflicts solve
* modify listResources mothod and remove getResourceIds mothod
* modify listResources mothod and remove getResourceIds mothod
* replace processDefinitionVersion with processDefinitionLog
* codestyle
* codestyle
* add mapper module ut
* codestyle
* fix ProcessInstanceMapperTest
* codestyle
* conflicts solve
* conflicts solve
* conflicts solve
* conflicts solve
* conflicts solve
* fix ProcessInstanceMapperTest
* fix ProjectMapperTest/ProjectUserMapperTest/ScheduleMapperTest
* fix ProjectMapperTest/ProjectUserMapperTest/ScheduleMapperTest
* fix TaskInstanceMapperTest
* add TaskDefinitionLogMapperTest/TaskDefinitionMapperTest and bugfix
* codestyle
* codestyle
* Refactor process lineage
* Refactor process lineage
* codestyle
* codestyle
* Refactor process lineage
* Refactor process lineage
* add WorkFlowLineageMapperTest/WorkFlowLineageServiceTest
* add WorkFlowLineageMapperTest/WorkFlowLineageServiceTest
* add ProcessServiceTest
* 1
* 1
---
.../dolphinscheduler/dao/utils/DagHelper.java | 66 ++++----
.../service/process/ProcessServiceTest.java | 179 ++++++++++++++++++++-
2 files changed, 206 insertions(+), 39 deletions(-)
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
index 970fc47..695a98a 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
@@ -24,15 +24,21 @@ import
org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.process.ProcessDag;
import org.apache.dolphinscheduler.common.task.conditions.ConditionsParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
-import org.apache.dolphinscheduler.common.utils.*;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-
/**
* dag tools
*/
@@ -68,10 +74,10 @@ public class DagHelper {
/**
* generate task nodes needed by dag
*
- * @param taskNodeList taskNodeList
- * @param startNodeNameList startNodeNameList
+ * @param taskNodeList taskNodeList
+ * @param startNodeNameList startNodeNameList
* @param recoveryNodeNameList recoveryNodeNameList
- * @param taskDependType taskDependType
+ * @param taskDependType taskDependType
* @return task node list
*/
public static List<TaskNode>
generateFlowNodeListByStartNode(List<TaskNode> taskNodeList, List<String>
startNodeNameList,
@@ -129,7 +135,7 @@ public class DagHelper {
/**
* find all the nodes that depended on the start node
*
- * @param startNode startNode
+ * @param startNode startNode
* @param taskNodeList taskNodeList
* @return task node list
*/
@@ -154,9 +160,9 @@ public class DagHelper {
/**
* find all nodes that start nodes depend on.
*
- * @param startNode startNode
+ * @param startNode startNode
* @param recoveryNodeNameList recoveryNodeNameList
- * @param taskNodeList taskNodeList
+ * @param taskNodeList taskNodeList
* @return task node list
*/
private static List<TaskNode> getFlowNodeListPre(TaskNode startNode,
List<String> recoveryNodeNameList, List<TaskNode> taskNodeList, List<String>
visitedNodeNameList) {
@@ -189,10 +195,10 @@ public class DagHelper {
/**
* generate dag by start nodes and recovery nodes
*
- * @param totalTaskNodeList totalTaskNodeList
- * @param startNodeNameList startNodeNameList
- * @param recoveryNodeNameList recoveryNodeNameList
- * @param depNodeType depNodeType
+ * @param totalTaskNodeList totalTaskNodeList
+ * @param startNodeNameList startNodeNameList
+ * @param recoveryNodeNameList recoveryNodeNameList
+ * @param depNodeType depNodeType
* @return process dag
* @throws Exception if error throws Exception
*/
@@ -216,7 +222,7 @@ public class DagHelper {
* find node by node name
*
* @param nodeDetails nodeDetails
- * @param nodeName nodeName
+ * @param nodeName nodeName
* @return task node
*/
public static TaskNode findNodeByName(List<TaskNode> nodeDetails, String
nodeName) {
@@ -231,8 +237,8 @@ public class DagHelper {
/**
* the task can be submit when all the depends nodes are forbidden or
complete
*
- * @param taskNode taskNode
- * @param dag dag
+ * @param taskNode taskNode
+ * @param dag dag
* @param completeTaskList completeTaskList
* @return can submit
*/
@@ -262,7 +268,6 @@ public class DagHelper {
* this function parse the condition node to find the right branch.
* also check all the depends nodes forbidden or complete
*
- * @param preNodeName
* @return successor nodes
*/
public static Set<String> parsePostNodes(String preNodeName,
@@ -299,9 +304,6 @@ public class DagHelper {
/**
* if all of the task dependence are skipped, skip it too.
- *
- * @param taskNode
- * @return
*/
private static boolean isTaskNodeNeedSkip(TaskNode taskNode,
Map<String, TaskNode>
skipTaskNodeList
@@ -321,9 +323,6 @@ public class DagHelper {
/**
* parse condition task find the branch process
* set skip flag for another one.
- *
- * @param nodeName
- * @return
*/
public static List<String> parseConditionTask(String nodeName,
Map<String, TaskNode>
skipTaskNodeList,
@@ -358,11 +357,6 @@ public class DagHelper {
/**
* set task node and the post nodes skip flag
- *
- * @param skipNodeName
- * @param dag
- * @param completeTaskList
- * @param skipTaskNodeList
*/
private static void setTaskNodeSkip(String skipNodeName,
DAG<String, TaskNode,
TaskNodeRelation> dag,
@@ -451,9 +445,13 @@ public class DagHelper {
List<TaskNodeRelation> taskNodeRelations = new ArrayList<>();
for (ProcessTaskRelation processTaskRelation : processTaskRelations) {
- if (processTaskRelation.getPreTaskCode() != 0) {
- TaskNode preNode =
taskNodeMap.get(processTaskRelation.getPreTaskCode());
- TaskNode postNode =
taskNodeMap.get(processTaskRelation.getPostTaskCode());
+ long preTaskCode = processTaskRelation.getPreTaskCode();
+ long postTaskCode = processTaskRelation.getPostTaskCode();
+
+ if (processTaskRelation.getPreTaskCode() != 0
+ && taskNodeMap.containsKey(preTaskCode) &&
taskNodeMap.containsKey(postTaskCode)) {
+ TaskNode preNode = taskNodeMap.get(preTaskCode);
+ TaskNode postNode = taskNodeMap.get(postTaskCode);
taskNodeRelations.add(new TaskNodeRelation(preNode.getName(),
postNode.getName()));
}
}
@@ -465,9 +463,6 @@ public class DagHelper {
/**
* is there have conditions after the parent node
- *
- * @param parentNodeName
- * @return
*/
public static boolean haveConditionsAfterNode(String parentNodeName,
DAG<String, TaskNode,
TaskNodeRelation> dag
@@ -489,9 +484,6 @@ public class DagHelper {
/**
* is there have conditions after the parent node
- *
- * @param parentNodeName
- * @return
*/
public static boolean haveConditionsAfterNode(String parentNodeName,
List<TaskNode> taskNodes
diff --git
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
index 364b283..1db3760 100644
---
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
+++
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
@@ -21,27 +21,37 @@ import static
org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PRO
import static
org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
import static
org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
+import static org.mockito.ArgumentMatchers.any;
+
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNode;
-import org.apache.dolphinscheduler.common.task.conditions.ConditionsParameters;
+import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Command;
import org.apache.dolphinscheduler.dao.entity.ProcessData;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProcessInstanceMap;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
+import org.apache.dolphinscheduler.dao.entity.Project;
+import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.CommandMapper;
import org.apache.dolphinscheduler.dao.mapper.ErrorCommandMapper;
+import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationLogMapper;
+import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper;
+import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtilsTest;
@@ -87,7 +97,13 @@ public class ProcessServiceTest {
@Mock
private UserMapper userMapper;
@Mock
- TaskInstanceMapper taskInstanceMapper;
+ private TaskInstanceMapper taskInstanceMapper;
+ @Mock
+ private TaskDefinitionLogMapper taskDefinitionLogMapper;
+ @Mock
+ private ProcessTaskRelationMapper processTaskRelationMapper;
+ @Mock
+ private ProcessDefinitionLogMapper processDefineLogMapper;
@Test
public void testCreateSubCommand() {
@@ -328,4 +344,163 @@ public class ProcessServiceTest {
processService.recurseFindSubProcessId(parentId, ids);
}
+
+ @Test
+ public void testSaveProcessDefinition() {
+ User user = new User();
+ user.setId(1);
+
+ Project project = new Project();
+ project.setCode(1L);
+
+ ProcessData processData = new ProcessData();
+
+ ProcessDefinition processDefinition = new ProcessDefinition();
+ processDefinition.setCode(1L);
+ processDefinition.setId(123);
+ processDefinition.setName("test");
+ processDefinition.setVersion(1);
+ processDefinition.setCode(11L);
+
+ Mockito.when(processDefineMapper.updateById(any())).thenReturn(1);
+ Mockito.when(processDefineLogMapper.insert(any())).thenReturn(1);
+
+ int i = processService.saveProcessDefinition(user, project, "name",
"desc", "locations", "connects", processData, processDefinition);
+ Assert.assertEquals(1, i);
+
+ }
+
+ @Test
+ public void testSwitchVersion() {
+ ProcessDefinition processDefinition = new ProcessDefinition();
+ processDefinition.setCode(1L);
+ processDefinition.setId(123);
+ processDefinition.setName("test");
+ processDefinition.setVersion(1);
+ processDefinition.setCode(11L);
+
+ ProcessDefinitionLog processDefinitionLog = new ProcessDefinitionLog();
+ processDefinitionLog.setCode(1L);
+
+ Mockito.when(processDefineMapper.updateById(any())).thenReturn(1);
+
+ int i = processService.switchVersion(processDefinition,
processDefinitionLog);
+
+ Assert.assertEquals(1, i);
+
+ }
+
+ @Test
+ public void testGenDagGraph() {
+ ProcessDefinition processDefinition = new ProcessDefinition();
+ processDefinition.setCode(1L);
+ processDefinition.setId(123);
+ processDefinition.setName("test");
+ processDefinition.setVersion(1);
+ processDefinition.setCode(11L);
+
+ ProcessTaskRelationLog processTaskRelationLog = new
ProcessTaskRelationLog();
+ processTaskRelationLog.setName("def 1");
+ processTaskRelationLog.setProcessDefinitionVersion(1);
+ processTaskRelationLog.setProjectCode(1L);
+ processTaskRelationLog.setProcessDefinitionCode(1L);
+ processTaskRelationLog.setPostTaskCode(3L);
+ processTaskRelationLog.setPreTaskCode(2L);
+ processTaskRelationLog.setUpdateTime(new Date());
+ processTaskRelationLog.setCreateTime(new Date());
+ List<ProcessTaskRelationLog> list = new ArrayList<>();
+ list.add(processTaskRelationLog);
+
+ TaskDefinitionLog taskDefinition = new TaskDefinitionLog();
+ taskDefinition.setCode(3L);
+ taskDefinition.setName("1-test");
+ taskDefinition.setProjectCode(1L);
+ taskDefinition.setTaskType(TaskType.SHELL);
+ taskDefinition.setUserId(1);
+ taskDefinition.setVersion(2);
+ taskDefinition.setCreateTime(new Date());
+ taskDefinition.setUpdateTime(new Date());
+
+ TaskDefinitionLog td2 = new TaskDefinitionLog();
+ td2.setCode(2L);
+ td2.setName("unit-test");
+ td2.setProjectCode(1L);
+ td2.setTaskType(TaskType.SHELL);
+ td2.setUserId(1);
+ td2.setVersion(1);
+ td2.setCreateTime(new Date());
+ td2.setUpdateTime(new Date());
+
+ List<TaskDefinitionLog> taskDefinitionLogs = new ArrayList<>();
+ taskDefinitionLogs.add(taskDefinition);
+ taskDefinitionLogs.add(td2);
+
+
Mockito.when(taskDefinitionLogMapper.queryByTaskDefinitions(any())).thenReturn(taskDefinitionLogs);
+
Mockito.when(processTaskRelationLogMapper.queryByProcessCodeAndVersion(Mockito.anyLong(),
Mockito.anyInt())).thenReturn(list);
+
+ DAG<String, TaskNode, TaskNodeRelation>
stringTaskNodeTaskNodeRelationDAG =
processService.genDagGraph(processDefinition);
+ Assert.assertNotEquals(0,
stringTaskNodeTaskNodeRelationDAG.getNodesCount());
+
+ }
+
+ @Test
+ public void testGenProcessData() {
+ String processDefinitionJson =
"{\"tasks\":[{\"id\":\"task-0\",\"code\":3,\"version\":0,\"name\":\"1-test\""
+ +
",\"desc\":null,\"type\":\"SHELL\",\"runFlag\":\"FORBIDDEN\",\"loc\":null,\"maxRetryTimes\":0"
+ +
",\"retryInterval\":0,\"params\":null,\"preTasks\":[\"unit-test\"],\"preTaskNodeList\":[{\"code\":2"
+ +
",\"name\":\"unit-test\",\"version\":0}],\"extras\":null,\"depList\":[\"unit-test\"],\"dependence\":null"
+ +
",\"conditionResult\":null,\"taskInstancePriority\":null,\"workerGroup\":null,\"workerGroupId\":null"
+ +
",\"timeout\":{\"enable\":false,\"strategy\":null,\"interval\":0},\"delayTime\":0}]"
+ + ",\"globalParams\":[],\"timeout\":0,\"tenantId\":0}";
+
+ ProcessDefinition processDefinition = new ProcessDefinition();
+ processDefinition.setCode(1L);
+ processDefinition.setId(123);
+ processDefinition.setName("test");
+ processDefinition.setVersion(1);
+ processDefinition.setCode(11L);
+
+ ProcessTaskRelationLog processTaskRelationLog = new
ProcessTaskRelationLog();
+ processTaskRelationLog.setName("def 1");
+ processTaskRelationLog.setProcessDefinitionVersion(1);
+ processTaskRelationLog.setProjectCode(1L);
+ processTaskRelationLog.setProcessDefinitionCode(1L);
+ processTaskRelationLog.setPostTaskCode(3L);
+ processTaskRelationLog.setPreTaskCode(2L);
+ processTaskRelationLog.setUpdateTime(new Date());
+ processTaskRelationLog.setCreateTime(new Date());
+ List<ProcessTaskRelationLog> list = new ArrayList<>();
+ list.add(processTaskRelationLog);
+
+ TaskDefinitionLog taskDefinition = new TaskDefinitionLog();
+ taskDefinition.setCode(3L);
+ taskDefinition.setName("1-test");
+ taskDefinition.setProjectCode(1L);
+ taskDefinition.setTaskType(TaskType.SHELL);
+ taskDefinition.setUserId(1);
+ taskDefinition.setVersion(2);
+ taskDefinition.setCreateTime(new Date());
+ taskDefinition.setUpdateTime(new Date());
+
+ TaskDefinitionLog td2 = new TaskDefinitionLog();
+ td2.setCode(2L);
+ td2.setName("unit-test");
+ td2.setProjectCode(1L);
+ td2.setTaskType(TaskType.SHELL);
+ td2.setUserId(1);
+ td2.setVersion(1);
+ td2.setCreateTime(new Date());
+ td2.setUpdateTime(new Date());
+
+ List<TaskDefinitionLog> taskDefinitionLogs = new ArrayList<>();
+ taskDefinitionLogs.add(taskDefinition);
+ taskDefinitionLogs.add(td2);
+
+
Mockito.when(taskDefinitionLogMapper.queryByTaskDefinitions(any())).thenReturn(taskDefinitionLogs);
+
Mockito.when(processTaskRelationLogMapper.queryByProcessCodeAndVersion(Mockito.anyLong(),
Mockito.anyInt())).thenReturn(list);
+ String json =
JSONUtils.toJsonString(processService.genProcessData(processDefinition));
+
+ Assert.assertEquals(processDefinitionJson, json);
+
+ }
}
