GODT-2223: Testing event processing scenarios.

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/Masterminds/semver/v3 v3.1.1
 	github.com/ProtonMail/gluon v0.14.2-0.20230123154940-b7793a0c0bd4
 	github.com/ProtonMail/go-autostart v0.0.0-20210130080809-00ed301c8e9a
-	github.com/ProtonMail/go-proton-api v0.3.1-0.20230124064827-4a09eaa5d1bb
+	github.com/ProtonMail/go-proton-api v0.3.1-0.20230125082844-35702fd064a5
 	github.com/ProtonMail/go-rfc5322 v0.11.0
 	github.com/ProtonMail/gopenpgp/v2 v2.4.10
 	github.com/PuerkitoBio/goquery v1.8.0

go.sum

@@ -41,10 +41,8 @@ github.com/ProtonMail/go-message v0.0.0-20210611055058-fabeff2ec753/go.mod h1:NB
 github.com/ProtonMail/go-mime v0.0.0-20220302105931-303f85f7fe0f/go.mod h1:NYt+V3/4rEeDuaev/zw1zCq8uqVEuPHzDPo3OZrlGJ4=
 github.com/ProtonMail/go-mime v0.0.0-20220429130430-2192574d760f h1:4IWzKjHzZxdrW9k4zl/qCwenOVHDbVDADPPHFLjs0Oc=
 github.com/ProtonMail/go-mime v0.0.0-20220429130430-2192574d760f/go.mod h1:qRZgbeASl2a9OwmsV85aWwRqic0NHPh+9ewGAzb4cgM=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230118091111-93ad9245e8ee h1:kXz09BKBbVLyzXgHztxIAMuvmSF0g8FgGpDaqi2IPiM=
+github.com/ProtonMail/go-proton-api v0.3.1-0.20230125082844-35702fd064a5 h1:ud9ktkYjkggRA7zuR/RfeqHGVWVgu2SypMvK1nThrzA=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230118091111-93ad9245e8ee/go.mod h1:JUo5IQG0hNuPRuDpOUsCOvtee6UjTEHHF1QN2i8RSos=
+github.com/ProtonMail/go-proton-api v0.3.1-0.20230125082844-35702fd064a5/go.mod h1:JUo5IQG0hNuPRuDpOUsCOvtee6UjTEHHF1QN2i8RSos=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230124064827-4a09eaa5d1bb h1:WSKNRN0p0rx1FXq1cGAor3RIoQkX9iD4jQeUlL2s+1o=
-github.com/ProtonMail/go-proton-api v0.3.1-0.20230124064827-4a09eaa5d1bb/go.mod h1:JUo5IQG0hNuPRuDpOUsCOvtee6UjTEHHF1QN2i8RSos=
 github.com/ProtonMail/go-rfc5322 v0.11.0 h1:o5Obrm4DpmQEffvgsVqG6S4BKwC1Wat+hYwjIp2YcCY=
 github.com/ProtonMail/go-rfc5322 v0.11.0/go.mod h1:6oOKr0jXvpoE6pwTx/HukigQpX2J9WUf6h0auplrFTw=
 github.com/ProtonMail/go-srp v0.0.5 h1:xhUioxZgDbCnpo9JehyFhwwsn9JLWkUGfB0oiKXgiGg=

