Add notification validation (flyteorg#442)

* checkpoint

Signed-off-by: Katrina Rogan <[email protected]>

* lint

Signed-off-by: Katrina Rogan <[email protected]>

* rename

Signed-off-by: Katrina Rogan <[email protected]>

* more tests, fine

Signed-off-by: Katrina Rogan <[email protected]>

* cleanup

Signed-off-by: Katrina Rogan <[email protected]>

flyteadmin/pkg/manager/impl/validation/execution_validator.go

@@ -62,6 +62,11 @@ func ValidateExecutionRequest(ctx context.Context, request admin.ExecutionCreate
 	if err := validateLiteralMap(request.Inputs, shared.Inputs); err != nil {
 		return err
 	}
+	if request.Spec.GetNotifications() != nil {
+		if err := validateNotifications(request.Spec.GetNotifications().Notifications); err != nil {
+			return err
+		}
+	}
 	// TODO: Remove redundant validation with the rest of the method.
 	// This final call to validating the request ensures the notification types are expected.
 	if err := request.Validate(); err != nil {

flyteadmin/pkg/manager/impl/validation/launch_plan_validator.go

@@ -50,6 +50,11 @@ func ValidateLaunchPlan(ctx context.Context,
 	}
 	// Augment default inputs with the unbound workflow inputs.
 	request.Spec.DefaultInputs = expectedInputs
+	if request.Spec.EntityMetadata != nil {
+		if err := validateNotifications(request.Spec.EntityMetadata.Notifications); err != nil {
+			return err
+		}
+	}
 	// TODO: Remove redundant validation that occurs with launch plan and the validate method for the message.
 	// Ensure the notification types are validated.
 	if err := request.Validate(); err != nil {

flyteadmin/pkg/manager/impl/validation/notifications_validator.go

@@ -0,0 +1,49 @@
+package validation
+
+import (
+	"github.com/flyteorg/flyteadmin/pkg/common"
+	"github.com/flyteorg/flyteadmin/pkg/manager/impl/shared"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+)
+
+// TODO: maybe add more sophisticated email validation.
+func validateRecipientsEmail(recipients []string) error {
+	if len(recipients) == 0 {
+		return shared.GetMissingArgumentError("recipients")
+	}
+	for _, recipient := range recipients {
+		if len(recipient) == 0 {
+			return shared.GetMissingArgumentError("recipient")
+		}
+	}
+	return nil
+}
+
+func validateNotifications(notifications []*admin.Notification) error {
+	for _, notif := range notifications {
+		switch {
+		case notif.GetEmail() != nil:
+			if err := validateRecipientsEmail(notif.GetEmail().RecipientsEmail); err != nil {
+				return err
+			}
+		case notif.GetSlack() != nil:
+			if err := validateRecipientsEmail(notif.GetSlack().RecipientsEmail); err != nil {
+				return err
+			}
+		case notif.GetPagerDuty() != nil:
+			if err := validateRecipientsEmail(notif.GetPagerDuty().RecipientsEmail); err != nil {
+				return err
+			}
+		default:
+			return shared.GetInvalidArgumentError("notification type")
+		}
+
+		for _, phase := range notif.Phases {
+			if !common.IsExecutionTerminal(phase) {
+				return shared.GetInvalidArgumentError("phase")
+			}
+		}
+	}
+
+	return nil
+}

flyteadmin/pkg/manager/impl/validation/notifications_validator_test.go

@@ -0,0 +1,144 @@
+package validation
+
+import (
+	"testing"
+
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+var recipients = []string{"[email protected]"}
+var emptyRecipients = make([]string, 0)
+
+func assertInvalidArgument(t *testing.T, err error) {
+	s, ok := status.FromError(err)
+	assert.True(t, ok)
+	assert.Equal(t, codes.InvalidArgument, s.Code())
+}
+
+func TestValidateRecipientsEmail(t *testing.T) {
+	t.Run("valid emails", func(t *testing.T) {
+		assert.NoError(t, validateRecipientsEmail(recipients))
+	})
+	t.Run("invalid recipients", func(t *testing.T) {
+		err := validateRecipientsEmail(nil)
+		assertInvalidArgument(t, err)
+	})
+	t.Run("invalid recipient", func(t *testing.T) {
+		err := validateRecipientsEmail(emptyRecipients)
+		assertInvalidArgument(t, err)
+	})
+}
+
+func TestValidateNotifications(t *testing.T) {
+	phases := []core.WorkflowExecution_Phase{
+		core.WorkflowExecution_FAILED,
+	}
+	t.Run("email type", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_Email{
+					Email: &admin.EmailNotification{
+						RecipientsEmail: recipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assert.NoError(t, err)
+	})
+	t.Run("email type - invalid", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_Email{
+					Email: &admin.EmailNotification{
+						RecipientsEmail: emptyRecipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assertInvalidArgument(t, err)
+	})
+	t.Run("slack type", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_Slack{
+					Slack: &admin.SlackNotification{
+						RecipientsEmail: recipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assert.NoError(t, err)
+	})
+	t.Run("slack type - invalid", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_Slack{
+					Slack: &admin.SlackNotification{
+						RecipientsEmail: emptyRecipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assertInvalidArgument(t, err)
+	})
+	t.Run("pagerduty type", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_PagerDuty{
+					PagerDuty: &admin.PagerDutyNotification{
+						RecipientsEmail: recipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assert.NoError(t, err)
+	})
+	t.Run("pagerduty type - invalid", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_PagerDuty{
+					PagerDuty: &admin.PagerDutyNotification{
+						RecipientsEmail: emptyRecipients,
+					},
+				},
+				Phases: phases,
+			},
+		})
+		assertInvalidArgument(t, err)
+	})
+	t.Run("invalid recipients", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_PagerDuty{
+					PagerDuty: &admin.PagerDutyNotification{},
+				},
+				Phases: phases,
+			},
+		})
+		assertInvalidArgument(t, err)
+	})
+	t.Run("invalid phases", func(t *testing.T) {
+		err := validateNotifications([]*admin.Notification{
+			{
+				Type: &admin.Notification_PagerDuty{
+					PagerDuty: &admin.PagerDutyNotification{
+						RecipientsEmail: recipients,
+					},
+				},
+				Phases: []core.WorkflowExecution_Phase{
+					core.WorkflowExecution_QUEUED,
+				},
+			},
+		})
+		assertInvalidArgument(t, err)
+	})
+}