TEP-0090: Refactor execution status validation

Today, the execution status validation is long and has duplicated
code. We need to validate execution status in matrix, a feature
proposed in TEP-0090. To set up for that, we are refactoring the
execution status validation in this change which includes splitting
up the function and reusing common logic.

pkg/apis/pipeline/v1beta1/pipeline_types.go

@@ -296,6 +296,82 @@ func (pt *PipelineTask) validateMatrix(ctx context.Context) (errs *apis.FieldErr
  return errs
 }
 
+func (pt *PipelineTask) validateExecutionStatusVariablesDisallowed() (errs *apis.FieldError) {
+ for _, param := range pt.Params {
+ if expressions, ok := GetVarSubstitutionExpressionsForParam(param); ok {
+ errs = errs.Also(validateContainsExecutionStatusVariablesDisallowed(expressions, "value").
+ ViaFieldKey("params", param.Name))
+ }
+ }
+ for i, we := range pt.WhenExpressions {
+ if expressions, ok := we.GetVarSubstitutionExpressions(); ok {
+ errs = errs.Also(validateContainsExecutionStatusVariablesDisallowed(expressions, "").
+ ViaFieldIndex("when", i))
+ }
+ }
+ return errs
+}
+
+func (pt *PipelineTask) validateExecutionStatusVariablesAllowed(ptNames sets.String) (errs *apis.FieldError) {
+ for _, param := range pt.Params {
+ if expressions, ok := GetVarSubstitutionExpressionsForParam(param); ok {
+ errs = errs.Also(validateExecutionStatusVariablesExpressions(expressions, ptNames, "value").
+ ViaFieldKey("params", param.Name))
+ }
+ }
+ for i, we := range pt.WhenExpressions {
+ if expressions, ok := we.GetVarSubstitutionExpressions(); ok {
+ errs = errs.Also(validateExecutionStatusVariablesExpressions(expressions, ptNames, "").
+ ViaFieldIndex("when", i))
+ }
+ }
+ return errs
+}
+
+func validateContainsExecutionStatusVariablesDisallowed(expressions []string, path string) (errs *apis.FieldError) {
+ if containsExecutionStatusReferences(expressions) {
+ errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("pipeline tasks can not refer to execution status"+
+ " of any other pipeline task or aggregate status of tasks"), path))
+ }
+ return errs
+}
+
+func containsExecutionStatusReferences(expressions []string) bool {
+ // validate tasks.pipelineTask.status/tasks.status if this expression is not a result reference
+ if !LooksLikeContainsResultRefs(expressions) {
+ for _, e := range expressions {
+ // check if it contains context variable accessing execution status - $(tasks.taskname.status)
+ // or an aggregate status - $(tasks.status)
+ if containsExecutionStatusRef(e) {
+ return true
+ }
+ }
+ }
+ return false
+}
+
+func validateExecutionStatusVariablesExpressions(expressions []string, ptNames sets.String, fieldPath string) (errs *apis.FieldError) {
+ // validate tasks.pipelineTask.status if this expression is not a result reference
+ if !LooksLikeContainsResultRefs(expressions) {
+ for _, expression := range expressions {
+ // its a reference to aggregate status of dag tasks - $(tasks.status)
+ if expression == PipelineTasksAggregateStatus {
+ continue
+ }
+ // check if it contains context variable accessing execution status - $(tasks.taskname.status)
+ if containsExecutionStatusRef(expression) {
+ // strip tasks. and .status from tasks.taskname.status to further verify task name
+ pt := strings.TrimSuffix(strings.TrimPrefix(expression, "tasks."), ".status")
+ // report an error if the task name does not exist in the list of dag tasks
+ if !ptNames.Has(pt) {
+ errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("pipeline task %s is not defined in the pipeline", pt), fieldPath))
+ }
+ }
+ }
+ }
+ return errs
+}
+
 // TaskSpecMetadata returns the metadata of the PipelineTask's EmbeddedTask spec.
 func (pt *PipelineTask) TaskSpecMetadata() PipelineTaskMetadata {
  return pt.TaskSpec.Metadata

pkg/apis/pipeline/v1beta1/pipeline_validation.go

@@ -205,89 +205,23 @@ func containsExecutionStatusRef(p string) bool {
 // dag tasks cannot have param value as $(tasks.pipelineTask.status)
 func validateExecutionStatusVariablesInTasks(tasks []PipelineTask) (errs *apis.FieldError) {
  for idx, t := range tasks {
- for _, param := range t.Params {
- // retrieve a list of substitution expression from a param
- if ps, ok := GetVarSubstitutionExpressionsForParam(param); ok {
- // validate tasks.pipelineTask.status/tasks.status if this expression is not a result reference
- if !LooksLikeContainsResultRefs(ps) {
- for _, p := range ps {
- // check if it contains context variable accessing execution status - $(tasks.taskname.status)
- // or an aggregate status - $(tasks.status)
- if containsExecutionStatusRef(p) {
- errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("pipeline tasks can not refer to execution status of any other pipeline task"+
- " or aggregate status of tasks"), "value").ViaFieldKey("params", param.Name).ViaFieldIndex("tasks", idx))
- }
- }
- }
- }
- }
- for i, we := range t.WhenExpressions {
- // retrieve a list of substitution expression from a when expression
- if expressions, ok := we.GetVarSubstitutionExpressions(); ok {
- // validate tasks.pipelineTask.status/tasks.status if this expression is not a result reference
- if !LooksLikeContainsResultRefs(expressions) {
- for _, e := range expressions {
- // check if it contains context variable accessing execution status - $(tasks.taskname.status)
- // or an aggregate status - $(tasks.status)
- if containsExecutionStatusRef(e) {
- errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("when expressions in pipeline tasks can not refer to execution status of any other pipeline task"+
- " or aggregate status of tasks"), "").ViaFieldIndex("when", i).ViaFieldIndex("tasks", idx))
- }
- }
- }
- }
- }
+ errs = errs.Also(t.validateExecutionStatusVariablesDisallowed()).ViaIndex(idx)
  }
  return errs
 }
 
 // validate finally tasks accessing execution status of a dag task specified in the pipeline
 // $(tasks.pipelineTask.status) is invalid if pipelineTask is not defined as a dag task