internal/bridge/user_event_test.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -29,6 +31,7 @@ import (
 	"github.com/ProtonMail/proton-bridge/v3/internal/bridge"
 	"github.com/ProtonMail/proton-bridge/v3/internal/constants"
 	"github.com/ProtonMail/proton-bridge/v3/internal/events"
+	"github.com/ProtonMail/proton-bridge/v3/internal/user"
 	"github.com/bradenaw/juniper/xslices"
 	"github.com/emersion/go-imap"
 	"github.com/emersion/go-imap/client"
@@ -37,7 +40,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestBridge_User_BadEvents(t *testing.T) {
+func TestBridge_User_BadMessage_BadEvent(t *testing.T) {
 	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
 		// Create a user.
 		userID, addrID, err := s.CreateUser("user", password)
@@ -51,50 +54,94 @@ func TestBridge_User_BadEvents(t *testing.T) {
 			createNumMessages(ctx, t, c, addrID, labelID, 10)
 		})
 
-		// The initial user should be fully synced.
+		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
-		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
+			userLoginAndSync(ctx, t, bridge, "user", password)
-			syncCh, done := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
-			defer done()
 
-			userID, err := bridge.LoginFull(ctx, "user", password, nil, nil)
+			var messageIDs []string
+
+			// Create 10 more messages for the user, generating events.
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				messageIDs = createNumMessages(ctx, t, c, addrID, labelID, 10)
+			})
+
+			// If bridge attempts to sync the new messages, it should get a BadRequest error.
+			doBadRequest := true
+			s.AddStatusHook(func(req *http.Request) (int, bool) {
+				if !doBadRequest {
+					return 0, false
+				}
+
+				if xslices.Index(xslices.Map(messageIDs, func(messageID string) string {
+					return "/mail/v4/messages/" + messageID
+				}), req.URL.Path) < 0 {
+					return 0, false
+				}
+
+				return http.StatusBadRequest, true
+			})
+
+			userReceiveBadErrorAndLogout(t, bridge, mocks)
+
+			// Remove messages
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				require.NoError(t, c.DeleteMessage(ctx, messageIDs...))
+			})
+			doBadRequest = false
+
+			// Login again
+			_, err := bridge.LoginFull(ctx, "user", password, nil, nil)
 			require.NoError(t, err)
 
-			require.Equal(t, userID, (<-syncCh).UserID)
+			userContinueEventProcess(ctx, t, s, bridge)
-		})
-
-		var messageIDs []string
-
-		// Create 10 more messages for the user, generating events.
-		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			messageIDs = createNumMessages(ctx, t, c, addrID, labelID, 10)
-		})
-
-		// If bridge attempts to sync the new messages, it should get a BadRequest error.
-		s.AddStatusHook(func(req *http.Request) (int, bool) {
-			if xslices.Index(xslices.Map(messageIDs, func(messageID string) string {
-				return "/mail/v4/messages/" + messageID
-			}), req.URL.Path) < 0 {
-				return 0, false
-			}
-
-			return http.StatusBadRequest, true
-		})
-
-		// The user will continue to process events and will receive bad request errors.
-		withMocks(t, func(mocks *bridge.Mocks) {
-			mocks.Reporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Any()).MinTimes(1)
-
-			// The user will eventually be logged out due to the bad request errors.
-			withBridgeNoMocks(ctx, t, mocks, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge) {
-				require.Eventually(t, func() bool {
-					return len(bridge.GetUserIDs()) == 1 && len(getConnectedUserIDs(t, bridge)) == 0
-				}, 10*time.Second, 100*time.Millisecond)
-			})
 		})
 	})
 }
 
-func TestBridge_User_NoBadEvent_SameMessageLabelCreated(t *testing.T) {
+func TestBridge_User_BadMessage_NoBadEvent(t *testing.T) {
+	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
+		// Create a user.
+		_, addrID, err := s.CreateUser("user", password)
+		require.NoError(t, err)
+
+		// Create 10 messages for the user.
+		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+			createNumMessages(ctx, t, c, addrID, proton.InboxLabel, 10)
+		})
+
+		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
+			userLoginAndSync(ctx, t, bridge, "user", password)
+
+			var messageIDs []string
+
+			// If bridge attempts to sync the new messages, it should get a BadRequest error.
+			s.AddStatusHook(func(req *http.Request) (int, bool) {
+				if len(messageIDs) < 2 {
+					return 0, false
+				}
+
+				if strings.Contains(req.URL.Path, "/mail/v4/messages/"+messageIDs[2]) {
+					return http.StatusUnprocessableEntity, true
+				}
+
+				return 0, false
+			})
+
+			// Create 10 more messages for the user, generating events.
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				messageIDs = createNumMessages(ctx, t, c, addrID, proton.InboxLabel, 10)
+			})
+
+			// Remove messages
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				require.NoError(t, c.DeleteMessage(ctx, messageIDs...))
+			})
+
+			userContinueEventProcess(ctx, t, s, bridge)
+		})
+	})
+}
+
+func TestBridge_User_SameMessageLabelCreated_NoBadEvent(t *testing.T) {
 	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
 		// Create a user.
 		userID, addrID, err := s.CreateUser("user", password)
@@ -107,29 +154,22 @@ func TestBridge_User_NoBadEvent_SameMessageLabelCreated(t *testing.T) {
 			messageIDs = createNumMessages(ctx, t, c, addrID, proton.InboxLabel, 10)
 		})
 
