chore: add some corner test

Author: Ethereal-O

vermeer/apps/master/bl/task_bl.go

@@ -67,23 +67,30 @@ func (tb *TaskBl) CreateTaskInfo(
 	// for scheduler
 	taskInfo.Priority = 0
 	taskInfo.Preorders = make([]int32, 0)
-	taskInfo.Exclusive = false // default to false, can be set to true if needed
-	// taskInfo.Exclusive = true
+	taskInfo.Exclusive = true // default to true for now, can be set false by params
 	if params != nil {
 		if priority, ok := params["priority"]; ok {
 			if p, err := strconv.ParseInt(priority, 10, 32); err == nil {
+				if p < 0 {
+					return nil, fmt.Errorf("priority should be non-negative")
+				}
 				taskInfo.Priority = int32(p)
 			} else {
 				logrus.Warnf("priority convert to int32 error:%v", err)
+				return nil, err
 			}
 		}
 		if preorders, ok := params["preorders"]; ok {
 			preorderList := strings.Split(preorders, ",")
 			for _, preorder := range preorderList {
 				if pid, err := strconv.ParseInt(preorder, 10, 32); err == nil {
+					if taskMgr.GetTaskByID(int32(pid)) == nil {
+						return nil, fmt.Errorf("preorder task id %d not exists", pid)
+					}
 					taskInfo.Preorders = append(taskInfo.Preorders, int32(pid))
 				} else {
 					logrus.Warnf("preorder convert to int32 error:%v", err)
+					return nil, err
 				}
 			}
 		}
@@ -92,9 +99,14 @@ func (tb *TaskBl) CreateTaskInfo(
 				taskInfo.Exclusive = ex
 			} else {
 				logrus.Warnf("exclusive convert to bool error:%v", err)
+				return nil, err
 			}
 		}
 		if cronExpr, ok := params["cron_expr"]; ok {
+			if err := Scheduler.cronManager.CheckCronExpression(cronExpr); err != nil {
+				logrus.Warnf("cron_expr parse error:%v", err)
+				return nil, err
+			}
 			taskInfo.CronExpr = cronExpr
 		}
 	}

vermeer/test/functional/compute_base.go

@@ -108,6 +108,19 @@ func (ctb *ComputeTaskBase) SendComputeReqAsyncNotWait(params map[string]string)
 	return int32(resp.Task.ID)
 }
 
