Implement Sandbox notifications processor and publisher

pkg/async/notifications/factory.go

@@ -120,6 +120,9 @@ func NewNotificationsProcessor(config runtimeInterfaces.NotificationsConfig, sco
 		}
 		emailer = GetEmailer(config, scope)
 		return implementations.NewGcpProcessor(sub, emailer, scope)
+	case common.Sandbox:
+		emailer = GetEmailer(config, scope)
+		return implementations.NewSandboxProcessor(emailer)
 	case common.Local:
 		fallthrough
 	default:
@@ -171,6 +174,8 @@ func NewNotificationsPublisher(config runtimeInterfaces.NotificationsConfig, sco
 			panic(err)
 		}
 		return implementations.NewPublisher(publisher, scope)
+	case common.Sandbox:
+		return implementations.NewSandboxPublisher()
 	case common.Local:
 		fallthrough
 	default:

pkg/async/notifications/factory_test.go

@@ -1,13 +1,32 @@
 package notifications
 
 import (
+	"context"
 	"testing"
 
+	"github.com/flyteorg/flyteadmin/pkg/async/notifications/implementations"
 	runtimeInterfaces "github.com/flyteorg/flyteadmin/pkg/runtime/interfaces"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
 	"github.com/flyteorg/flytestdlib/promutils"
 	"github.com/stretchr/testify/assert"
 )
 