-func validateExecutionStatusVariablesInFinally(tasks []PipelineTask, finally []PipelineTask) (errs *apis.FieldError) {
- // creating a list of pipelineTask names to validate tasks.<name>.status
- ptNames := PipelineTaskList(tasks).Names()
+func validateExecutionStatusVariablesInFinally(tasksNames sets.String, finally []PipelineTask) (errs *apis.FieldError) {
  for idx, t := range finally {
- for _, param := range t.Params {
- if expressions, ok := GetVarSubstitutionExpressionsForParam(param); ok {
- errs = errs.Also(validateExecutionStatusVariablesExpressions(expressions, ptNames, "value").ViaFieldKey(
- "params", param.Name).ViaFieldIndex("finally", idx))
- }
- }
- for i, we := range t.WhenExpressions {
- if expressions, ok := we.GetVarSubstitutionExpressions(); ok {
- errs = errs.Also(validateExecutionStatusVariablesExpressions(expressions, ptNames, "").ViaFieldIndex(
- "when", i).ViaFieldIndex("finally", idx))
- }
- }
- }
- return errs
-}
-
-func validateExecutionStatusVariablesExpressions(expressions []string, ptNames sets.String, fieldPath string) (errs *apis.FieldError) {
- // validate tasks.pipelineTask.status if this expression is not a result reference
- if !LooksLikeContainsResultRefs(expressions) {
- for _, expression := range expressions {
- // its a reference to aggregate status of dag tasks - $(tasks.status)
- if expression == PipelineTasksAggregateStatus {
- continue
- }
- // check if it contains context variable accessing execution status - $(tasks.taskname.status)
- if containsExecutionStatusRef(expression) {
- // strip tasks. and .status from tasks.taskname.status to further verify task name
- pt := strings.TrimSuffix(strings.TrimPrefix(expression, "tasks."), ".status")
- // report an error if the task name does not exist in the list of dag tasks
- if !ptNames.Has(pt) {
- errs = errs.Also(apis.ErrInvalidValue(fmt.Sprintf("pipeline task %s is not defined in the pipeline", pt), fieldPath))
- }
- }
- }
+ errs = errs.Also(t.validateExecutionStatusVariablesAllowed(tasksNames).ViaIndex(idx))
  }
  return errs
 }
 
 func validateExecutionStatusVariables(tasks []PipelineTask, finallyTasks []PipelineTask) (errs *apis.FieldError) {
- errs = errs.Also(validateExecutionStatusVariablesInTasks(tasks))
- errs = errs.Also(validateExecutionStatusVariablesInFinally(tasks, finallyTasks))
+ errs = errs.Also(validateExecutionStatusVariablesInTasks(tasks).ViaField("tasks"))
+ errs = errs.Also(validateExecutionStatusVariablesInFinally(PipelineTaskList(tasks).Names(), finallyTasks).ViaField("finally"))
  return errs
 }
 

pkg/apis/pipeline/v1beta1/pipeline_validation_test.go

@@ -2420,10 +2420,7 @@ func TestPipelineTasksExecutionStatus(t *testing.T) {
  }},
  expectedError: *apis.ErrGeneric("").Also(&apis.FieldError{
  Message: `invalid value: pipeline tasks can not refer to execution status of any other pipeline task or aggregate status of tasks`,
- Paths: []string{"tasks[0].params[bar-status].value"},
- }).Also(&apis.FieldError{
- Message: `invalid value: when expressions in pipeline tasks can not refer to execution status of any other pipeline task or aggregate status of tasks`,
- Paths: []string{"tasks[0].when[0]"},
+ Paths: []string{"tasks[0].params[bar-status].value", "tasks[0].when[0]"},
  }),
  }, {
  name: "invalid string variable in dag task accessing aggregate status of tasks",