+func (ctb *ComputeTaskBase) SendComputeReqAsyncNotWaitWithError(params map[string]string) (int32, error) {
+	//create Compute Task
+	resp, err := ctb.masterHttp.CreateTaskAsync(client.TaskCreateRequest{
+		TaskType:  "compute",
+		GraphName: ctb.graphName,
+		Params:    params,
+	})
+	if err != nil {
+		return -1, err
+	}
+	return int32(resp.Task.ID), nil
+}
+
 func (ctb *ComputeTaskBase) SendComputeReqAsyncBatchPriority(params []map[string]string) ([]int32, []int32) {
 	//create Compute Task
 	tasks := make([]client.TaskInfo, 0, len(params))

vermeer/test/functional/compute_task.go

@@ -37,6 +37,7 @@ type ComputeTask interface {
 	TaskComputeBody() map[string]string
 	SendComputeReqAsync(params map[string]string)
 	SendComputeReqAsyncNotWait(params map[string]string) int32
+	SendComputeReqAsyncNotWaitWithError(params map[string]string) (int32, error)
 	SendComputeReqAsyncBatchPriority(params []map[string]string) ([]int32, []int32)
 	SendComputeReqSync(params map[string]string)
 	LoadComputeRes() ([]interface{}, error)

vermeer/test/functional/http_interface.go

@@ -76,6 +76,16 @@ func (ct CancelTask) CancelTask(t *testing.T, master *client.VermeerClient, grap
 	require.Equal(t, "canceled", task.Task.Status)
 }
 
+func (ct CancelTask) DirectCancelTask(t *testing.T, master *client.VermeerClient, taskID int32) {
+	ok, err := master.GetTaskCancel(int(taskID))
+	require.NoError(t, err)
+	require.Equal(t, true, ok)
+
+	task, err := master.GetTask(int(taskID))
+	require.NoError(t, err)
+	require.Equal(t, "canceled", task.Task.Status)
+}
+
 type GetGraphs struct {
 }
 

vermeer/test/scheduler/priority.go

@@ -2,9 +2,11 @@ package scheduler
 
 import (
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
+	"vermeer/apps/structure"
 	"vermeer/client"
 	"vermeer/test/functional"
 
@@ -143,6 +145,100 @@ func SubTestDepends(t *testing.T, expectRes *functional.ExpectRes, healthCheck *
 	fmt.Printf("Test Depends: %-30s [OK], cost: %v\n", computeTask, time.Since(bTime))
 }
 
+// SubTestInvalidDependency 测试当任务依赖一个不存在的任务ID时，调度器的行为。
+// 调度器应该拒绝此任务，并返回一个错误。
+func SubTestInvalidDependency(t *testing.T, expectRes *functional.ExpectRes, healthCheck *functional.HealthCheck, masterHttp *client.VermeerClient, graphName []string, computeTask string, waitSecond int) {
+	fmt.Printf("Test Invalid Dependency start with task: %s\n", computeTask)
+	bTime := time.Now()
+
+	computeTest, err := functional.MakeComputeTask(computeTask)
+	require.NoError(t, err)
+	computeTest.Init(graphName[0], computeTask, expectRes, waitSecond, masterHttp, t, healthCheck)
+
+	taskBody := computeTest.TaskComputeBody()
+	// 设置 preorders 为一个非常大的、理论上不存在的任务ID
+	invalidTaskID := 999999999
+	taskBody["preorders"] = fmt.Sprintf("%d", invalidTaskID)
+
+	logrus.Infof("Attempting to submit a task with invalid dependency on ID: %d", invalidTaskID)
+
+	// 尝试异步提交任务，并检查是否返回了错误
+	taskID, err := computeTest.SendComputeReqAsyncNotWaitWithError(taskBody)
+
+	// 断言提交操作失败
+	require.Error(t, err, "Submitting a task with a non-existent dependency should return an error.")
+	// 断言返回的任务ID为0，或者其他表示失败的值
+	require.Equal(t, int32(-1), taskID, "The task ID should be zero or invalid on failure.")
+
+	fmt.Printf("Test Invalid Dependency: %-30s [OK], cost: %v\n", computeTask, time.Since(bTime))
+}
+
+func SubTestConcurrentCancellation(t *testing.T, expectRes *functional.ExpectRes, healthCheck *functional.HealthCheck, masterHttp *client.VermeerClient, graphName []string, computeTask string, waitSecond int) {
+	fmt.Printf("Test Concurrent Cancellation start with task: %s\n", computeTask)
+	bTime := time.Now()
+
+	computeTest, err := functional.MakeComputeTask(computeTask)
+	require.NoError(t, err)
+	computeTest.Init(graphName[0], computeTask, expectRes, waitSecond, masterHttp, t, healthCheck)
+
+	// 设置任务数量
+	const numTasks = 20
+	taskBodies := make([]map[string]string, numTasks)
+	for i := 0; i < numTasks; i++ {
+		taskBodies[i] = computeTest.TaskComputeBody()
+	}
+
+	taskIDs := make(chan int32, numTasks)
+	var wg sync.WaitGroup
+
+	// 1. 并发提交任务
+	for i := 0; i < numTasks; i++ {
+		wg.Add(1)
+		go func(body map[string]string) {
+			defer wg.Done()
+			taskID := computeTest.SendComputeReqAsyncNotWait(body)
+			if taskID != 0 {
+				taskIDs <- taskID
+			} else {
+				logrus.Errorf("Failed to submit task: %v", err)
+			}
+		}(taskBodies[i])
+	}
+
+	wg.Wait()
+	close(taskIDs)
+
+	submittedTaskIDs := make([]int32, 0, numTasks)
+	for id := range taskIDs {
+		submittedTaskIDs = append(submittedTaskIDs, id)
+	}
+
+	logrus.Infof("Submitted %d tasks concurrently: %+v", len(submittedTaskIDs), submittedTaskIDs)
+	require.Equal(t, numTasks, len(submittedTaskIDs), "Not all tasks were successfully submitted.")
+
+	cancelTask := functional.CancelTask{}
+	cancelTask.DirectCancelTask(t, masterHttp, submittedTaskIDs[len(submittedTaskIDs)-1])
+
+	// 3. 验证任务状态
+	// 这里需要一个循环来检查所有任务的最终状态
+	// 实际实现中，您可能需要根据调度器的API来轮询任务状态
+	// 在这个示例中，我们只做基本的断言，因为没有实际的取消和状态查询逻辑
+	logrus.Info("Waiting for tasks to settle...")
+	time.Sleep(time.Duration(waitSecond) * time.Second)
+
+	checkTask, err := masterHttp.GetTask(int(submittedTaskIDs[numTasks-1]))
+
+	require.NoError(t, err, "Error fetching task status after cancellation.")
+	require.NotNil(t, checkTask, "Task should exist after cancellation.")
+
+	if structure.TaskState(checkTask.Task.Status) != structure.TaskStateCanceled {
+		logrus.Warn("No tasks were cancelled; check scheduler behavior.")
+		require.Fail(t, "Expected at least some tasks to be cancelled.")
+	}
+
+	fmt.Printf("Test Concurrent Cancellation: %-30s [OK], cost: %v\n", computeTask, time.Since(bTime))
+}
+
 func TestPriority(t *testing.T, expectRes *functional.ExpectRes, healthCheck *functional.HealthCheck, masterHttp *client.VermeerClient, graphName []string, factor string, waitSecond int) {
 	fmt.Print("start test priority\n")
 
@@ -176,4 +272,8 @@ func TestPriority(t *testing.T, expectRes *functional.ExpectRes, healthCheck *fu
 	// 6. send tasks to different graphs
 	// expect: the tasks should be executed concurrently
 	// have been tested in SubTestSmall and SubTestDepends
+
+	SubTestInvalidDependency(t, expectRes, healthCheck, masterHttp, graphName, computeTask, waitSecond)
+
+	SubTestConcurrentCancellation(t, expectRes, healthCheck, masterHttp, graphName, computeTask, 3)
 }