-		// The initial user should be fully synced.
 		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
-			syncCh, done := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
+			userLoginAndSync(ctx, t, bridge, "user", password)
-			defer done()
 
-			userID, err := bridge.LoginFull(ctx, "user", password, nil, nil)
+			labelID, err := s.CreateLabel(userID, "folder", "", proton.LabelTypeFolder)
 			require.NoError(t, err)
 
-			require.Equal(t, userID, (<-syncCh).UserID)
+			// Add NOOP events
+			require.NoError(t, s.AddLabelCreatedEvent(userID, labelID))
+			require.NoError(t, s.AddMessageCreatedEvent(userID, messageIDs[9]))
+
+			userContinueEventProcess(ctx, t, s, bridge)
 		})
-
-		labelID, err := s.CreateLabel(userID, "folder", "", proton.LabelTypeFolder)
-		require.NoError(t, err)
-
-		// Add NOOP events
-		require.NoError(t, s.AddLabelCreatedEvent(userID, labelID))
-		require.NoError(t, s.AddMessageCreatedEvent(userID, messageIDs[9]))
-
-		userContinuEventProcess(ctx, t, s, netCtl, locator, storeKey)
 	})
 }
 
-func TestBridge_User_NoBadEvents_MessageLabelDeleted(t *testing.T) {
+func TestBridge_User_MessageLabelDeleted_NoBadEvent(t *testing.T) {
 	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
 		// Create a user.
 		userID, addrID, err := s.CreateUser("user", password)
@@ -143,73 +183,178 @@ func TestBridge_User_NoBadEvents_MessageLabelDeleted(t *testing.T) {
 			createNumMessages(ctx, t, c, addrID, labelID, 10)
 		})
 
-		// The initial user should be fully synced.
 		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
-			syncCh, done := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
+			userLoginAndSync(ctx, t, bridge, "user", password)
-			defer done()
 
-			userID, err := bridge.LoginFull(ctx, "user", password, nil, nil)
+			// Create and delete 10 more messages for the user, generating delete events.
-			require.NoError(t, err)
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				messageIDs := createNumMessages(ctx, t, c, addrID, labelID, 10)
+				require.NoError(t, c.DeleteMessage(ctx, messageIDs...))
+			})
 
-			require.Equal(t, userID, (<-syncCh).UserID)
+			// Create and delete 10 labels for the user, generating delete events.
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				for i := 0; i < 10; i++ {
+					label, err := c.CreateLabel(ctx, proton.CreateLabelReq{
+						Name:  uuid.NewString(),
+						Color: "#f66",
+						Type:  proton.LabelTypeLabel,
+					})
+					require.NoError(t, err)
+
+					require.NoError(t, c.DeleteLabel(ctx, label.ID))
+				}
+			})
+
+			userContinueEventProcess(ctx, t, s, bridge)
 		})
-
-		// Create and delete 10 more messages for the user, generating delete events.
-		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			messageIDs := createNumMessages(ctx, t, c, addrID, labelID, 10)
-			require.NoError(t, c.DeleteMessage(ctx, messageIDs...))
-		})
-
-		// Create and delete 10 labels for the user, generating delete events.
-		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			for i := 0; i < 10; i++ {
-				label, err := c.CreateLabel(ctx, proton.CreateLabelReq{
-					Name:  uuid.NewString(),
-					Color: "#f66",
-					Type:  proton.LabelTypeLabel,
-				})
-				require.NoError(t, err)
-
-				require.NoError(t, c.DeleteLabel(ctx, label.ID))
-			}
-		})
-
-		userContinuEventProcess(ctx, t, s, netCtl, locator, storeKey)
 	})
 }
 
