package e2e import ( "fmt" "os" "path/filepath" "testing" "time" "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/model" user_migration "github.com/QuantumNous/new-api/service/user_migration" helpers "github.com/QuantumNous/new-api/test/e2e/helpers" "github.com/stretchr/testify/require" ) var ( repoRoot string artifactRoot string binaryPath string ) type simpleResponse struct { Success bool `json:"success"` Message string `json:"message,omitempty"` Error string `json:"error,omitempty"` } type createBatchResponse struct { Success bool `json:"success"` Data model.UserMigrationBatch `json:"data"` } type adminUserResponse struct { Success bool `json:"success"` Data model.User `json:"data"` } type batchPageResponse struct { Success bool `json:"success"` Data model.UserMigrationBatch `json:"data"` Items []model.UserMigrationItem `json:"items"` Total int64 `json:"total"` } type preparedBatch struct { BatchID int Detail *helpers.BatchDetail Items map[int]*model.UserMigrationItem } type executedBatch struct { Prepared *preparedBatch Detail *helpers.BatchDetail CreatedUser *model.User SoftAffUser *model.User VerifyResult map[int]helpers.VerifyEntry } func TestMain(m *testing.M) { var err error wd, err := os.Getwd() if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } repoRoot = filepath.Clean(filepath.Join(wd, "..", "..")) artifactRoot = filepath.Join(repoRoot, "test-artifacts", "user-migration-e2e") if err := os.MkdirAll(artifactRoot, 0o755); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } binaryPath, err = helpers.BuildBinary(repoRoot, artifactRoot) if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } os.Exit(m.Run()) } func TestUserMigrationE2E_FullFlow_GrayScenarios(t *testing.T) { env := newEnvironment(t) result := runFullFlow(t, env) require.Equal(t, model.UserMigrationBatchStatusCompleted, result.Detail.Batch.Status) require.Equal(t, 3, result.Detail.Batch.MigratedUsers) require.Equal(t, 0, result.Detail.Batch.FailedUsers) require.Len(t, result.VerifyResult, 3) } func TestUserMigrationE2E_IdempotentRetry(t *testing.T) { env := newEnvironment(t) result := runFullFlow(t, env) forceRetryableState(t, env.CNDBPath, result.Prepared.BatchID, []int{ result.Prepared.Items[helpers.OVCreateUserID].Id, result.Prepared.Items[helpers.OVMergeUserID].Id, result.Prepared.Items[helpers.OVSoftAffUserID].Id, }) var retryResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/retry", result.Prepared.BatchID), nil, 202, &retryResp, )) require.True(t, retryResp.Success) retried, err := helpers.WaitBatchStatus(env.CNClient, result.Prepared.BatchID, model.UserMigrationBatchStatusCompleted, 25*time.Second) require.NoError(t, err) retriedItems := indexItemsBySourceUserID(retried.Items) require.Equal(t, model.UserMigrationItemStatusMigrated, retriedItems[helpers.OVCreateUserID].Status) require.Equal(t, model.UserMigrationItemStatusMigrated, retriedItems[helpers.OVMergeUserID].Status) require.Equal(t, model.UserMigrationItemStatusMigrated, retriedItems[helpers.OVSoftAffUserID].Status) require.EqualValues(t, 1, helpers.CountUsersByUsername(t, env.CNDBPath, helpers.UsernameOVCreate)) require.EqualValues(t, 1, helpers.CountUsersByUsername(t, env.CNDBPath, helpers.UsernameOVSoftAff)) helpers.AssertQuota(t, env.CNDBPath, helpers.CNMergeTargetUserID, 700) helpers.AssertNoDuplicateQuotaGrant(t, env.CNDBPath, result.Prepared.Items[helpers.OVMergeUserID].MigrationToken) verifyEntries, err := helpers.VerifyBatch(env.CNClient, result.Prepared.BatchID) require.NoError(t, err) require.Len(t, verifyEntries, 3) for _, entry := range verifyEntries { require.Empty(t, entry.Error) require.NotNil(t, entry.Result) require.True(t, entry.Result.TargetUserExists) require.True(t, entry.Result.RemoteConverted) require.True(t, entry.Result.QuotaMatched) } } func TestUserMigrationE2E_ExecutionTimeDriftFails(t *testing.T) { env := newEnvironment(t) prepared := prepareGrayScenarioBatch(t, env) renameOVUser(t, env, helpers.OVCreateUserID, helpers.UsernameCNMergeTarget) var executeResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/execute", prepared.BatchID), nil, 202, &executeResp, )) require.True(t, executeResp.Success) failedDetail, err := helpers.WaitBatchStatus(env.CNClient, prepared.BatchID, model.UserMigrationBatchStatusFailed, 25*time.Second) require.NoError(t, err) items := indexItemsBySourceUserID(failedDetail.Items) require.Equal(t, model.UserMigrationItemStatusFailed, items[helpers.OVCreateUserID].Status) require.Contains(t, items[helpers.OVCreateUserID].ErrorMessage, "drift") require.Equal(t, model.UserMigrationItemStatusMigrated, items[helpers.OVMergeUserID].Status) require.Equal(t, model.UserMigrationItemStatusMigrated, items[helpers.OVSoftAffUserID].Status) require.Equal(t, model.UserMigrationItemStatusSkipped, items[helpers.OVHardSyncedUserID].Status) require.Equal(t, model.UserMigrationItemStatusSkipped, items[helpers.OVUsernameConflictUserID].Status) require.EqualValues(t, 0, helpers.CountUsersByEmail(t, env.CNDBPath, helpers.EmailOVCreate)) driftedRemote, err := helpers.GetRemoteUserSnapshot(env.Cluster.OV.BaseURL, helpers.OVCreateUserID) require.NoError(t, err) require.NotEqual(t, common.UserSourceSynced, driftedRemote.Source) } func TestUserMigrationE2E_RescanPreservesHumanDecision(t *testing.T) { env := newEnvironment(t) prepared := prepareGrayScenarioBatch(t, env) var scanResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", prepared.BatchID), nil, 202, &scanResp, )) require.True(t, scanResp.Success) rescannedDetail, err := helpers.WaitBatchStatus(env.CNClient, prepared.BatchID, model.UserMigrationBatchStatusScanned, 20*time.Second) require.NoError(t, err) items := indexItemsBySourceUserID(rescannedDetail.Items) require.Equal(t, model.UserMigrationStrategySkip, items[helpers.OVHardSyncedUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusConflict, items[helpers.OVHardSyncedUserID].Status) require.Equal(t, model.UserMigrationStrategySkip, items[helpers.OVUsernameConflictUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusConflict, items[helpers.OVUsernameConflictUserID].Status) require.Equal(t, model.UserMigrationStrategyMergeExisting, items[helpers.OVMergeUserID].ResolutionStrategy) require.Equal(t, helpers.CNMergeTargetUserID, items[helpers.OVMergeUserID].TargetUserId) } func TestUserMigrationE2E_ScanPagination(t *testing.T) { env := newEnvironment(t) createManagedUsers(t, env.OVClient, "ov-page", 97, common.RoleCommonUser) batchID := createBatch(t, env, "pagination") var scanResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", batchID), nil, 202, &scanResp, )) require.True(t, scanResp.Success) detail, err := helpers.WaitBatchStatus(env.CNClient, batchID, model.UserMigrationBatchStatusScanned, 60*time.Second) require.NoError(t, err) require.EqualValues(t, 102, detail.Total) require.Equal(t, 102, detail.Batch.TotalUsers) page0 := getBatchPage(t, env.CNClient, batchID, 0, 100) require.Len(t, page0.Items, 100) page1 := getBatchPage(t, env.CNClient, batchID, 1, 100) require.Len(t, page1.Items, 2) require.EqualValues(t, 102, page1.Total) } func TestUserMigrationE2E_RootPermission(t *testing.T) { env := newEnvironment(t) adminClient := createAndLoginManagedUser(t, env.Cluster.CN.BaseURL, env.CNClient, "migration-admin", helpers.DefaultRootPassword, common.RoleAdminUser) var listResp simpleResponse require.NoError(t, adminClient.GetJSON("/api/user-migrations/batches", 200, &listResp)) require.False(t, listResp.Success) require.NotEmpty(t, listResp.Message) var createResp simpleResponse require.NoError(t, adminClient.PostJSON("/api/user-migrations/batches", map[string]any{ "name": "no-access", "source_region": "ov", "target_region": "cn", }, 200, &createResp)) require.False(t, createResp.Success) require.NotEmpty(t, createResp.Message) } func TestUserMigrationE2E_ResolveInvalidPayload(t *testing.T) { env := newEnvironment(t) scanned := scanGrayScenarioBatch(t, env, "resolve-invalid") conflictItem := scanned.Items[helpers.OVHardSyncedUserID] var invalidStrategyResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/items/%d/resolve", conflictItem.Id), map[string]any{ "resolution_strategy": "invalid", }, 400, &invalidStrategyResp, )) require.False(t, invalidStrategyResp.Success) require.Contains(t, invalidStrategyResp.Error, "invalid resolution_strategy") var mergeWithoutTargetResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/items/%d/resolve", conflictItem.Id), map[string]any{ "resolution_strategy": model.UserMigrationStrategyMergeExisting, }, 400, &mergeWithoutTargetResp, )) require.False(t, mergeWithoutTargetResp.Success) require.Contains(t, mergeWithoutTargetResp.Error, "target_user_id") } func TestUserMigrationE2E_ScanConflict_WhenAlreadyScanning(t *testing.T) { env := newEnvironment(t) batchID := createBatch(t, env, "scan-conflict") setBatchStatus(t, env.CNDBPath, batchID, model.UserMigrationBatchStatusScanning) var resp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", batchID), nil, 409, &resp, )) require.False(t, resp.Success) require.Contains(t, resp.Error, "already being scanned") } func TestUserMigrationE2E_ExecuteConflict_WhenAlreadyRunning(t *testing.T) { env := newEnvironment(t) prepared := prepareGrayScenarioBatch(t, env) setBatchStatus(t, env.CNDBPath, prepared.BatchID, model.UserMigrationBatchStatusRunning) var resp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/execute", prepared.BatchID), nil, 409, &resp, )) require.False(t, resp.Success) require.Contains(t, resp.Error, "already running") } func TestUserMigrationE2E_VerifyQuotaMismatchDetected(t *testing.T) { env := newEnvironment(t) result := runFullFlow(t, env) convertRemoteUserToSynced(t, env.Cluster.OV.BaseURL, helpers.OVMergeUserID, helpers.CNMergeTargetUserID, 199) verifyEntries, err := helpers.VerifyBatch(env.CNClient, result.Prepared.BatchID) require.NoError(t, err) verifyBySource := make(map[int]helpers.VerifyEntry, len(verifyEntries)) for _, entry := range verifyEntries { verifyBySource[entry.SourceUserID] = entry } require.NotNil(t, verifyBySource[helpers.OVMergeUserID].Result) require.False(t, verifyBySource[helpers.OVMergeUserID].Result.QuotaMatched) require.True(t, verifyBySource[helpers.OVCreateUserID].Result.QuotaMatched) require.True(t, verifyBySource[helpers.OVSoftAffUserID].Result.QuotaMatched) } func TestUserMigrationE2E_SkipNeverConvertsRemote(t *testing.T) { env := newEnvironment(t) prepared := prepareGrayScenarioBatch(t, env) resolveItem(t, env, prepared.Items[helpers.OVCreateUserID].Id, model.UserMigrationStrategySkip, 0) readyDetail, err := helpers.WaitBatchStatus(env.CNClient, prepared.BatchID, model.UserMigrationBatchStatusReady, 10*time.Second) require.NoError(t, err) var executeResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/execute", prepared.BatchID), nil, 202, &executeResp, )) require.True(t, executeResp.Success) completedDetail, err := helpers.WaitBatchStatus(env.CNClient, prepared.BatchID, model.UserMigrationBatchStatusCompleted, 25*time.Second) require.NoError(t, err) items := indexItemsBySourceUserID(completedDetail.Items) require.Equal(t, model.UserMigrationItemStatusSkipped, items[helpers.OVCreateUserID].Status) require.EqualValues(t, 0, helpers.CountUsersByEmail(t, env.CNDBPath, helpers.EmailOVCreate)) remoteUser, err := helpers.GetRemoteUserSnapshot(env.Cluster.OV.BaseURL, helpers.OVCreateUserID) require.NoError(t, err) require.NotEqual(t, common.UserSourceSynced, remoteUser.Source) require.Equal(t, readyDetail.Items[0].BatchId, items[helpers.OVCreateUserID].BatchId) } func TestUserMigrationE2E_ExplicitIDs_OnlySelectedUsersScanned(t *testing.T) { env := newEnvironment(t) batchID := createBatchWithSelection(t, env, "explicit-selected", []int{ helpers.OVCreateUserID, helpers.OVMergeUserID, }) var scanResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", batchID), nil, 202, &scanResp, )) require.True(t, scanResp.Success) detail, err := helpers.WaitBatchStatus(env.CNClient, batchID, model.UserMigrationBatchStatusReady, 20*time.Second) require.NoError(t, err) require.EqualValues(t, 2, detail.Total) require.Equal(t, 2, detail.Batch.TotalUsers) require.Equal(t, user_migration.SelectionModeExplicitIDs, detail.Batch.SelectionMode) require.Equal(t, 2, detail.Batch.RequestedUserCount) items := indexItemsBySourceUserID(detail.Items) require.Len(t, items, 2) require.Contains(t, items, helpers.OVCreateUserID) require.Contains(t, items, helpers.OVMergeUserID) require.NotContains(t, items, helpers.OVHardSyncedUserID) } func TestUserMigrationE2E_ExplicitIDs_SelectionSummaryRecorded(t *testing.T) { env := newEnvironment(t) batchID := createBatchWithSelection(t, env, "explicit-summary", []int{ helpers.OVCreateUserID, helpers.OVSyncedCopyUserID, 1, 19999999, }) var scanResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", batchID), nil, 202, &scanResp, )) require.True(t, scanResp.Success) detail, err := helpers.WaitBatchStatus(env.CNClient, batchID, model.UserMigrationBatchStatusReady, 20*time.Second) require.NoError(t, err) require.Equal(t, 4, detail.Batch.RequestedUserCount) require.Equal(t, 1, detail.Batch.MatchedUserCount) require.Equal(t, 3, detail.Batch.ExcludedUserCount) require.EqualValues(t, 1, detail.Total) var summary user_migration.SelectionSummary require.NoError(t, common.UnmarshalJsonStr(detail.Batch.SelectionSummary, &summary)) require.Equal(t, 4, summary.Requested) require.Equal(t, 1, summary.Matched) require.Len(t, summary.Excluded, 3) reasonsByID := make(map[int]string, len(summary.Excluded)) for _, item := range summary.Excluded { reasonsByID[item.UserID] = item.Reason } require.Equal(t, "already_synced", reasonsByID[helpers.OVSyncedCopyUserID]) require.Equal(t, "root_user_not_migratable", reasonsByID[1]) require.Equal(t, "not_found", reasonsByID[19999999]) } func TestUserMigrationE2E_ExplicitIDs_ActiveBatchOverlapRejected(t *testing.T) { env := newEnvironment(t) firstBatchID := createBatchWithSelection(t, env, "explicit-overlap-first", []int{ helpers.OVCreateUserID, }) require.NotZero(t, firstBatchID) var resp struct { Success bool `json:"success"` Error string `json:"error"` Conflicts []struct { SourceUserID int `json:"source_user_id"` BatchID int `json:"batch_id"` BatchStatus string `json:"batch_status"` } `json:"conflicts"` } require.NoError(t, env.CNClient.PostJSON("/api/user-migrations/batches", map[string]any{ "name": "explicit-overlap-second", "source_region": "ov", "target_region": "cn", "selection_mode": "explicit_ids", "source_user_ids": []int{helpers.OVCreateUserID, helpers.OVMergeUserID}, }, 409, &resp)) require.False(t, resp.Success) require.Equal(t, "some selected users already exist in active migration batches", resp.Error) require.Len(t, resp.Conflicts, 1) require.Equal(t, helpers.OVCreateUserID, resp.Conflicts[0].SourceUserID) require.Equal(t, firstBatchID, resp.Conflicts[0].BatchID) require.Equal(t, model.UserMigrationBatchStatusDraft, resp.Conflicts[0].BatchStatus) } func newEnvironment(t *testing.T) *helpers.Environment { t.Helper() env, err := helpers.PrepareEnvironment(repoRoot, artifactRoot, binaryPath, t.Name()) require.NoError(t, err) t.Cleanup(func() { require.NoError(t, env.Close()) }) return env } func runFullFlow(t *testing.T, env *helpers.Environment) *executedBatch { t.Helper() prepared := prepareGrayScenarioBatch(t, env) var executeResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/execute", prepared.BatchID), nil, 202, &executeResp, )) require.True(t, executeResp.Success) completedDetail, err := helpers.WaitBatchStatus(env.CNClient, prepared.BatchID, model.UserMigrationBatchStatusCompleted, 25*time.Second) require.NoError(t, err) require.EqualValues(t, 5, completedDetail.Total) items := indexItemsBySourceUserID(completedDetail.Items) require.Equal(t, model.UserMigrationItemStatusMigrated, items[helpers.OVCreateUserID].Status) require.Equal(t, model.UserMigrationItemStatusMigrated, items[helpers.OVMergeUserID].Status) require.Equal(t, model.UserMigrationItemStatusMigrated, items[helpers.OVSoftAffUserID].Status) require.Equal(t, model.UserMigrationItemStatusSkipped, items[helpers.OVHardSyncedUserID].Status) require.Equal(t, model.UserMigrationItemStatusSkipped, items[helpers.OVUsernameConflictUserID].Status) createdUser := helpers.AssertCNUserExists(t, env.CNDBPath, helpers.UsernameOVCreate) require.Equal(t, 100, createdUser.Quota) helpers.AssertQuota(t, env.CNDBPath, helpers.CNMergeTargetUserID, 700) softAffUser := helpers.AssertCNUserExists(t, env.CNDBPath, helpers.UsernameOVSoftAff) require.Equal(t, 150, softAffUser.Quota) require.NotEqual(t, helpers.AffCodeCNSoftAffOwner, softAffUser.AffCode) helpers.AssertOVUserSynced(t, env.Cluster.OV.BaseURL, helpers.OVCreateUserID, createdUser.Id, 100) helpers.AssertOVUserSynced(t, env.Cluster.OV.BaseURL, helpers.OVMergeUserID, helpers.CNMergeTargetUserID, 700) helpers.AssertOVUserSynced(t, env.Cluster.OV.BaseURL, helpers.OVSoftAffUserID, softAffUser.Id, 150) hardSnapshot, err := helpers.GetRemoteUserSnapshot(env.Cluster.OV.BaseURL, helpers.OVHardSyncedUserID) require.NoError(t, err) require.NotEqual(t, common.UserSourceSynced, hardSnapshot.Source) usernameSnapshot, err := helpers.GetRemoteUserSnapshot(env.Cluster.OV.BaseURL, helpers.OVUsernameConflictUserID) require.NoError(t, err) require.NotEqual(t, common.UserSourceSynced, usernameSnapshot.Source) verifyEntries, err := helpers.VerifyBatch(env.CNClient, prepared.BatchID) require.NoError(t, err) require.Len(t, verifyEntries, 3) verifyBySource := make(map[int]helpers.VerifyEntry, len(verifyEntries)) for _, entry := range verifyEntries { require.Empty(t, entry.Error) require.NotNil(t, entry.Result) require.True(t, entry.Result.TargetUserExists) require.True(t, entry.Result.RemoteConverted) require.True(t, entry.Result.QuotaMatched) verifyBySource[entry.SourceUserID] = entry } return &executedBatch{ Prepared: prepared, Detail: completedDetail, CreatedUser: createdUser, SoftAffUser: softAffUser, VerifyResult: verifyBySource, } } func prepareGrayScenarioBatch(t *testing.T, env *helpers.Environment) *preparedBatch { t.Helper() scanned := scanGrayScenarioBatch(t, env, "gray-scenarios") scannedItems := scanned.Items require.Equal(t, model.UserMigrationStrategyCreateNew, scannedItems[helpers.OVCreateUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusReady, scannedItems[helpers.OVCreateUserID].Status) require.Equal(t, model.UserMigrationStrategyMergeExisting, scannedItems[helpers.OVMergeUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusReady, scannedItems[helpers.OVMergeUserID].Status) require.Equal(t, helpers.CNMergeTargetUserID, scannedItems[helpers.OVMergeUserID].TargetUserId) require.Equal(t, model.UserMigrationItemStatusConflict, scannedItems[helpers.OVHardSyncedUserID].Status) require.Contains(t, decodeFlags(t, scannedItems[helpers.OVHardSyncedUserID].ConflictFlags), "cn_already_synced_to_ov") require.Equal(t, model.UserMigrationItemStatusConflict, scannedItems[helpers.OVUsernameConflictUserID].Status) require.Contains(t, decodeFlags(t, scannedItems[helpers.OVUsernameConflictUserID].ConflictFlags), "username") require.Equal(t, model.UserMigrationStrategyCreateNew, scannedItems[helpers.OVSoftAffUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusReady, scannedItems[helpers.OVSoftAffUserID].Status) require.Contains(t, decodeFlags(t, scannedItems[helpers.OVSoftAffUserID].ConflictFlags), "aff_code") resolveItem(t, env, scannedItems[helpers.OVHardSyncedUserID].Id, model.UserMigrationStrategySkip, 0) resolveItem(t, env, scannedItems[helpers.OVUsernameConflictUserID].Id, model.UserMigrationStrategySkip, 0) readyDetail, err := helpers.WaitBatchStatus(env.CNClient, scanned.BatchID, model.UserMigrationBatchStatusReady, 10*time.Second) require.NoError(t, err) readyItems := indexItemsBySourceUserID(readyDetail.Items) require.Equal(t, model.UserMigrationStrategySkip, readyItems[helpers.OVHardSyncedUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusReady, readyItems[helpers.OVHardSyncedUserID].Status) require.Equal(t, model.UserMigrationStrategySkip, readyItems[helpers.OVUsernameConflictUserID].ResolutionStrategy) require.Equal(t, model.UserMigrationItemStatusReady, readyItems[helpers.OVUsernameConflictUserID].Status) return &preparedBatch{ BatchID: scanned.BatchID, Detail: readyDetail, Items: readyItems, } } func scanGrayScenarioBatch(t *testing.T, env *helpers.Environment, name string) *preparedBatch { t.Helper() batchID := createBatch(t, env, name) var scanResp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/batches/%d/scan", batchID), nil, 202, &scanResp, )) require.True(t, scanResp.Success) scannedDetail, err := helpers.WaitBatchStatus(env.CNClient, batchID, model.UserMigrationBatchStatusScanned, 20*time.Second) require.NoError(t, err) require.EqualValues(t, 5, scannedDetail.Total) scannedItems := indexItemsBySourceUserID(scannedDetail.Items) require.Len(t, scannedItems, 5) return &preparedBatch{ BatchID: batchID, Detail: scannedDetail, Items: scannedItems, } } func createBatch(t *testing.T, env *helpers.Environment, name string) int { t.Helper() return createBatchWithSelection(t, env, name, nil) } func createBatchWithSelection(t *testing.T, env *helpers.Environment, name string, sourceUserIDs []int) int { t.Helper() payload := map[string]any{ "name": name, "source_region": "ov", "target_region": "cn", } if len(sourceUserIDs) > 0 { payload["selection_mode"] = "explicit_ids" payload["source_user_ids"] = sourceUserIDs } var resp createBatchResponse require.NoError(t, env.CNClient.PostJSON("/api/user-migrations/batches", payload, 200, &resp)) require.True(t, resp.Success) require.NotZero(t, resp.Data.Id) return resp.Data.Id } func resolveItem(t *testing.T, env *helpers.Environment, itemID int, strategy string, targetUserID int) { t.Helper() var resp simpleResponse require.NoError(t, env.CNClient.PostJSON( fmt.Sprintf("/api/user-migrations/items/%d/resolve", itemID), map[string]any{ "resolution_strategy": strategy, "target_user_id": targetUserID, }, 200, &resp, )) require.True(t, resp.Success) } func renameOVUser(t *testing.T, env *helpers.Environment, userID int, newUsername string) { t.Helper() var getResp adminUserResponse require.NoError(t, env.OVClient.GetJSON(fmt.Sprintf("/api/user/%d", userID), 200, &getResp)) require.True(t, getResp.Success) var updateResp simpleResponse require.NoError(t, env.OVClient.PutJSON("/api/user", map[string]any{ "id": getResp.Data.Id, "username": newUsername, "display_name": getResp.Data.DisplayName, "group": getResp.Data.Group, "quota": getResp.Data.Quota, "remark": getResp.Data.Remark, "capture_relay": getResp.Data.CaptureRelay, }, 200, &updateResp)) require.True(t, updateResp.Success) } func createAndLoginManagedUser(t *testing.T, baseURL string, rootClient *helpers.APIClient, username, password string, role int) *helpers.APIClient { t.Helper() var resp simpleResponse require.NoError(t, rootClient.PostJSON("/api/user/", map[string]any{ "username": username, "password": password, "display_name": username, "role": role, }, 200, &resp)) require.True(t, resp.Success) client, err := helpers.NewAPIClient(baseURL) require.NoError(t, err) require.NoError(t, helpers.Login(client, username, password)) return client } func createManagedUsers(t *testing.T, client *helpers.APIClient, prefix string, count int, role int) { t.Helper() for i := 0; i < count; i++ { username := prefix email := fmt.Sprintf("%s@example.com", prefix) displayName := prefix if count > 1 { username = fmt.Sprintf("%s-%03d", prefix, i) email = fmt.Sprintf("%s-%03d@example.com", prefix, i) displayName = fmt.Sprintf("%s %03d", prefix, i) } var resp simpleResponse require.NoError(t, client.PostJSON("/api/user/", map[string]any{ "username": username, "password": helpers.DefaultRootPassword, "display_name": displayName, "email": email, "role": role, }, 200, &resp)) require.True(t, resp.Success) } } func forceRetryableState(t *testing.T, dbPath string, batchID int, itemIDs []int) { t.Helper() db, err := helpers.OpenSQLite(dbPath) require.NoError(t, err) defer func() { require.NoError(t, helpers.CloseSQLite(db)) }() require.NoError(t, db.Model(&model.UserMigrationBatch{}).Where("id = ?", batchID). Update("status", model.UserMigrationBatchStatusFailed).Error) require.NoError(t, db.Model(&model.UserMigrationItem{}).Where("id IN ?", itemIDs). Updates(map[string]any{ "status": model.UserMigrationItemStatusFailed, "error_message": "forced retry for e2e", }).Error) } func setBatchStatus(t *testing.T, dbPath string, batchID int, status string) { t.Helper() db, err := helpers.OpenSQLite(dbPath) require.NoError(t, err) defer func() { require.NoError(t, helpers.CloseSQLite(db)) }() require.NoError(t, db.Model(&model.UserMigrationBatch{}).Where("id = ?", batchID).Update("status", status).Error) } func getBatchPage(t *testing.T, client *helpers.APIClient, batchID, page, pageSize int) *batchPageResponse { t.Helper() var resp batchPageResponse require.NoError(t, client.GetJSON( fmt.Sprintf("/api/user-migrations/batches/%d?page=%d&page_size=%d", batchID, page, pageSize), 200, &resp, )) require.True(t, resp.Success) return &resp } func convertRemoteUserToSynced(t *testing.T, ovBaseURL string, sourceUserID, remoteUserID, syncedQuota int) { t.Helper() client, err := helpers.NewAPIClient(ovBaseURL) require.NoError(t, err) var resp simpleResponse require.NoError(t, client.DoJSONWithHeaders( "POST", fmt.Sprintf("/api/internal/migration/users/%d/convert-to-synced", sourceUserID), map[string]any{ "remote_user_id": remoteUserID, "synced_quota": syncedQuota, }, 200, map[string]string{ "X-Sync-API-Key": helpers.DefaultSyncAPIKey, "X-Sync-Node": "e2e-test", }, &resp, )) require.True(t, resp.Success) } func indexItemsBySourceUserID(items []model.UserMigrationItem) map[int]*model.UserMigrationItem { indexed := make(map[int]*model.UserMigrationItem, len(items)) for i := range items { indexed[items[i].SourceUserId] = &items[i] } return indexed } func decodeFlags(t *testing.T, flagsJSON string) []string { t.Helper() var flags []string require.NoError(t, common.UnmarshalJsonStr(flagsJSON, &flags)) return flags }