don't throw error on backup with service disabled (#3426)

<!-- PR description-->

If a user does not have a service(like exchange) enabled- don't throw an error but just skip that user and move forward with other backups.

#### Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [ ] 🕐 Yes, but in a later PR
- [x] ⛔ No

#### Type of change

<!--- Please check the type of change your PR introduces: --->
- [x] 🐛 Bugfix

#### Issue(s)

<!-- Can reference multiple issues. Use one of the following "magic words" - "closes, fixes" to auto-close the Github issue. -->
* https://github.com/alcionai/corso/issues/3392

#### Test Plan

<!-- How will this be tested prior to merging.-->
- [x] 💪 Manual
- [ ] ⚡ Unit test
- [x] 💚 E2E

src/cli/backup/backup.go

@@ -12,6 +12,7 @@ import (
 	. "github.com/alcionai/corso/src/cli/print"
 	"github.com/alcionai/corso/src/cli/utils"
 	"github.com/alcionai/corso/src/internal/common/idname"
+	"github.com/alcionai/corso/src/internal/connector/graph"
 	"github.com/alcionai/corso/src/internal/data"
 	"github.com/alcionai/corso/src/pkg/backup"
 	"github.com/alcionai/corso/src/pkg/logger"
@@ -225,6 +226,12 @@ func runBackups(
 
 		err = bo.Run(ictx)
 		if err != nil {
+			if errors.Is(err, graph.ErrServiceNotEnabled) {
+				logger.Ctx(ctx).Infow("service not enabled", "resource_owner_name", bo.ResourceOwner.Name())
+
+				continue
+			}
+
 			errs = append(errs, clues.Wrap(err, owner).WithClues(ictx))
 			Errf(ictx, "%v\n", err)
 

src/cli/backup/exchange_e2e_test.go

@@ -175,6 +175,39 @@ func runExchangeBackupCategoryTest(suite *BackupExchangeE2ESuite, category strin
 	assert.Contains(t, result, suite.m365UserID)
 }
 
+func (suite *BackupExchangeE2ESuite) TestExchangeBackupCmd_ServiceNotEnabled_email() {
+	runExchangeBackupServiceNotEnabledTest(suite, "email")
+}
+
+func runExchangeBackupServiceNotEnabledTest(suite *BackupExchangeE2ESuite, category string) {
+	recorder := strings.Builder{}
+	recorder.Reset()
+
+	t := suite.T()
+
+	ctx, flush := tester.NewContext(t)
+	ctx = config.SetViper(ctx, suite.vpr)
+
+	defer flush()
+
+	// run the command
+
+	cmd, ctx := buildExchangeBackupCmd(
+		ctx,
+		suite.cfgFP,
+		fmt.Sprintf("%s,%s", tester.UnlicensedM365UserID(suite.T()), suite.m365UserID),
+		category,
+		&recorder)
+	err := cmd.ExecuteContext(ctx)
+	require.NoError(t, err, clues.ToCore(err))
+
+	result := recorder.String()
+	t.Log("backup results", result)
+
+	// as an offhand check: the result should contain the m365 user id
+	assert.Contains(t, result, suite.m365UserID)
+}
+
 func (suite *BackupExchangeE2ESuite) TestExchangeBackupCmd_userNotFound_email() {
 	runExchangeBackupUserNotFoundTest(suite, "email")
 }

src/internal/connector/data_collections.go

@@ -153,6 +153,31 @@ func (gc *GraphConnector) ProduceBackupCollections(
 	return colls, ssmb, nil
 }
 
+// IsBackupRunnable verifies that the users provided has the services enabled and
+// data can be backed up. The canMakeDeltaQueries provides info if the mailbox is
+// full and delta queries can be made on it.
+func (gc *GraphConnector) IsBackupRunnable(
+	ctx context.Context,
+	service path.ServiceType,
+	resourceOwner string,
+) (bool, error) {
+	if service == path.SharePointService {
+		// No "enabled" check required for sharepoint
+		return true, nil
+	}
+
+	info, err := gc.Discovery.Users().GetInfo(ctx, resourceOwner)
+	if err != nil {
+		return false, err
+	}
+
+	if !info.ServiceEnabled(service) {
+		return false, clues.Wrap(graph.ErrServiceNotEnabled, "checking service access")
+	}
+
+	return true, nil
+}
+
 func verifyBackupInputs(sels selectors.Selector, siteIDs []string) error {
 	var ids []string
 

src/internal/connector/mock/connector.go

@@ -11,6 +11,7 @@ import (
 	"github.com/alcionai/corso/src/pkg/backup/details"
 	"github.com/alcionai/corso/src/pkg/control"
 	"github.com/alcionai/corso/src/pkg/fault"
+	"github.com/alcionai/corso/src/pkg/path"
 	"github.com/alcionai/corso/src/pkg/selectors"
 )
 
@@ -43,6 +44,14 @@ func (gc GraphConnector) ProduceBackupCollections(
 	return gc.Collections, gc.Exclude, gc.Err
 }
 
+func (gc GraphConnector) IsBackupRunnable(
+	_ context.Context,
+	_ path.ServiceType,
+	_ string,
+) (bool, error) {
+	return true, gc.Err
+}
+
 func (gc GraphConnector) Wait() *data.CollectionStats {
 	return &gc.Stats
 }

src/internal/operations/backup.go

@@ -11,6 +11,7 @@ import (
 	"github.com/alcionai/corso/src/internal/common/dttm"
 	"github.com/alcionai/corso/src/internal/common/idname"
 	"github.com/alcionai/corso/src/internal/common/prefixmatcher"
+	"github.com/alcionai/corso/src/internal/connector/graph"
 	"github.com/alcionai/corso/src/internal/data"
 	"github.com/alcionai/corso/src/internal/diagnostics"
 	"github.com/alcionai/corso/src/internal/events"
@@ -142,6 +143,25 @@ func (op *BackupOperation) Run(ctx context.Context) (err error) {
 	ctx, flushMetrics := events.NewMetrics(ctx, logger.Writer{Ctx: ctx})
 	defer flushMetrics()
 
+	var runnable bool
+
+	// IsBackupRunnable checks if the user has services enabled to run a backup.
+	// it also checks for conditions like mailbox full.
+	runnable, err = op.bp.IsBackupRunnable(ctx, op.Selectors.PathService(), op.ResourceOwner.ID())
+	if err != nil {
+		logger.CtxErr(ctx, err).Error("verifying backup is runnable")
+		op.Errors.Fail(clues.Wrap(err, "verifying backup is runnable"))
+
+		return
+	}
+
+	if !runnable {
+		logger.CtxErr(ctx, graph.ErrServiceNotEnabled).Error("checking if backup is enabled")
+		op.Errors.Fail(clues.Wrap(err, "checking if backup is enabled"))
+
+		return
+	}
+
 	// -----
 	// Setup
 	// -----
@@ -376,7 +396,14 @@ func produceBackupDataCollections(
 		closer()
 	}()
 
-	return bp.ProduceBackupCollections(ctx, resourceOwner, sel, metadata, lastBackupVersion, ctrlOpts, errs)
+	return bp.ProduceBackupCollections(
+		ctx,
+		resourceOwner,
+		sel,
+		metadata,
+		lastBackupVersion,
+		ctrlOpts,
+		errs)
 }
 
 // ---------------------------------------------------------------------------

src/internal/operations/inject/inject.go

@@ -13,6 +13,7 @@ import (
 	"github.com/alcionai/corso/src/pkg/control"
 	"github.com/alcionai/corso/src/pkg/control/repository"
 	"github.com/alcionai/corso/src/pkg/fault"
+	"github.com/alcionai/corso/src/pkg/path"
 	"github.com/alcionai/corso/src/pkg/selectors"
 )
 
@@ -27,6 +28,7 @@ type (
 			ctrlOpts control.Options,
 			errs *fault.Bus,
 		) ([]data.BackupCollection, prefixmatcher.StringSetReader, error)
+		IsBackupRunnable(ctx context.Context, service path.ServiceType, resourceOwner string) (bool, error)
 
 		Wait() *data.CollectionStats
 	}

src/internal/tester/config.go

@@ -31,6 +31,7 @@ const (
 	TestCfgLoadTestUserID   = "loadtestm365userid"
 	TestCfgLoadTestOrgUsers = "loadtestm365orgusers"
 	TestCfgAccountProvider  = "account_provider"
+	TestCfgUnlicensedUserID = "unlicensedm365userid"
 )
 
 // test specific env vars
@@ -43,6 +44,7 @@ const (
 	EnvCorsoM365LoadTestUserID       = "CORSO_M365_LOAD_TEST_USER_ID"
 	EnvCorsoM365LoadTestOrgUsers     = "CORSO_M365_LOAD_TEST_ORG_USERS"
 	EnvCorsoTestConfigFilePath       = "CORSO_TEST_CONFIG_FILE"
+	EnvCorsoUnlicensedM365TestUserID = "CORSO_M365_TEST_UNLICENSED_USER"
 )
 
 // global to hold the test config results.
@@ -152,6 +154,12 @@ func readTestConfig() (map[string]string, error) {
 		os.Getenv(EnvCorsoM365TestSiteURL),
 		vpr.GetString(TestCfgSiteURL),
 		"https://10rqc2.sharepoint.com/sites/CorsoCI")
+	fallbackTo(
+		testEnv,
+		TestCfgUnlicensedUserID,
+		os.Getenv(EnvCorsoUnlicensedM365TestUserID),
+		vpr.GetString(TestCfgUnlicensedUserID),
+		"testevents@10rqc2.onmicrosoft.com")
 
 	testEnv[EnvCorsoTestConfigFilePath] = os.Getenv(EnvCorsoTestConfigFilePath)
 	testConfig = testEnv

src/internal/tester/resource_owners.go

@@ -113,7 +113,7 @@ func LoadTestM365UserID(t *testing.T) string {
 // the delimiter must be a |.
 func LoadTestM365OrgSites(t *testing.T) []string {
 	cfg, err := readTestConfig()
-	require.NoError(t, err, "retrieving load test m365 org sites from test configuration", clues.ToCore(err))
+	require.NoError(t, err, "retrieving load test m365 org sites from test configuration %+v", clues.ToCore(err))
 
 	// TODO: proper handling of site slice input.
 	// sites := cfg[TestCfgLoadTestOrgSites]
@@ -133,7 +133,7 @@ func LoadTestM365OrgSites(t *testing.T) []string {
 // the delimiter may be either a , or |.
 func LoadTestM365OrgUsers(t *testing.T) []string {
 	cfg, err := readTestConfig()
-	require.NoError(t, err, "retrieving load test m365 org users from test configuration", clues.ToCore(err))
+	require.NoError(t, err, "retrieving load test m365 org users from test configuration %+v", clues.ToCore(err))
 
 	users := cfg[TestCfgLoadTestOrgUsers]
 	users = strings.TrimPrefix(users, "[")
@@ -169,7 +169,7 @@ func LoadTestM365OrgUsers(t *testing.T) []string {
 // last-attempt fallback that will only work on alcion's testing org.
 func M365SiteID(t *testing.T) string {
 	cfg, err := readTestConfig()
-	require.NoError(t, err, "retrieving m365 site id from test configuration", clues.ToCore(err))
+	require.NoError(t, err, "retrieving m365 site id from test configuration: %+v", clues.ToCore(err))
 
 	return strings.ToLower(cfg[TestCfgSiteID])
 }
@@ -180,7 +180,7 @@ func M365SiteID(t *testing.T) string {
 // last-attempt fallback that will only work on alcion's testing org.
 func M365SiteURL(t *testing.T) string {
 	cfg, err := readTestConfig()
-	require.NoError(t, err, "retrieving m365 site url from test configuration", clues.ToCore(err))
+	require.NoError(t, err, "retrieving m365 site url from test configuration: %+v", clues.ToCore(err))
 
 	return strings.ToLower(cfg[TestCfgSiteURL])
 }
@@ -197,3 +197,15 @@ func GetM365SiteID(ctx context.Context) string {
 
 	return strings.ToLower(cfg[TestCfgSiteID])
 }
+
+// UnlicensedM365UserID returns an userID string representing the m365UserID
+// described by either the env var CORSO_M365_TEST_UNLICENSED_USER, the
+// corso_test.toml config file or the default value (in that order of priority).
+// The default is a last-attempt fallback that will only work on alcion's
+// testing org.
+func UnlicensedM365UserID(t *testing.T) string {
+	cfg, err := readTestConfig()
+	require.NoError(t, err, "retrieving unlicensed m365 user id from test configuration: %+v", clues.ToCore(err))
+
+	return strings.ToLower(cfg[TestCfgSecondaryUserID])
+}