-// userContinuEventProcess checks that user will continue to process events and will not receive any bad request errors.
+func TestBridge_User_AddressEvents_NoBadEvent(t *testing.T) {
-func userContinuEventProcess(
+	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
+		// Create a user.
+		userID, addrID, err := s.CreateUser("user", password)
+		require.NoError(t, err)
+
+		// Create 10 messages for the user.
+		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+			createNumMessages(ctx, t, c, addrID, proton.InboxLabel, 10)
+		})
+
+		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
+			userLoginAndSync(ctx, t, bridge, "user", password)
+
+			addrID, err = s.CreateAddress(userID, "other@pm.me", password)
+			require.NoError(t, err)
+			userContinueEventProcess(ctx, t, s, bridge)
+
+			require.NoError(t, s.AddAddressCreatedEvent(userID, addrID))
+			userContinueEventProcess(ctx, t, s, bridge)
+		})
+
+		otherID, err := s.CreateAddress(userID, "another@pm.me", password)
+		require.NoError(t, err)
+		require.NoError(t, s.RemoveAddress(userID, otherID))
+
+		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
+			userContinueEventProcess(ctx, t, s, bridge)
+
+			require.NoError(t, s.CreateAddressKey(userID, addrID, password))
+			userContinueEventProcess(ctx, t, s, bridge)
+
+			require.NoError(t, s.RemoveAddress(userID, addrID))
+			userContinueEventProcess(ctx, t, s, bridge)
+		})
+	})
+}
+
+func TestBridge_User_Network_NoBadEvents(t *testing.T) {
+	withEnv(t, func(ctx context.Context, s *server.Server, netCtl *proton.NetCtl, locator bridge.Locator, storeKey []byte) {
+		retVal := int32(0)
+
+		setResponseAndWait := func(status int32) {
+			atomic.StoreInt32(&retVal, status)
+			time.Sleep(user.EventPeriod)
+		}
+
+		s.AddStatusHook(func(req *http.Request) (int, bool) {
+			status := atomic.LoadInt32(&retVal)
+			if strings.Contains(req.URL.Path, "/core/v4/events/") {
+				return int(status), status != 0
+			}
+
+			return 0, false
+		})
+
+		// Create a user.
+		_, addrID, err := s.CreateUser("user", password)
+		require.NoError(t, err)
+
+		withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, mocks *bridge.Mocks) {
+			userLoginAndSync(ctx, t, bridge, "user", password)
+
+			// Create 10 more messages for the user, generating events.
+			withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
+				createNumMessages(ctx, t, c, addrID, proton.InboxLabel, 10)
+
+				setResponseAndWait(http.StatusInternalServerError)
+				setResponseAndWait(http.StatusServiceUnavailable)
+				setResponseAndWait(http.StatusPaymentRequired)
+				setResponseAndWait(http.StatusForbidden)
+				setResponseAndWait(http.StatusBadRequest)
+				setResponseAndWait(http.StatusUnprocessableEntity)
+				setResponseAndWait(http.StatusTooManyRequests)
+				time.Sleep(10 * time.Second) // needs minimum of 10 seconds to retry
+			})
+
+			setResponseAndWait(0)
+			time.Sleep(10 * time.Second) // needs up to 20 seconds to retry
+			userContinueEventProcess(ctx, t, s, bridge)
+		})
+	})
+}
+
+// userLoginAndSync logs in user and waits until user is fully synced.
+func userLoginAndSync(
+	ctx context.Context,
+	t *testing.T,
+	bridge *bridge.Bridge,
+	username string, password []byte, //nolint:unparam
+) {
+	syncCh, done := chToType[events.Event, events.SyncFinished](bridge.GetEvents(events.SyncFinished{}))
+	defer done()
+
+	userID, err := bridge.LoginFull(ctx, username, password, nil, nil)
+	require.NoError(t, err)
+
+	require.Equal(t, userID, (<-syncCh).UserID)
+}
+
+func userReceiveBadErrorAndLogout(
+	t *testing.T,
+	bridge *bridge.Bridge,
+	mocks *bridge.Mocks,
+) {
+	// The user will continue to process events and will receive bad request errors.
+	mocks.Reporter.EXPECT().ReportMessageWithContext(gomock.Any(), gomock.Any()).MinTimes(1)
+
+	// The user will eventually be logged out due to the bad request errors.
+	require.Eventually(t, func() bool {
+		return len(bridge.GetUserIDs()) == 1 && len(getConnectedUserIDs(t, bridge)) == 0
+	}, 100*user.EventPeriod, user.EventPeriod)
+}
+
+// userContinueEventProcess checks that user will continue to process events and will not receive any bad request errors.
+func userContinueEventProcess(
 	ctx context.Context,
 	t *testing.T,
 	s *server.Server,
-	netCtl *proton.NetCtl,
+	bridge *bridge.Bridge,
-	locator bridge.Locator,
-	storeKey []byte,
 ) {
-	withBridge(ctx, t, s.GetHostURL(), netCtl, locator, storeKey, func(bridge *bridge.Bridge, _ *bridge.Mocks) {
+	info, err := bridge.QueryUserInfo("user")
-		info, err := bridge.QueryUserInfo("user")
+	require.NoError(t, err)
-		require.NoError(t, err)
 
-		client, err := client.Dial(fmt.Sprintf("%v:%v", constants.Host, bridge.GetIMAPPort()))
+	client, err := client.Dial(fmt.Sprintf("%v:%v", constants.Host, bridge.GetIMAPPort()))
-		require.NoError(t, err)
+	require.NoError(t, err)
-		require.NoError(t, client.Login(info.Addresses[0], string(info.BridgePass)))
+	require.NoError(t, client.Login(info.Addresses[0], string(info.BridgePass)))
-		defer func() { _ = client.Logout() }()
+	defer func() { _ = client.Logout() }()
 
-		// Create a new label.
+	randomLabel := uuid.NewString()
-		withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			require.NoError(t, getErr(c.CreateLabel(ctx, proton.CreateLabelReq{
-				Name:  "blabla",
-				Color: "#f66",
-				Type:  proton.LabelTypeLabel,
-			})))
-		})
 
-		// Wait for the label to be created.
+	// Create a new label.
-		require.Eventually(t, func() bool {
+	withClient(ctx, t, s, "user", password, func(ctx context.Context, c *proton.Client) {
-			return xslices.IndexFunc(clientList(client), func(mailbox *imap.MailboxInfo) bool {
+		require.NoError(t, getErr(c.CreateLabel(ctx, proton.CreateLabelReq{
-				return mailbox.Name == "Labels/blabla"
+			Name:  randomLabel,
-			}) >= 0
+			Color: "#f66",
-		}, 10*time.Second, 100*time.Millisecond)
+			Type:  proton.LabelTypeLabel,
+		})))
 	})
+
+	// Wait for the label to be created.
+	require.Eventually(t, func() bool {
+		return xslices.IndexFunc(clientList(client), func(mailbox *imap.MailboxInfo) bool {
+			return mailbox.Name == "Labels/"+randomLabel
+		}) >= 0
+	}, 100*user.EventPeriod, user.EventPeriod)
 }