+var (
+	scope               = promutils.NewScope("test_sandbox_processor")
+	notificationsConfig = runtimeInterfaces.NotificationsConfig{
+		Type: "sandbox",
+	}
+	testEmail = admin.EmailMessage{
+		RecipientsEmail: []string{
+			"a@example.com",
+			"b@example.com",
+		},
+		SenderEmail: "no-reply@example.com",
+		SubjectLine: "Test email",
+		Body:        "This is a sample email.",
+	}
+)
+
 func TestGetEmailer(t *testing.T) {
 	defer func() { r := recover(); assert.NotNil(t, r) }()
 	cfg := runtimeInterfaces.NotificationsConfig{
@@ -23,3 +42,18 @@ func TestGetEmailer(t *testing.T) {
 	// shouldn't reach here
 	t.Errorf("did not panic")
 }
+
+func TestNewNotificationPublisherAndProcessor(t *testing.T) {
+	testSandboxPublisher := NewNotificationsPublisher(notificationsConfig, scope)
+	assert.IsType(t, testSandboxPublisher, &implementations.SandboxPublisher{})
+	testSandboxProcessor := NewNotificationsProcessor(notificationsConfig, scope)
+	assert.IsType(t, testSandboxProcessor, &implementations.SandboxProcessor{})
+
+	go func() {
+		testSandboxProcessor.StartProcessing()
+	}()
+
+	assert.Nil(t, testSandboxPublisher.Publish(context.Background(), "TEST_NOTIFICATION", &testEmail))
+
+	assert.Nil(t, testSandboxProcessor.StopProcessing())
+}

pkg/async/notifications/implementations/sandbox_processor.go

@@ -0,0 +1,60 @@
+package implementations
+
+import (
+	"context"
+	"time"
+
+	"github.com/flyteorg/flyteadmin/pkg/async"
+	"github.com/flyteorg/flyteadmin/pkg/async/notifications/interfaces"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+	"github.com/flyteorg/flytestdlib/logger"
+	"github.com/golang/protobuf/proto"
+)
+
+type SandboxProcessor struct {
+	email interfaces.Emailer
+}
+
+func (p *SandboxProcessor) StartProcessing() {
+	for {
+		logger.Warningf(context.Background(), "Starting SandBox notifications processor")
+		err := p.run()
+		logger.Errorf(context.Background(), "error with running processor err: [%v] ", err)
+		time.Sleep(async.RetryDelay)
+	}
+}
+
+func (p *SandboxProcessor) run() error {
+	var emailMessage admin.EmailMessage
+
+	for {
+		select {
+		case msg := <-msgChan:
+			err := proto.Unmarshal(msg, &emailMessage)
+			if err != nil {
+				logger.Errorf(context.Background(), "error with unmarshalling message [%v]", err)
+				return err
+			}
+
+			err = p.email.SendEmail(context.Background(), emailMessage)
+			if err != nil {
+				logger.Errorf(context.Background(), "error with sendemail message [%v] ", err)
+				return err
+			}
+		default:
+			logger.Debugf(context.Background(), "no message to process")
+			return nil
+		}
+	}
+}
+
+func (p *SandboxProcessor) StopProcessing() error {
+	logger.Debug(context.Background(), "call to sandbox stop processing.")
+	return nil
+}
+
+func NewSandboxProcessor(emailer interfaces.Emailer) interfaces.Processor {
+	return &SandboxProcessor{
+		email: emailer,
+	}
+}

pkg/async/notifications/implementations/sandbox_processor_test.go

@@ -0,0 +1,39 @@
+package implementations
+
+import (
+	"context"
+	"testing"
+
+	"github.com/flyteorg/flyteadmin/pkg/async/notifications/mocks"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+	"github.com/golang/protobuf/proto"
+	"github.com/stretchr/testify/assert"
+)
+
+var mockSandboxEmailer mocks.MockEmailer
+
+func TestSandboxProcessor_UnmarshalMessage(t *testing.T) {
+	var emailMessage admin.EmailMessage
+	err := proto.Unmarshal(msg, &emailMessage)
+	assert.Nil(t, err)
+
+	assert.Equal(t, emailMessage.Body, testEmail.Body)
+	assert.Equal(t, emailMessage.RecipientsEmail, testEmail.RecipientsEmail)
+	assert.Equal(t, emailMessage.SubjectLine, testEmail.SubjectLine)
+	assert.Equal(t, emailMessage.SenderEmail, testEmail.SenderEmail)
+}
+
+func TestSandboxProcessor_StartProcessing(t *testing.T) {
+
+	testSandboxProcessor := NewSandboxProcessor(&mockSandboxEmailer)
+
+	sendEmailValidationFunc := func(ctx context.Context, email admin.EmailMessage) error {
+		assert.Equal(t, testEmail.Body, email.Body)
+		assert.Equal(t, testEmail.RecipientsEmail, email.RecipientsEmail)
+		assert.Equal(t, testEmail.SubjectLine, email.SubjectLine)
+		assert.Equal(t, testEmail.SenderEmail, email.SenderEmail)
+		return nil
+	}
+	mockSandboxEmailer.SetSendEmailFunc(sendEmailValidationFunc)
+	assert.Nil(t, testSandboxProcessor.(*SandboxProcessor).run())
+}

pkg/async/notifications/implementations/sandbox_publisher.go

@@ -0,0 +1,31 @@
+package implementations
+
+import (
+	"context"
+
+	"github.com/flyteorg/flytestdlib/logger"
+	"github.com/golang/protobuf/proto"
+)
+
+type SandboxPublisher struct{}
+
+var msgChan = make(chan []byte)
+
+func (p *SandboxPublisher) Publish(ctx context.Context, notificationType string, msg proto.Message) error {
+	logger.Debugf(ctx, "Publishing the following message [%s]", msg.String())
+
+	data, err := proto.Marshal(msg)
+
+	if err != nil {
+		logger.Errorf(ctx, "Failed to publish a message with key [%s] and message [%s] and error: %v", notificationType, msg.String(), err)
+		return err
+	}
+
+	msgChan <- data
+
+	return nil
+}
+
+func NewSandboxPublisher() *SandboxPublisher {
+	return &SandboxPublisher{}
+}

pkg/async/notifications/implementations/sandbox_publisher_test.go

@@ -0,0 +1,36 @@
+package implementations
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSandboxPublisher_Publish(t *testing.T) {
+	publisher := NewSandboxPublisher()
+
+	errChan := make(chan string)
+
+	go func() {
+		select {
+		case <-msgChan:
+			// if message received, no need to send an error
+		case <-time.After(time.Second * 5):
+			errChan <- "No data was received in the channel within the expected time frame"
+		}
+	}()
+
+	err := publisher.Publish(context.Background(), "NOTIFICATION_TYPE", &testEmail)
+
+	// Check if there was an error in the goroutine
+	select {
+	case errMsg := <-errChan:
+		t.Fatal(errMsg)
+	default:
+		// no error from the goroutine
+	}
+
+	assert.Nil(t, err)
+}

pkg/common/cloud.go

@@ -5,8 +5,9 @@ package common
 type CloudProvider = string
 
 const (
-	AWS   CloudProvider = "aws"
-	GCP   CloudProvider = "gcp"
-	Local CloudProvider = "local"
-	None  CloudProvider = "none"
+	AWS     CloudProvider = "aws"
+	GCP     CloudProvider = "gcp"
+	Sandbox CloudProvider = "sandbox"
+	Local   CloudProvider = "local"
+	None    CloudProvider = "none"
 )