internal/user/events.go

@@ -506,6 +506,7 @@ func (user *User) handleCreateMessageEvent(ctx context.Context, event proton.Mes
 	if err != nil {
 		// If the message is not found, it means that it has been deleted before we could fetch it.
 		if apiErr := new(proton.APIError); errors.As(err, &apiErr) && apiErr.Status == http.StatusUnprocessableEntity {
+			user.log.WithField("messageID", event.Message.ID).Warn("Cannot add new message: full message is missing on API")
 			return nil, nil
 		}
 
@@ -602,6 +603,7 @@ func (user *User) handleUpdateDraftEvent(ctx context.Context, event proton.Messa
 		if err != nil {
 			// If the message is not found, it means that it has been deleted before we could fetch it.
 			if apiErr := new(proton.APIError); errors.As(err, &apiErr) && apiErr.Status == http.StatusUnprocessableEntity {
+				user.log.WithField("messageID", event.Message.ID).Warn("Cannot add new draft: full message is missing on API")
 				return nil, nil
 			}
 

tests/user_test.go

@@ -24,6 +24,7 @@ import (
 	"net/mail"
 	"strings"
 
+	"github.com/ProtonMail/gluon/rfc822"
 	"github.com/ProtonMail/go-proton-api"
 	"github.com/ProtonMail/gopenpgp/v2/crypto"
 	"github.com/ProtonMail/proton-bridge/v3/internal/bridge"
@@ -265,9 +266,12 @@ func (s *scenario) theFollowingFieldsWereChangedInDraftForAddressOfAccount(draft
 			var changes proton.DraftTemplate
 
 			if wantMessages[0].From != "" {
-				return fmt.Errorf("changing from address is not supported")
+				return fmt.Errorf("changing From address is not supported")
 			}
 
+			changes.Sender = &mail.Address{Address: address}
+			changes.MIMEType = rfc822.TextPlain
+
 			if wantMessages[0].To != "" {
 				changes.ToList = []*mail.Address{{Address: wantMessages[0].To}}
 			}